fix(runner): isolate black-box groups in per-group work dirs; add test

cobol_testgen/runner.py

@@ -12,6 +12,7 @@ from dataclasses import dataclass, field
 from pathlib import Path
 
 from . import file_io
+from .blackbox import BlackBoxGroup
 
 logger = logging.getLogger(__name__)
 
@@ -1002,7 +1003,7 @@ def run_external_groups(program_name: str, outdir: str, temp_dir: str,
                         fields_dict: list[dict], fd_fields: dict,
                         select_info: dict, open_dir: dict,
                         roles: dict, field_to_fd: dict,
-                        groups: list, source_dir: str,
+                        groups: list[BlackBoxGroup], source_dir: str,
                         data_fields: list[dict] | None = None,
                         multi_write_fds: set | None = None,
                         java_src_dir: str | None = None
@@ -1049,8 +1050,14 @@ def run_external_groups(program_name: str, outdir: str, temp_dir: str,
         base = Path(outdir) / 'black_box' / label
         src_in_dir = base / 'cobol' / 'input'
         dst_out_dir = base / 'cobol' / 'output'
+        group_work = work_dir / f'bb_{label}'
+        if group_work.exists():
+            shutil.rmtree(str(group_work), ignore_errors=True)
+        group_work.mkdir(parents=True, exist_ok=True)
         if src_in_dir.exists():
             shutil.rmtree(str(src_in_dir), ignore_errors=True)
+        if dst_out_dir.exists():
+            shutil.rmtree(str(dst_out_dir), ignore_errors=True)
 
         output_input_files(
             records, src_in_dir, f'{program_name}_{label}', roles,
@@ -1064,19 +1071,19 @@ def run_external_groups(program_name: str, outdir: str, temp_dir: str,
         if src_in_dir.is_dir():
             for src in sorted(src_in_dir.iterdir()):
                 if src.is_file() and src.suffix.lower() != '.json':
-                    shutil.copy2(str(src), str(work_dir / src.name))
+                    shutil.copy2(str(src), str(group_work / src.name))
 
-        _clean_gcda(str(work_dir))
+        _clean_gcda(str(group_work))
         group = GroupInfo(
             name=f'{program_name}_black_box_{label}',
             records=records,
-            expected_outputs=[{}] * len(records),
+            expected_outputs=[],
             fd_field_dicts=fd_field_dicts,
             open_dir=open_dir,
             select_info=select_info,
             multi_write_fds=multi_write_fds or set(),
         )
-        r = run_group(group, exe_path, str(work_dir), log_dir=log_dir)
+        r = run_group(group, exe_path, str(group_work), log_dir=log_dir)
         results.append(r)
         logger.info(f"  组 '{group.name}': returncode={r.returncode}, "
                     f"{'✓' if r.passed else '✗'}")
@@ -1087,15 +1094,15 @@ def run_external_groups(program_name: str, outdir: str, temp_dir: str,
                 continue
             sel = select_info.get(fd_name, {})
             assign = sel.get('assign', fd_name) if isinstance(sel, dict) else fd_name
-            src = os.path.join(str(work_dir), assign)
+            src = os.path.join(str(group_work), assign)
             if os.path.exists(src):
                 shutil.copy2(src, str(dst_out_dir / assign))
 
         if java_cfg:
             jrep = _run_java_scene(
-                program_name, label,
+                program_name, f'bb_{label}',
                 base / 'java' / 'output',
-                str(work_dir),
+                str(group_work),
                 src_in_dir, dst_out_dir,
                 fd_field_dicts, open_dir, select_info,
                 java_cfg, log_dir=log_dir, label=label,
@@ -1103,6 +1110,7 @@ def run_external_groups(program_name: str, outdir: str, temp_dir: str,
             jrep['section'] = 'black_box'
             java_reports.append(jrep)
 
+    _clean_gcda(str(work_dir))
     return results, java_reports
 
 

tests/cobol_testgen/test_run_external_groups.py

@@ -0,0 +1,58 @@
+"""run_external_groups 编排测试（mock 编译/执行，不依赖 cobol/java 工具链）"""
+
+import os
+import sys
+from pathlib import Path
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
+
+from cobol_testgen import runner
+from cobol_testgen.blackbox import BlackBoxGroup
+
+
+def test_run_external_groups_orchestration(tmp_path, monkeypatch):
+    fields_dict = [
+        {"name": "R01APPL-ID", "level": 3, "pic": "X(8)",
+         "pic_info": {"type": "alphanumeric", "digits": 0, "decimal": 0,
+                      "length": 8, "signed": False}},
+    ]
+    fd_fields = {"R01INNFIL": ["R01APPL-ID"], "W01OUTFIL": ["R01APPL-ID"]}
+    select_info = {"R01INNFIL": {"assign": "ZAN04R01"},
+                   "W01OUTFIL": {"assign": "ZAN04W01",
+                                 "organization": "SEQUENTIAL"}}
+    open_dir = {"R01INNFIL": "INPUT", "W01OUTFIL": "OUTPUT"}
+    roles = {"R01APPL-ID": "input"}
+    field_to_fd = {"R01APPL-ID": "R01INNFIL"}
+
+    monkeypatch.setattr(runner, "compile_sub_modules", lambda *a, **k: [])
+    monkeypatch.setattr(runner, "compile_program",
+                        lambda *a, **k: str(tmp_path / "P.exe"))
+    monkeypatch.setattr(runner, "_discover_java", lambda *a, **k: None)
+
+    calls = []
+
+    def fake_run_group(group, exe_path, temp_dir, log_dir=None):
+        calls.append((group.name, temp_dir))
+        Path(temp_dir, "ZAN04W01").write_bytes(b"A0000001")
+        return runner.GroupResult(name=group.name, returncode=0, passed=True)
+
+    monkeypatch.setattr(runner, "run_group", fake_run_group)
+
+    groups = [
+        BlackBoxGroup(label="g1",
+                      records=[{"R01APPL-ID": "A0000001"}],
+                      term_types=["normal"]),
+        BlackBoxGroup(label="g2", records=[], term_types=[]),
+    ]
+    outdir = tmp_path / "out"
+    results, jreports = runner.run_external_groups(
+        "P", str(outdir), str(tmp_path / "work"),
+        fields_dict, fd_fields, select_info, open_dir,
+        roles, field_to_fd, groups, source_dir=str(tmp_path / "src"),
+    )
+
+    assert [r.name for r in results] == ["P_black_box_g1"]
+    assert jreports == []
+    out_file = outdir / "black_box" / "g1" / "cobol" / "output" / "ZAN04W01"
+    assert out_file.read_bytes() == b"A0000001"
+    assert "bb_g1" in calls[0][1]