测试基建：规范化粘连的 API 配置行

test_pipeline/config_preprocess/config_lines.py

@@ -0,0 +1,91 @@
+"""Paddle API 配置行的共享解析工具。"""
+
+# 每个逻辑调用必须以 paddle. 开头并具有完整外层括号。
+# 同一物理行允许连续记录多个调用，调用之间可以没有空白分隔符。
+# 字符串和嵌套参数中的 paddle. 只属于当前调用，不能成为切分边界。
+# 不能无损归属的残余文本必须报错，由调用方决定拒绝或终止。
+
+from __future__ import annotations
+
+import string
+
+
+# 采集配置中的 API 名只接受 ASCII 标识符，避免 Unicode 或运算符混入边界。
+API_NAME_CHARS = frozenset(string.ascii_letters + string.digits + "_.")
+
+
+def matching_close(text, start, opening, closing):
+    """返回与 start 处括号匹配的结束位置。"""
+    depth = 0
+    quote = None
+    escaped = False
+    for index in range(start, len(text)):
+        char = text[index]
+        if quote is not None:
+            if escaped:
+                escaped = False
+            elif char == "\\":
+                escaped = True
+            elif char == quote:
+                quote = None
+            continue
+        if char in {'"', "'"}:
+            quote = char
+        elif char == opening:
+            depth += 1
+        elif char == closing:
+            depth -= 1
+            if depth == 0:
+                return index
+    return None
+
+
+def _api_name_end(text, start):
+    """返回 API 名结束位置，并阻止粘连的第二个 paddle 前缀混入名称。"""
+    index = start
+    while index < len(text) and text[index] in API_NAME_CHARS:
+        # 名称内部再次出现 paddle. 表示前一个调用缺少左括号，而非更长 API 名。
+        if index > start and text.startswith("paddle.", index):
+            break
+        index += 1
+    return index
+
+
+def _is_valid_api_name(api_name):
+    """验证 paddle 后的每级名称均为 Python 标识符。"""
+    parts = api_name.split(".")
+    return parts[0] == "paddle" and len(parts) > 1 and all(
+        part
+        and (part[0] in string.ascii_letters or part[0] == "_")
+        and all(char in string.ascii_letters + string.digits + "_" for char in part[1:])
+        for part in parts[1:]
+    )
+
+
+def split_top_level_calls(text):
+    """将一行中连续的顶层 paddle.* 调用无损拆分。"""
+    calls = []
+    cursor = 0
+    while cursor < len(text):
+        while cursor < len(text) and text[cursor].isspace():
+            cursor += 1
+        if cursor >= len(text):
+            break
+        if not text.startswith("paddle.", cursor):
+            # 非 paddle 残余文本不能静默丢弃，否则会造成测试覆盖缺失。
+            raise ValueError(
+                "顶层调用结束后存在无法识别的内容，期望下一个 paddle.* 调用"
+            )
+
+        name_end = _api_name_end(text, cursor)
+        if not _is_valid_api_name(text[cursor:name_end]):
+            raise ValueError("顶层 paddle.* API 名不合法")
+        if name_end >= len(text) or text[name_end] != "(":
+            raise ValueError("顶层 paddle.* 调用缺少左括号")
+        closing = matching_close(text, name_end, "(", ")")
+        if closing is None:
+            raise ValueError("顶层 paddle.* 调用括号不匹配")
+        # 调用文本原样保留，规范化阶段不重排参数或改变空白。
+        calls.append(text[cursor : closing + 1])
+        cursor = closing + 1
+    return calls

test_pipeline/config_preprocess/normalize_config_lines.py

@@ -0,0 +1,91 @@
+#!/usr/bin/env python3
+"""将原始配置规范化为每行一个完整的 Paddle API 调用。"""
+
+# 规范化只修复物理换行丢失，不改变调用内容、顺序或重复次数。
+# 成功拆出的调用立即写入输出，避免 GB 级输入常驻内存。
+# 失败的原始行连同行号写入拒绝文件，供修复后重新处理。
+# 严格模式在完整扫描结束后失败，已完成工作和错误证据都会保留。
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+from test_pipeline.config_preprocess.config_lines import split_top_level_calls
+
+
+def normalize_file(input_path, output_path, reject_path):
+    """流式拆分文件并记录拒绝行。"""
+    # 返回计数只描述结构规范化，不做 APIConfig 语义解析或文本去重。
+    input_path = Path(input_path)
+    output_path = Path(output_path)
+    reject_path = Path(reject_path)
+    if input_path.resolve() == output_path.resolve():
+        raise ValueError("输入和输出文件不能相同")
+
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    reject_path.parent.mkdir(parents=True, exist_ok=True)
+    counts = {"lines": 0, "calls": 0, "split_lines": 0, "rejected": 0}
+    with (
+        input_path.open(encoding="utf-8") as source_file,
+        output_path.open("w", encoding="utf-8") as normalized_file,
+        reject_path.open("w", encoding="utf-8") as reject_file,
+    ):
+        for line_number, raw_line in enumerate(source_file, start=1):
+            counts["lines"] += 1
+            config = raw_line.strip()
+            if not config:
+                continue
+            try:
+                calls = split_top_level_calls(config)
+            except ValueError as error:
+                # 错误行完整落盘，后续可以修复后单独重放，不会丢失原始数据。
+                counts["rejected"] += 1
+                reject_file.write(f"{input_path}:{line_number}\t{error}\t{config}\n")
+                continue
+            if len(calls) > 1:
+                counts["split_lines"] += 1
+            for call in calls:
+                normalized_file.write(call + "\n")
+                counts["calls"] += 1
+
+    if counts["rejected"] == 0:
+        reject_path.unlink()
+    return counts
+
+
+def parse_args(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("-i", "--input", required=True, help="原始配置文件")
+    parser.add_argument("-o", "--output", required=True, help="规范化配置文件")
+    parser.add_argument(
+        "--rejects",
+        default=None,
+        help="拒绝文件路径，默认是 <output>.unparsed.txt",
+    )
+    parser.add_argument(
+        "--strict",
+        action="store_true",
+        help="完成扫描并写出拒绝文件后，对任何坏行返回非零状态",
+    )
+    return parser.parse_args(argv)
+
+
+def main(argv=None):
+    args = parse_args(argv)
+    reject_path = args.rejects or f"{args.output}.unparsed.txt"
+    counts = normalize_file(args.input, args.output, reject_path)
+    print(
+        f"规范化: {args.input} -> {args.output}，"
+        f"输入 {counts['lines']} 行，输出 {counts['calls']} 个调用，"
+        f"拆分 {counts['split_lines']} 行，拒绝 {counts['rejected']} 行"
+    )
+    if counts["rejected"]:
+        print(f"警告: 无法拆分的配置已写入 {reject_path}")
+        if args.strict:
+            return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

test_pipeline/config_preprocess/test_to_0_size_config.py

@@ -0,0 +1,104 @@
+"""验证配置行的顶层调用拆分协议。"""
+
+# 测试同时覆盖共享结构规范化和 0-size 入口的 APIConfig 语义解析。
+# 重点保证脏数据可定位、可落盘，并且不会阻止后续合法调用被处理。
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from test_pipeline.config_preprocess.config_lines import split_top_level_calls
+from test_pipeline.config_preprocess.normalize_config_lines import normalize_file
+from test_pipeline.config_preprocess.to_0_size_config import iter_api_configs
+
+
+class SplitTopLevelCallsTest(unittest.TestCase):
+    """确保拆分不会改变独立调用的内容和顺序。"""
+
+    def test_split_two_calls_without_separator(self):
+        text = "paddle.zeros([1])paddle.randn([2])"
+        self.assertEqual(
+            split_top_level_calls(text),
+            ["paddle.zeros([1])", "paddle.randn([2])"],
+        )
+
+    def test_nested_parentheses_and_string_are_not_split(self):
+        # 字符串中的伪调用和右括号不能提前结束外层调用。
+        text = 'paddle.foo((1, 2), "paddle.bar())")paddle.zeros([1])'
+        self.assertEqual(
+            split_top_level_calls(text),
+            ['paddle.foo((1, 2), "paddle.bar())")', "paddle.zeros([1])"],
+        )
+
+    def test_unbalanced_or_trailing_text_is_rejected(self):
+        # 无法无损拆分的数据必须失败，不能只保留可识别的前缀。
+        with self.assertRaisesRegex(ValueError, "括号不匹配"):
+            split_top_level_calls("paddle.zeros([1]")
+        with self.assertRaisesRegex(ValueError, "无法识别"):
+            split_top_level_calls("paddle.zeros([1]) + 1")
+
+    def test_api_name_scan_does_not_cross_into_next_prefix(self):
+        with self.assertRaisesRegex(ValueError, "缺少左括号"):
+            split_top_level_calls("paddle.foopaddle.bar(1)")
+
+    def test_iter_api_configs_parses_each_call(self):
+        with tempfile.NamedTemporaryFile("w", encoding="utf-8") as config_file:
+            config_file.write("paddle.zeros([1])paddle.randn([2])\n")
+            config_file.flush()
+            configs = list(iter_api_configs(config_file.name))
+        self.assertEqual(
+            [config.api_name for config in configs],
+            ["paddle.zeros", "paddle.randn"],
+        )
+
+    def test_api_config_error_contains_source_line(self):
+        with tempfile.NamedTemporaryFile("w", encoding="utf-8") as config_file:
+            config_file.write("paddle.add(Tensor([1]), Tensor([1]))\n")
+            config_file.flush()
+            with self.assertRaisesRegex(
+                ValueError, rf"{config_file.name}:1: 无法解析拆分后的调用"
+            ):
+                list(iter_api_configs(config_file.name))
+
+    def test_api_config_reject_callback_continues_with_later_calls(self):
+        rejected = []
+        with tempfile.NamedTemporaryFile("w", encoding="utf-8") as config_file:
+            config_file.write(
+                "paddle.add(Tensor([1]), Tensor([1]))\n"
+                "paddle.zeros([1])\n"
+            )
+            config_file.flush()
+            configs = list(
+                iter_api_configs(
+                    config_file.name,
+                    on_reject=lambda *details: rejected.append(details),
+                )
+            )
+
+        self.assertEqual([config.api_name for config in configs], ["paddle.zeros"])
+        self.assertEqual(rejected[0][1], 1)
+        self.assertIn("无法解析拆分后的调用", str(rejected[0][3]))
+
+    def test_normalize_file_keeps_valid_calls_and_records_rejects(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            source = Path(temp_dir) / "input.txt"
+            output = Path(temp_dir) / "normalized.txt"
+            rejects = Path(temp_dir) / "normalized.txt.unparsed.txt"
+            source.write_text(
+                "paddle.zeros([1])paddle.randn([2])\n"
+                "paddle.foopaddle.bar(1)\n",
+                encoding="utf-8",
+            )
+            counts = normalize_file(source, output, rejects)
+
+            self.assertEqual(
+                output.read_text(encoding="utf-8").splitlines(),
+                ["paddle.zeros([1])", "paddle.randn([2])"],
+            )
+            self.assertEqual(counts["split_lines"], 1)
+            self.assertEqual(counts["rejected"], 1)
+            self.assertIn(":2\t", rejects.read_text(encoding="utf-8"))
+
+
+if __name__ == "__main__":
+    unittest.main()