cobol-java-v3/tests/test_orchestrator.py

"""OR-01~12: orchestrator 管道中枢单元测试 (mock 所有外部依赖)"""

import sys, os, json, time
from pathlib import Path
from unittest.mock import MagicMock, patch, Mock
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from orchestrator import run_pipeline, _done
from data.diff_result import VerificationRun, FieldResult
from config import Config


def _min_cfg():
    c = Config()
    c.runner_mode = "native"
    c.llm_model = "mock-model"
    c.llm_timeout = 5
    c.llm_cache_dir = ".cache/test-llm"
    c.max_llm_cost = 10
    c.quality_gate_mode = "warn"
    c.quality_gate_decision_threshold = 0.5
    c.quality_gate_paragraph_threshold = 0.5
    c.max_quality_retries = 1
    c.dialect = "ibm"
    c.tolerance = 0.01
    c.coverage_default = "boundary"
    c.num_records = 100
    c.spark_master = "local[*]"
    return c


def _real_field(name="WS-A", level=5):
    from data.field_tree import Field
    return Field(name=name, level=level, pic="9(4)", usage="DISPLAY",
                 offset=0, length=4, decimal=0, signed=False)


# ── OR-01: Normal path ──

@patch("orchestrator.Path")
@patch("orchestrator.LLMClient")
@patch("orchestrator.Agent1Parser")
@patch("orchestrator.extract_structure")
@patch("orchestrator.generate_data")
@patch("orchestrator.classify_program")
@patch("hina.strategy.supplement")
@patch("orchestrator.check_coverage")
@patch("orchestrator.gate_check")
@patch("orchestrator.CobolRunner")
@patch("orchestrator.NativeJavaRunner")
@patch("orchestrator.shutil")
@patch("orchestrator.DataWriter")
@patch("orchestrator.CobolBinaryReader")
@patch("orchestrator.align_records")
@patch("orchestrator.compare_field")
@patch("orchestrator.Agent3Diagnostic")
@patch("orchestrator.ReportGenerator")
def test_orchestrator_normal(mock_rg, mock_a3, mock_cf, mock_align, mock_cbr,
                              mock_dw, mock_shutil, mock_njr, mock_cobr,
                              mock_gate, mock_cov, mock_supp, mock_hina,
                              mock_data, mock_struct, mock_a1p, mock_llm,
                              mock_path):
    """OR-01: 正常路径 → VerificationRun"""
    mock_shutil.which.return_value = "/usr/bin/java"
    mock_struct.return_value = {"total_branches": 4, "branch_tree_obj": None,
                                 "decision_points": [{"kind": "IF"}]}
    mock_data.return_value = [{"WS-A": "100"}, {"WS-A": "200"}]
    mock_hina.return_value = {"category": "condition_heavy", "confidence": 0.85,
                               "features": {}, "required_tests": 5,
                               "strategy_params": {}}
    mock_supp.return_value = []
    mock_cov.return_value = {"branch_rate": 0.8, "decision_rate": 0.5, "note": "static"}
    mock_gate.return_value = {"passed": True}
    mock_cf.return_value = FieldResult(field_name="WS-A", status="PASS")

    # CobolRunner
    mock_cobr_inst = MagicMock()
    mock_cobr_inst.compile.return_value = MagicMock(success=True, artifact_path="/tmp/test")
    mock_cobr_inst.run.return_value = MagicMock(success=True)
    mock_cobr.return_value = mock_cobr_inst

    # NativeJavaRunner
    mock_njr_inst = MagicMock()
    mock_njr_inst.compile.return_value = MagicMock(success=True, artifact_path="/tmp/java.jar")
    mock_njr_inst.run.return_value = MagicMock(success=True, records=[{"CUST-ID": "1", "WS-A": "100"}])
    mock_njr.return_value = mock_njr_inst

    # align_records
    mock_align.return_value = [({"CUST-ID": "1", "WS-A": "100"}, {"CUST-ID": "1", "WS-A": "100"}, "MATCHED")]

    # Agent1Parser
    mock_a1p_inst = MagicMock()
    mock_tree = MagicMock()
    f1 = _real_field("WS-A", 5)
    f2 = _real_field("WS-B", 10)
    mock_tree.fields = [f1, f2]
    mock_tree.flatten.return_value = {"WS-A": f1, "WS-B": f2}
    mock_a1p_inst.parse.return_value = mock_tree
    mock_a1p.return_value = mock_a1p_inst

    # Path read_text
    mock_path.return_value.read_text.return_value = "01 WS-GROUP. 05 WS-A PIC 9(4)."
    mock_path.return_value.stem = "TestProg"
    mock_path.return_value.parent = MagicMock()

    # Agent2Data
    from data.test_case import TestSuite
    mock_a2_inst = MagicMock()
    mock_a2_inst.design.return_value = TestSuite(test_cases=[])
    with patch("orchestrator.Agent2Data", return_value=mock_a2_inst):
        cfg = _min_cfg()
        vr = run_pipeline(cfg, "/fake/copybook.cpy", "/fake/program.cbl",
                          "/fake/java", "/fake/mapping.yaml")
    assert isinstance(vr, VerificationRun)


# ── OR-02: cobol_testgen empty structure ──

@patch("orchestrator.Path")
@patch("orchestrator.LLMClient")
@patch("orchestrator.Agent1Parser")
@patch("orchestrator.extract_structure")
def test_orchestrator_empty_structure(mock_struct, mock_a1p, mock_llm, mock_path):
    """OR-02: empty structure → pipeline continues"""
    mock_a1p_inst = MagicMock()
    mock_tree = MagicMock()
    f1 = _real_field("WS-A", 5)
    mock_tree.fields = [f1]
    mock_tree.flatten.return_value = {"WS-A": f1}
    mock_a1p_inst.parse.return_value = mock_tree
    mock_a1p.return_value = mock_a1p_inst
    mock_struct.return_value = {"total_branches": 0, "branch_tree_obj": None}
    mock_path.return_value.read_text.return_value = "01 WS-GROUP. 05 WS-A PIC 9(4)."
    mock_path.return_value.stem = "Test"

    cfg = _min_cfg()
    with patch("orchestrator.Agent2Data") as m_a2:
        m_a2_inst = MagicMock()
        from data.test_case import TestSuite
        m_a2_inst.design.return_value = TestSuite(test_cases=[])
        m_a2.return_value = m_a2_inst
        vr = run_pipeline(cfg, "/f/cpy", "/f/cbl", "/f/java", "/f/map")
    assert isinstance(vr, VerificationRun)


# ── OR-03: HINA Agent throws ──

@patch("orchestrator.Path")
@patch("orchestrator.LLMClient")
@patch("orchestrator.Agent1Parser")
@patch("orchestrator.extract_structure")
@patch("orchestrator.generate_data")
@patch("orchestrator.classify_program")
def test_orchestrator_hina_exception(mock_hina, mock_data, mock_struct,
                                      mock_a1p, mock_llm, mock_path):
    """OR-03: HINA 异常 → pipeline 继续"""
    mock_hina.side_effect = Exception("HINA failed")
    mock_data.return_value = []
    mock_struct.return_value = {"total_branches": 0, "branch_tree_obj": None}
    mock_a1p_inst = MagicMock()
    mock_tree = MagicMock()
    f1 = _real_field("WS-A", 5)
    mock_tree.fields = [f1]
    mock_tree.flatten.return_value = {"WS-A": f1}
    mock_a1p_inst.parse.return_value = mock_tree
    mock_a1p.return_value = mock_a1p_inst
    mock_path.return_value.read_text.return_value = "01 WS-GROUP."
    mock_path.return_value.stem = "Test"

    cfg = _min_cfg()
    with patch("orchestrator.Agent2Data") as m_a2:
        m_a2_inst = MagicMock()
        from data.test_case import TestSuite
        m_a2_inst.design.return_value = TestSuite(test_cases=[])
        m_a2.return_value = m_a2_inst
        vr = run_pipeline(cfg, "/f/cpy", "/f/cbl", "/f/java", "/f/map")
    assert isinstance(vr, VerificationRun)


# ── OR-04: Quality gate fails ──

@patch("orchestrator.Path")
@patch("orchestrator.LLMClient")
@patch("orchestrator.Agent1Parser")
@patch("orchestrator.extract_structure")
@patch("orchestrator.generate_data")
@patch("orchestrator.classify_program")
@patch("hina.strategy.supplement")
@patch("orchestrator.check_coverage")
@patch("orchestrator.gate_check")
def test_orchestrator_quality_warn(mock_gate, mock_cov, mock_supp, mock_hina,
                                    mock_data, mock_struct, mock_a1p,
                                    mock_llm, mock_path):
    """OR-04: 质量门禁失败 → QUALITY_WARN"""
    mock_hina.return_value = {"category": "test", "confidence": 0.5,
                               "features": {}, "required_tests": 3, "strategy_params": {}}
    mock_data.return_value = []
    mock_struct.return_value = {"total_branches": 10, "branch_tree_obj": None}
    mock_supp.return_value = []
    mock_cov.return_value = {"branch_rate": 0.3}
    mock_gate.return_value = {"passed": False, "issues": {"decision_gaps": [1]}}
    mock_a1p_inst = MagicMock()
    mock_tree = MagicMock()
    f1 = _real_field("WS-A", 5)
    mock_tree.fields = [f1]
    mock_tree.flatten.return_value = {"WS-A": f1}
    mock_a1p_inst.parse.return_value = mock_tree
    mock_a1p.return_value = mock_a1p_inst
    mock_path.return_value.read_text.return_value = "01 WS-GROUP."
    mock_path.return_value.stem = "Test"

    cfg = _min_cfg()
    with patch("orchestrator.Agent2Data") as m_a2:
        m_a2_inst = MagicMock()
        from data.test_case import TestSuite
        m_a2_inst.design.return_value = TestSuite(test_cases=[])
        m_a2.return_value = m_a2_inst
        vr = run_pipeline(cfg, "/f/cpy", "/f/cbl", "/f/java", "/f/map")
    assert isinstance(vr, VerificationRun)


# ── OR-05: cobc compile fails → BLOCKED ──

@patch("orchestrator.Path")
@patch("orchestrator.LLMClient")
@patch("orchestrator.Agent1Parser")
@patch("orchestrator.extract_structure")
@patch("orchestrator.generate_data")
@patch("orchestrator.classify_program")
@patch("hina.strategy.supplement")
@patch("orchestrator.check_coverage")
@patch("orchestrator.gate_check")
@patch("orchestrator.CobolRunner")
def test_orchestrator_cobc_fail(mock_cobr, mock_gate, mock_cov, mock_supp,
                                 mock_hina, mock_data, mock_struct, mock_a1p,
                                 mock_llm, mock_path):
    """OR-07: cobc 编译失败 → BLOCKED"""
    mock_hina.return_value = {"category": "test", "confidence": 0.5,
                               "features": {}, "required_tests": 3, "strategy_params": {}}
    mock_data.return_value = []
    mock_struct.return_value = {"total_branches": 2, "branch_tree_obj": None}
    mock_supp.return_value = []
    mock_cov.return_value = {"branch_rate": 1.0}
    mock_gate.return_value = {"passed": True}
    mock_cobr_inst = MagicMock()
    mock_cobr_inst.compile.return_value = MagicMock(success=False, log="cobc error",
                                                     artifact_path="")
    mock_cobr.return_value = mock_cobr_inst
    mock_a1p_inst = MagicMock()
    mock_tree = MagicMock()
    f1 = _real_field("WS-A", 5)
    mock_tree.fields = [f1]
    mock_tree.flatten.return_value = {"WS-A": f1}
    mock_a1p_inst.parse.return_value = mock_tree
    mock_a1p.return_value = mock_a1p_inst
    mock_path.return_value.read_text.return_value = "01 WS-GROUP. 05 WS-A PIC 9(4)."
    mock_path.return_value.stem = "Test"
    mock_path.return_value.parent = MagicMock()

    cfg = _min_cfg()
    with patch("orchestrator.Agent2Data") as m_a2, \
         patch("orchestrator.shutil") as m_shutil:
        m_shutil.which.return_value = None  # java not needed at this stage
        m_a2_inst = MagicMock()
        from data.test_case import TestSuite
        m_a2_inst.design.return_value = TestSuite(test_cases=[])
        m_a2.return_value = m_a2_inst
        vr = run_pipeline(cfg, "/f/cpy", "/f/cbl", "/f/java", "/f/map")
    # Pipeline should exit with BLOCKED from cobc compile failure
    assert vr.status in ("BLOCKED", "ERROR")


# ── OR-12: _done helper ──

def test_done_helper():
    """OR-12: _done 设置正确的状态/exit_code/duration"""
    vr = VerificationRun(program="T")
    t0 = time.time() - 0.1  # 100ms ago so duration is reliable
    result = _done(vr, t0, "PASS", 0)
    assert result.status == "PASS"
    assert result.exit_code == 0
    # duration might be 0 in fast environments; check that helper ran
    assert result == vr