From a2fabcc7a6e1b6bf165702503153359d51994d58 Mon Sep 17 00:00:00 2001 From: Peter Zhang <18501667167@qq.com> Date: Sun, 31 May 2026 17:28:26 +0800 Subject: [PATCH] =?UTF-8?q?test:=20=E4=BF=AE=E5=A4=8D=E7=AB=AF=E5=88=B0?= =?UTF-8?q?=E7=AB=AF=E7=AE=A1=E9=81=93=E8=BF=90=E8=A1=8C=E5=99=A8=E5=92=8C?= =?UTF-8?q?=20Layer=20B=20IndexError=20-=20Closes=20#14?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - run_pipeline.py: 修复 subprocess env 传递、parsed_path 检测、Unicode 编码 - test_main_health.py: 修复 _is_functional_section 空章节名 IndexError - 端到端测试管道: doc_parser → ir_generation(4 steps) → acceptance tests - 测试发现问题汇总至 dev issue #15 --- scripts/run_pipeline.py | 33 ++++++++++++++++------------ tests/acceptance/test_main_health.py | 2 ++ 2 files changed, 21 insertions(+), 14 deletions(-) diff --git a/scripts/run_pipeline.py b/scripts/run_pipeline.py index 8886477..ab27e23 100644 --- a/scripts/run_pipeline.py +++ b/scripts/run_pipeline.py @@ -42,11 +42,14 @@ def run_doc_parser(docx_path: str, output_dir: str) -> str | None: print(f"[1/3] Parsing document: {docx_path}") result = parse_document(docx_path, output_dir, dry_run=False) - parsed_path = result.get("output") - if parsed_path and os.path.isfile(parsed_path): + # parse_document returns {source, sections, image_sources, image_analysis} + # Output is saved as _parsed.json in output_dir + basename = os.path.splitext(os.path.basename(docx_path))[0] + parsed_path = os.path.join(output_dir, f"{basename}_parsed.json") + if os.path.isfile(parsed_path): print(f" → {parsed_path}") return parsed_path - print(" ✗ doc_parser failed to produce output", file=sys.stderr) + print(f" [FAIL] doc_parser output not found: {parsed_path}", file=sys.stderr) return None @@ -55,10 +58,11 @@ def run_doc_parser(docx_path: str, output_dir: str) -> str | None: def run_ir_pipeline(parsed_path: str) -> str | None: """Run the ir_generation steps. Returns path to ir_final.json or None.""" - config.set_input_file(parsed_path) os.makedirs(config.PROJECT_OUTPUT, exist_ok=True) os.makedirs(config.IR_OUTPUT, exist_ok=True) os.makedirs(config.FINAL_OUTPUT, exist_ok=True) + env = os.environ.copy() + env["IR_INPUT_JSON"] = parsed_path steps = [ ("step1_semantic_index.py", "Semantic Index"), @@ -72,7 +76,7 @@ def run_ir_pipeline(parsed_path: str) -> str | None: for script, label in steps: script_path = PROJECT_ROOT / "skills" / "ir_generation_skill" / script if not script_path.exists(): - print(f" ✗ Missing: {script}", file=sys.stderr) + print(f" [FAIL] Missing: {script}", file=sys.stderr) continue print(f" Running {script} ({label})...") @@ -80,28 +84,29 @@ def run_ir_pipeline(parsed_path: str) -> str | None: [sys.executable, str(script_path)], cwd=str(PROJECT_ROOT), capture_output=True, text=True, + env=env, ) if result.returncode != 0: - print(f" ✗ {script} failed (exit {result.returncode})", file=sys.stderr) + print(f" [FAIL] {script} failed (exit {result.returncode})", file=sys.stderr) print(result.stderr[-500:], file=sys.stderr) else: # Print last line of stdout for brief progress lines = result.stdout.strip().split("\n") last = lines[-1] if lines else "done" - print(f" ✓ {label}: {last[:120]}") + print(f" [OK] {label}: {last[:120]}") if os.path.isfile(config.IR_FINAL_JSON): print(f" → {config.IR_FINAL_JSON}") return config.IR_FINAL_JSON - print(" ✗ IR generation did not produce ir_final.json", file=sys.stderr) + print(" [FAIL] IR generation did not produce ir_final.json", file=sys.stderr) return None # ── Stage 3: Acceptance Tests ──────────────────────────────────────────────── -def run_acceptance_tests() -> int: +def run_acceptance_tests(parsed_json_path: str) -> int: """Run QE acceptance tests. Returns pytest exit code.""" print("[3/3] Running QE acceptance tests...") @@ -111,7 +116,7 @@ def run_acceptance_tests() -> int: sys.executable, "-m", "pytest", str(test_dir), "-v", "--run-acceptance", "--ir-path", config.IR_FINAL_JSON, - "--parsed-path", config.INPUT_JSON, + "--parsed-path", parsed_json_path, "--tb=short", ], cwd=str(PROJECT_ROOT), @@ -141,7 +146,7 @@ def main(): out_dir = args.output_dir or str(PROJECT_ROOT / "output") parsed_path = run_doc_parser(docx, out_dir) if not parsed_path: - print("\n✗ Pipeline blocked at Stage 1 (doc_parser)", file=sys.stderr) + print("\n[FAIL] Pipeline blocked at Stage 1 (doc_parser)", file=sys.stderr) # Create tracking issue for dev-agent _maybe_create_blocking_issue("doc_parser", f"Input: {docx}") sys.exit(1) @@ -157,15 +162,15 @@ def main(): # Stage 2: IR generation ir_path = run_ir_pipeline(parsed_path) if not ir_path: - print("\n✗ Pipeline blocked at Stage 2 (ir_generation)", file=sys.stderr) + print("\n[FAIL] Pipeline blocked at Stage 2 (ir_generation)", file=sys.stderr) _maybe_create_blocking_issue("ir_generation", f"Parsed: {parsed_path}") sys.exit(1) - print(f"\n✓ Pipeline complete: {ir_path}") + print(f"\n[OK] Pipeline complete: {ir_path}") # Stage 3: Acceptance tests if args.test: - exit_code = run_acceptance_tests() + exit_code = run_acceptance_tests(parsed_path) sys.exit(exit_code) diff --git a/tests/acceptance/test_main_health.py b/tests/acceptance/test_main_health.py index 3224e19..0a55bca 100644 --- a/tests/acceptance/test_main_health.py +++ b/tests/acceptance/test_main_health.py @@ -95,6 +95,8 @@ def _is_functional_section(section_name: str) -> bool: return False # Documents with only a title (no section number) — check for functional keywords sec_num = _section_number(section_name) + if not sec_num: + return False if "." not in sec_num and not sec_num[0].isdigit(): func_keywords = ["策略", "规则", "功能", "限制", "流程", "配置", "场景", "约束", "条件", "方案", "逻辑", "处理", "机制", "禁止"] -- 2.52.0