86 lines
3.4 KiB
Python
86 lines
3.4 KiB
Python
"""MERGE 管道支持：USING 输入识别 + OUTPUT PROCEDURE 内联 + 合并输入记录注入。
|
||
|
||
对应根因：
|
||
1. MERGE/SORT 的 USING 文件从不 OPEN → scan_open_statements 不识别为输入 → 无输入数据。
|
||
2. MERGE 的 OUTPUT PROCEDURE 段不被内联进分支树 → 其决策点未被采集。
|
||
3. 即使识别输入，也没生成供 MERGE 消费的 SA*/BO* 记录。
|
||
"""
|
||
|
||
import sys, os
|
||
from pathlib import Path
|
||
import pytest
|
||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
|
||
|
||
from cobol_testgen.read import (
|
||
preprocess, resolve_copybooks, extract_procedure_division, extract_data_division,
|
||
parse_data_division, scan_open_statements, scan_sort_merge_directions,
|
||
scan_all_file_directions, parse_file_control, parse_file_section,
|
||
)
|
||
from cobol_testgen.core import build_branch_tree
|
||
from cobol_testgen.coverage import collect_decision_points
|
||
|
||
# TNA 测试数据目录: 环境变量优先 → 大赛兄弟目录兜底（见 .env.example）
|
||
_TNA = Path(os.environ.get(
|
||
"COBOL_TNA_ROOT",
|
||
Path(__file__).resolve().parent.parent.parent.parent / "cobol-tna-system",
|
||
))
|
||
_SRC = str(_TNA / "src")
|
||
_CPY = str(_TNA / "cpy")
|
||
|
||
pytestmark = pytest.mark.skipif(
|
||
not _TNA.exists(), reason=f"TNA test data not found: {_TNA} (set COBOL_TNA_ROOT)"
|
||
)
|
||
|
||
|
||
def _load(pid):
|
||
with open(os.path.join(_SRC, pid + '.cbl'), encoding='utf-8-sig') as f:
|
||
src = f.read()
|
||
resolved = resolve_copybooks(src, _SRC, extra_search_paths=[_CPY])
|
||
pp = preprocess(resolved)
|
||
return resolved, pp
|
||
|
||
|
||
def test_merge_using_files_recognized_as_input():
|
||
"""MERGE ... USING SALARY-FILE BONUS-FILE → 两者应识别为 INPUT。"""
|
||
_, pp = _load('KYU09MRG')
|
||
dirs = scan_sort_merge_directions(extract_procedure_division(pp))
|
||
assert dirs.get('SALARY-FILE') == 'INPUT'
|
||
assert dirs.get('BONUS-FILE') == 'INPUT'
|
||
# W01/W02 仅 OPEN OUTPUT，不受影响
|
||
assert 'W01OUTFIL' not in dirs
|
||
|
||
|
||
def test_scan_all_file_directions_includes_open_and_using():
|
||
"""combined 函数应同时含 OPEN 与 MERGE USING 方向。"""
|
||
_, pp = _load('KYU09MRG')
|
||
dirs = scan_all_file_directions(extract_procedure_division(pp))
|
||
assert dirs.get('SALARY-FILE') == 'INPUT'
|
||
assert dirs.get('BONUS-FILE') == 'INPUT'
|
||
assert dirs.get('W01OUTFIL') == 'OUTPUT'
|
||
assert dirs.get('W02OUTFIL') == 'OUTPUT'
|
||
|
||
|
||
def test_output_procedure_inlined_into_branch_tree():
|
||
"""2000MRGOUTSOR 的决策点（IF NOT WRK-MERGE-EOF-Y / IF MERGE-REC-TYPE）应被采集。"""
|
||
resolved, pp = _load('KYU09MRG')
|
||
fields = parse_data_division(extract_data_division(pp))
|
||
fields_dict = [f.__dict__ for f in fields]
|
||
proc_div = extract_procedure_division(pp)
|
||
tree, _ = build_branch_tree(proc_div, fields_dict, full_source=pp)
|
||
dps, _ = collect_decision_points(tree, fields_dict)
|
||
labels = [dp.label for dp in dps]
|
||
# 主流程 D01FKICOD
|
||
assert any('D01FKICOD' in l for l in labels), labels
|
||
# OUTPUT PROCEDURE 决策点应出现（此前只有 1 个）
|
||
assert any('WRK-MERGE-EOF-Y' in l for l in labels), labels
|
||
assert any('MERGE-REC-TYPE' in l for l in labels), labels
|
||
assert len(dps) >= 3, [dp.label for dp in dps]
|
||
|
||
|
||
def test_merge_using_files_in_file_section():
|
||
"""SALARY-FILE/BONUS-FILE 应存在于 FILE-SECTION，供输入文件写入。"""
|
||
_, pp = _load('KYU09MRG')
|
||
fs = parse_file_section(pp)
|
||
assert 'SALARY-FILE' in fs
|
||
assert 'BONUS-FILE' in fs
|