From ff51bda96270b1f95cc44afa009ad9f328889193 Mon Sep 17 00:00:00 2001 From: hangshuo652 Date: Sun, 9 Aug 2026 17:43:16 +0800 Subject: [PATCH] test: cover between/hostvar/gcov-merge, class conditions, schema drop-tables, gixsql fixes --- test_tertiary.py | 63 ++++ tests/cobol_testgen/test_coverage_locator.py | 121 ++++++++ tests/cobol_testgen/test_fileio_pd_binary.py | 71 +++++ tests/cobol_testgen/test_kin07_coordinate.py | 216 ++++++++++++++ tests/cobol_testgen/test_merge_pipeline.py | 74 +++++ tests/cobol_testgen/test_read_into_layout.py | 138 +++++++++ tests/cobol_testgen/test_search_phase3.py | 108 +++++++ tests/cobol_testgen/test_skill_key_suffix.py | 148 ++++++++++ tests/cobol_testgen/test_sqlca_byteorder.py | 38 +++ tests/cobol_testgen/test_to_sql_between.py | 86 ++++++ .../test_to_sql_between_dbinput.py | 188 ++++++++++++ .../test_to_sql_dispatch_seed.py | 126 ++++++++ tests/cobol_testgen/test_to_sql_insert_pk.py | 177 +++++++++++ .../cobol_testgen/test_to_sql_schema_table.py | 135 +++++++++ .../test_to_sql_select_hostvar.py | 279 ++++++++++++++++++ .../config/test_program_schema_drop_tables.py | 37 +++ .../runners/test_gixsql_current_timestamp.py | 58 ++++ tests/runners/test_gixsql_schema_qualifier.py | 99 +++++++ tests/runners/test_sqlcode_normalize.py | 66 +++++ tests/test_kin08dbu_fixes.py | 169 +++++++++++ tests/test_kin09csv_fixes.py | 97 ++++++ tests/test_orchestrator_db_gcov_merge.py | 65 ++++ 22 files changed, 2559 insertions(+) create mode 100644 test_tertiary.py create mode 100644 tests/cobol_testgen/test_coverage_locator.py create mode 100644 tests/cobol_testgen/test_fileio_pd_binary.py create mode 100644 tests/cobol_testgen/test_kin07_coordinate.py create mode 100644 tests/cobol_testgen/test_merge_pipeline.py create mode 100644 tests/cobol_testgen/test_read_into_layout.py create mode 100644 tests/cobol_testgen/test_search_phase3.py create mode 100644 tests/cobol_testgen/test_skill_key_suffix.py create mode 100644 tests/cobol_testgen/test_sqlca_byteorder.py create mode 100644 tests/cobol_testgen/test_to_sql_between.py create mode 100644 tests/cobol_testgen/test_to_sql_between_dbinput.py create mode 100644 tests/cobol_testgen/test_to_sql_dispatch_seed.py create mode 100644 tests/cobol_testgen/test_to_sql_insert_pk.py create mode 100644 tests/cobol_testgen/test_to_sql_schema_table.py create mode 100644 tests/cobol_testgen/test_to_sql_select_hostvar.py create mode 100644 tests/config/test_program_schema_drop_tables.py create mode 100644 tests/runners/test_gixsql_current_timestamp.py create mode 100644 tests/runners/test_gixsql_schema_qualifier.py create mode 100644 tests/runners/test_sqlcode_normalize.py create mode 100644 tests/test_kin08dbu_fixes.py create mode 100644 tests/test_kin09csv_fixes.py create mode 100644 tests/test_orchestrator_db_gcov_merge.py diff --git a/test_tertiary.py b/test_tertiary.py new file mode 100644 index 0000000..1864015 --- /dev/null +++ b/test_tertiary.py @@ -0,0 +1,63 @@ +"""Quick unit test for _coordinate_tertiary_fd (new front-sequence design). + +与 tests/cobol_testgen/test_kin07_coordinate.py 对应，验证： +- 记录0 主+全部明细键一致（全匹配） +- rec1 主+明细1、rec2 主+明细2 单匹配，键单调递增 +- R03 同键重复块（休暇种别 01-04）+ 长休暇时长 +""" +import sys, os +sys.path.insert(0, os.path.dirname(__file__)) +import logging +logging.basicConfig(level=logging.INFO, format='%(message)s') + +from cobol_testgen.__init__ import _coordinate_tertiary_fd, _find_key_pairs_inner + +data_fields = [ + {'name': 'SWEMP-ID', 'pic': 'X(008)'}, + {'name': 'SWDATE', 'pic': '9(008)'}, + {'name': 'SREMP-ID', 'pic': 'X(008)'}, + {'name': 'SRWORK-DATE', 'pic': '9(008)'}, + {'name': 'SLEMP-ID', 'pic': 'X(008)'}, + {'name': 'SLDATE', 'pic': '9(008)'}, + {'name': 'SLLEAVE-TYPE', 'pic': 'X(002)'}, + {'name': 'SLSTART-TIME', 'pic': '9(004)'}, + {'name': 'SLEND-TIME', 'pic': '9(004)'}, +] + +records = [ + {'SWEMP-ID': f'R{i:07d}', 'SWDATE': f'{20000101 + i:08d}', + 'SREMP-ID': f'C{i:07d}', 'SRWORK-DATE': f'{20000101 + i:08d}', + 'SLEMP-ID': f'I{i:07d}', 'SLDATE': f'{20000101 + i:08d}', + 'SLLEAVE-TYPE': '99', 'SLSTART-TIME': '0822', 'SLEND-TIME': '0922'} + for i in range(12) +] + +fd_prefixes = ['SW', 'SR', 'SL'] + +print('=== _find_key_pairs_inner ===') +print('SW-SR:', _find_key_pairs_inner('SW', 'SR', data_fields)) +print('SW-SL:', _find_key_pairs_inner('SW', 'SL', data_fields)) + +print() +print('=== _coordinate_tertiary_fd ===') +_coordinate_tertiary_fd(records, fd_prefixes, data_fields) +print() +for i, r in enumerate(records[:8]): + print(f' records[{i}]: SWEMP={r["SWEMP-ID"]}, SWDATE={r["SWDATE"]}, ' + f'SREMP={r["SREMP-ID"]}, SLEMP={r["SLEMP-ID"]}, LT={r["SLLEAVE-TYPE"]}, ' + f'{r["SLSTART-TIME"]}-{r["SLEND-TIME"]}') + +# Verify +assert records[0]['SWEMP-ID'] == records[0]['SREMP-ID'] == records[0]['SLEMP-ID'], 'rec0 三键应一致' +assert records[1]['SWEMP-ID'] == records[1]['SREMP-ID'], 'rec1 应为主+明细1 匹配' +assert records[2]['SWEMP-ID'] != records[2]['SREMP-ID'], 'rec2 明细1 不应匹配' +# 键单调递增（merge-join 指针对齐前提） +k0, k1, k2 = records[0]['SWEMP-ID'], records[1]['SWEMP-ID'], records[2]['SWEMP-ID'] +assert k0 < k1 < k2, f'前置序列主键应递增: {k0} < {k1} < {k2}' +# R03 同键重复块 + 休暇种别 + 长时长（块2） +assert all(records[i]['SLEMP-ID'] == records[0]['SLEMP-ID'] for i in range(4)), 'R03 前4条键应一致' +assert [records[i]['SLLEAVE-TYPE'] for i in range(4)] == ['01', '02', '03', '04'], '休暇种别 01-04' +for i in range(4, 8): + assert records[i]['SLSTART-TIME'] == '0600' and records[i]['SLEND-TIME'] == '2200', '块2 长时长' +print() +print('ALL ASSERTIONS PASSED') diff --git a/tests/cobol_testgen/test_coverage_locator.py b/tests/cobol_testgen/test_coverage_locator.py new file mode 100644 index 0000000..0d34524 --- /dev/null +++ b/tests/cobol_testgen/test_coverage_locator.py @@ -0,0 +1,121 @@ +"""决策点行号定位测试（T4 修复）。 + +注入 SQLCODE 映射代码后，gixpp 输出的 pp.cbl 中 `IF SQLCODE = -803` +等语句行号发生偏移。`locate_decision_lines` 必须精确定位到 +`IF SQLCODE = -803` 所在行，而非 `IF SQLCODE = -1555` 或映射块的 END-IF。 +""" + +import sys, os, re +sys.stdout.reconfigure(encoding='utf-8') +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +from cobol_testgen.coverage import _build_search_patterns, _normalize, locate_decision_lines + + +class D: + kind = 'IF' + label = 'IF SQLCODE = -803' + condition = None + + +def test_if_pattern_does_not_duplicate_if_keyword(): + """IF 决策点模式不能是 'IF IF ...'（label 已含 IF）""" + dp = D() + pats = _build_search_patterns(dp) + assert pats, "patterns should not be empty" + for p in pats: + assert "IF\\b\\s+IF" not in p, f"duplicated IF keyword in pattern: {p}" + + +def test_if_pattern_matches_exact_condition_line(): + """模式应匹配 'IF SQLCODE = -803' 行，而不匹配 -1555/-19""" + dp = D() + pats = _build_search_patterns(dp) + assert pats[0] is not None + assert re.search(pats[0], "IF SQLCODE = -803") is not None + assert re.search(pats[0], "IF SQLCODE = -1555") is None + assert re.search(pats[0], "IF SQLCODE = -19") is None + assert re.search(pats[0], "IF SQLCODE = -2067") is None + + +def test_locate_finds_shifted_line_in_pp_source(): + """pp.cbl 中注入映射后，决策点定位到正确的 -803 行""" + pp = ( + " CALL \"GIXSQLEndSQL\"\n" + " END-CALL.\n" + " IF SQLCODE = -1555\n" + " MOVE -803 TO SQLCODE\n" + " END-IF\n" + " IF SQLCODE = -2067\n" + " MOVE -803 TO SQLCODE\n" + " END-IF\n" + " IF SQLCODE = -19\n" + " MOVE -803 TO SQLCODE\n" + " END-IF\n" + " IF SQLCODE = 0\n" + " ADD 1 TO CUN-DB-INS\n" + " END-IF.\n" + " IF SQLCODE = -803\n" + " PERFORM 2000MAJSOR-UPD\n" + " END-IF.\n" + ) + dp = D() + dp.source_line = 0 + locate_decision_lines([dp], pp) + assert dp.source_line == 15, f"expected line 15 (IF SQLCODE = -803), got {dp.source_line}" + + +class MultiLineD: + kind = 'IF' + label = 'MERGE-REC-TYPE = CNS-PAY-TYPE-SALARY' + condition = None + + +def test_locate_multiline_if_condition(): + """条件被换行拆分的 IF 应定位到 'IF <首字段>' 起始行。 + + MERGE 输出过程的 IF MERGE-REC-TYPE / = CNS-PAY-TYPE-SALARY 分两行， + 单行模式无法匹配；应回退到按首字段定位 IF 行（gcov 分支标记依赖源行）。 + """ + pp = ( + " PERFORM WITH TEST AFTER\n" + " UNTIL WRK-MERGE-EOF-Y\n" + " RETURN MERGE-FILE\n" + " INTO MERGE-REC\n" + " AT END\n" + " SET WRK-MERGE-EOF-Y TO TRUE\n" + " END-RETURN\n" + " IF NOT WRK-MERGE-EOF-Y\n" + " MOVE MERGE-EMP-ID\n" + " TO WRK-CURRENT-EMP\n" + " IF MERGE-REC-TYPE\n" + " = CNS-PAY-TYPE-SALARY\n" + " MOVE MERGE-PAY-AMOUNT\n" + " TO WRK-SALARY-AMOUNT\n" + " ELSE\n" + " MOVE MERGE-PAY-AMOUNT\n" + " TO WRK-BONUS-AMOUNT\n" + " END-IF\n" + " END-IF\n" + " END-PERFORM.\n" + ) + dp = MultiLineD() + dp.source_line = 0 + locate_decision_lines([dp], pp) + assert dp.source_line == 11, f"expected line 11 (IF MERGE-REC-TYPE), got {dp.source_line}" + + +def test_locate_multiline_if_does_not_mislocate_other_if(): + """首字段定位不应误命中其它字段相同的 IF（如 IF NOT WRK-MERGE-EOF-Y 前的 WRK 前缀）。""" + pp = ( + " IF NOT WRK-MERGE-EOF-Y\n" + " IF MERGE-REC-TYPE\n" + " = CNS-PAY-TYPE-SALARY\n" + " MOVE MERGE-PAY-AMOUNT\n" + " TO WRK-SALARY-AMOUNT\n" + " END-IF\n" + " END-IF.\n" + ) + dp = MultiLineD() + dp.source_line = 0 + locate_decision_lines([dp], pp) + assert dp.source_line == 2, f"expected line 2 (IF MERGE-REC-TYPE), got {dp.source_line}" diff --git a/tests/cobol_testgen/test_fileio_pd_binary.py b/tests/cobol_testgen/test_fileio_pd_binary.py new file mode 100644 index 0000000..e104d8f --- /dev/null +++ b/tests/cobol_testgen/test_fileio_pd_binary.py @@ -0,0 +1,71 @@ +"""file_io PACKED-DECIMAL/COMP-3 小数位字节长度 + BINARY 大端序测试。 + +复现根因（JIN03RAN）： +1. get_storage_length 对 COMP-3/PACKED-DECIMAL 用 (digits+2)//2，忽略 decimal。 + 9(3)V9(2) 应为 3 字节（5 位有效数字+符号），却按 2 字节 → R01 记录 199 字节 + vs GnuCOBOL FD 200 字节 → 逐记录错位 → AVG-SCORE 运行时乱值 → 'A' 等级分支不可达。 +2. pack_value/unpack_value 对 COMP/BINARY 用小端 '<'，但本 GnuCOBOL（GC32-BDB-SP1， + 主机兼容）COMP 按大端存储（见 test_sqlca_byteorder.py 记载）→ EMP-COUNT 运行时乱值。 +""" + +import sys +import os + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) + +import pytest + +from cobol_testgen.file_io import get_storage_length, pack_value, unpack_value + + +def _pd(digits, decimal, signed=False): + return { + 'usage': 'PACKED-DECIMAL', + 'pic_info': {'type': 'numeric', 'length': digits + decimal, + 'digits': digits, 'decimal': decimal, 'signed': signed}, + } + + +def _bin(digits, signed=False): + return { + 'usage': 'BINARY', + 'pic_info': {'type': 'numeric', 'length': digits, + 'digits': digits, 'decimal': 0, 'signed': signed}, + } + + +# ── 1. PACKED-DECIMAL 小数位字节长度 ── + +def test_pd_storage_length_includes_decimal(): + """9(3)V9(2) = 5 位 + 符号 = 3 字节；9(2)（无小数）仍为 2 字节。""" + assert get_storage_length(_pd(3, 2)) == 3, '9(3)V9(2) PACKED-DECIMAL 应占 3 字节' + assert get_storage_length(_pd(2, 0)) == 2, '9(2) PACKED-DECIMAL 应占 2 字节' + assert get_storage_length(_pd(7, 2, signed=True)) == 5, 'S9(7)V99 PACKED-DECIMAL 应占 5 字节' + + +def test_pd_pack_decimal_full_digits(): + """900.00 应打包为 90 00 0F（3 字节，保留全部 5 位）。""" + b = pack_value('90000', _pd(3, 2)) + assert b == bytes.fromhex('90000f'), f'900.00 打包应为 90000f, 实际 {b.hex()}' + b2 = pack_value('60000', _pd(3, 2)) + assert b2 == bytes.fromhex('60000f'), f'600.00 打包应为 60000f, 实际 {b2.hex()}' + + +def test_pd_unpack_decimal_roundtrip(): + """打包/解包回读应保留 5 位（含小数位）。""" + f = _pd(3, 2) + for val in ('90000', '60000', '00423'): + assert unpack_value(pack_value(val, f), f) == val, f'{val} 回读应一致' + + +# ── 2. BINARY 大端序（匹配本 GnuCOBOL 的 COMP 存储）── + +def test_binary_pack_big_endian(): + """203 在 PIC 9(5) BINARY 应打包为大端 00 00 00 CB（4 字节）。""" + b = pack_value('00203', _bin(5)) + assert b == bytes.fromhex('000000cb'), f'203 BINARY 应大端 000000cb, 实际 {b.hex()}' + + +def test_binary_unpack_big_endian(): + """大端 00 00 00 CB 解包应得到 00203。""" + assert unpack_value(bytes.fromhex('000000cb'), _bin(5)) == '00203' diff --git a/tests/cobol_testgen/test_kin07_coordinate.py b/tests/cobol_testgen/test_kin07_coordinate.py new file mode 100644 index 0000000..17fc068 --- /dev/null +++ b/tests/cobol_testgen/test_kin07_coordinate.py @@ -0,0 +1,216 @@ +"""KIN07DAI 型多文件照合程序的前缀检测与前置序列构造测试。 + +复现根因：_coordinate_multi_file_keys 从 01 记录名推导前缀（R01INNREC→'R01'）， +但 COPY REPLACING 把字段前缀改为 SW/SR/SL，导致键对匹配不到、跨文件键协调失效。 +修复后应生成前置匹配序列，运行时 merge-join 应覆盖全匹配/单明细匹配等组合。 +""" + +import sys +import os + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) + +import pytest + +from cobol_testgen.__init__ import ( + _coordinate_multi_file_keys, + _coordinate_tertiary_fd, + _find_key_pairs_inner, +) + + +def _group(name): + return {'name': name, 'level': 1, 'pic': None, + 'pic_info': {'type': 'group', 'digits': 0, 'decimal': 0, 'length': 80, 'signed': False}, + 'section': 'FILE', 'is_filler': False, 'redefines': None, 'usage': 'DISPLAY', + 'occurs': 0, 'occurs_depending': None, 'value': None, 'values': None} + + +def _field(name, level, pic, ftype, length): + return {'name': name, 'level': level, 'pic': pic, + 'pic_info': {'type': ftype, 'digits': 0, 'decimal': 0, 'length': length, 'signed': False}, + 'section': 'FILE', 'is_filler': False, 'redefines': None, 'usage': 'DISPLAY', + 'occurs': 0, 'occurs_depending': None, 'value': None, 'values': None} + + +def _make_data_fields(): + # 字段顺序与源文件一致：01 记录紧跟其子字段 + return [ + _group('R01INNREC'), + _field('SWEMP-ID', 3, 'X(008)', 'alphanumeric', 8), + _field('SWDATE', 3, '9(008)', 'numeric', 8), + _group('R02INNREC'), + _field('SREMP-ID', 3, 'X(008)', 'alphanumeric', 8), + _field('SRWORK-DATE', 3, '9(008)', 'numeric', 8), + _field('SRLEAVE-TYPE', 3, 'X(002)', 'alphanumeric', 2), + _group('R03INNREC'), + _field('SLEMP-ID', 3, 'X(008)', 'alphanumeric', 8), + _field('SLDATE', 3, '9(008)', 'numeric', 8), + _field('SLLEAVE-TYPE', 3, 'X(002)', 'alphanumeric', 2), + _field('SLSTART-TIME', 3, '9(004)', 'numeric', 4), + _field('SLEND-TIME', 3, '9(004)', 'numeric', 4), + ] + + +def _make_records(n=12): + recs = [] + for i in range(n): + recs.append({ + 'SWEMP-ID': f'R{i:07d}', 'SWDATE': f'{20000101 + i:08d}', + 'SREMP-ID': f'C{i:07d}', 'SRWORK-DATE': f'{20000101 + i:08d}', + 'SRLEAVE-TYPE': '99', + 'SLEMP-ID': f'I{i:07d}', 'SLDATE': f'{20000101 + i:08d}', + 'SLLEAVE-TYPE': '99', + 'SLSTART-TIME': '0822', 'SLEND-TIME': '0922', + }) + return recs + + +FILE_SEC = { + 'R01INNFIL': ['R01INNREC'], + 'R02INNFIL': ['R02INNREC'], + 'R03INNFIL': ['R03INNREC'], + 'W01OUTFIL': ['W01OUTREC'], +} +OPEN_DIR = { + 'R01INNFIL': 'INPUT', + 'R02INNFIL': 'INPUT', + 'R03INNFIL': 'INPUT', + 'W01OUTFIL': 'OUTPUT', +} + + +def test_find_key_pairs_uses_field_prefix_not_record_prefix(): + """字段前缀 SW/SR/SL 应能配对。""" + fields = _make_data_fields() + assert _find_key_pairs_inner('SW', 'SR', fields), 'SW-SR 应配对' + assert _find_key_pairs_inner('SW', 'SL', fields), 'SW-SL 应配对' + + +def test_coordinate_multi_file_keys_builds_front_sequence(): + """KIN07DAI 形态（记录名 R01INNREC + 字段 SW/SR/SL）下协调应生效： + 记录0 三文件键一致、R03 同键重复块 + 休暇种别注入、键单调递增。""" + fields = _make_data_fields() + records = _make_records(12) + _coordinate_multi_file_keys( + records, [], fields, {}, FILE_SEC, + term_types=['normal'] * len(records), open_dir=OPEN_DIR, + ) + + r0 = records[0] + assert r0['SWEMP-ID'] == r0['SREMP-ID'] == r0['SLEMP-ID'], '记录0 三文件键应一致' + assert r0['SWDATE'] == r0['SRWORK-DATE'] == r0['SLDATE'], '记录0 三文件日期应一致' + + # R03 同键重复块（休暇种别 01-04 + WHEN OTHER '99'） + r03_keys = [records[i]['SLEMP-ID'] for i in range(5)] + assert all(k == r03_keys[0] for k in r03_keys), 'R03 前5条记录键应一致' + lts = [records[i]['SLLEAVE-TYPE'] for i in range(5)] + assert lts == ['01', '02', '03', '04', '99'], "休暇种别应注入 01-04+'99', 实际 {lts}".format(lts=lts) + + # 键单调递增（避免 merge-join 指针跳过） + k0, k1 = records[0]['SWEMP-ID'], records[1]['SWEMP-ID'] + assert k0 < k1, f'前置序列主键应递增: {k0} < {k1}' + + +def _simulate_merge(R01, R02, R03): + """模拟 KIN07DAI 的 3 文件 merge-join，返回 (pattern_counts, match_flags)。 + + pattern 集合: A=全匹配, B=仅R02, C=仅R03, D=无匹配 + match_flags: (r02_match, r03_match, leave_calc, lt_seen) + """ + def key(e, d): + return (str(e).strip(), str(d)) + + n1, n2, n3 = len(R01), len(R02), len(R03) + r1 = r2 = r3 = 0 + e1 = e2 = e3 = False + + def rdR1(): + nonlocal r1, e1 + if r1 < n1: + r1 += 1 + else: + e1 = True + + def rdR2(): + nonlocal r2, e2 + if r2 < n2: + r2 += 1 + else: + e2 = True + + def rdR3(): + nonlocal r3, e3 + if r3 < n3: + r3 += 1 + else: + e3 = True + + rdR1(); rdR2(); rdR3() + pat = {'A': 0, 'B': 0, 'C': 0, 'D': 0} + r02m = r03m = calcs = 0 + lt_seen = set() + while not e1: + r01k = key(R01[r1 - 1][0], R01[r1 - 1][1]) + ptn = 4 + while not e2: + if key(R02[r2 - 1][0], R02[r2 - 1][1]) >= r01k: + break + rdR2() + if not e2 and key(R02[r2 - 1][0], R02[r2 - 1][1]) == r01k: + ptn = 2 + r02m += 1 + rdR2() + while not e3: + if key(R03[r3 - 1][0], R03[r3 - 1][1]) >= r01k: + break + rdR3() + if not e3 and key(R03[r3 - 1][0], R03[r3 - 1][1]) == r01k: + r03m += 1 + while True: + calcs += 1 + lt_seen.add(R03[r3 - 1][3]) + rdR3() + if e3: + break + if key(R03[r3 - 1][0], R03[r3 - 1][1]) != r01k: + break + ptn = 1 if ptn == 2 else 3 + pat[['A', 'B', 'C', 'D'][ptn - 1]] += 1 + rdR1() + return pat, (r02m, r03m, calcs, lt_seen) + + +def test_coordination_reaches_all_match_combinations_at_runtime(): + """协调后运行时 merge-join 应覆盖全匹配/仅R02/仅R03 三种组合。""" + fields = _make_data_fields() + records = _make_records(12) + _coordinate_multi_file_keys( + records, [], fields, {}, FILE_SEC, + term_types=['normal'] * len(records), open_dir=OPEN_DIR, + ) + + R01 = [(r['SWEMP-ID'], r['SWDATE']) for r in records] + R02 = [(r['SREMP-ID'], r['SRWORK-DATE']) for r in records] + R03 = [(r['SLEMP-ID'], r['SLDATE'], r['SLLEAVE-TYPE'], r['SLLEAVE-TYPE']) for r in records] + pat, (r02m, r03m, calcs, lt_seen) = _simulate_merge(R01, R02, R03) + + assert r02m >= 1, 'R02 照合应至少发生 1 次' + assert r03m >= 1, 'R03 照合应至少发生 1 次' + assert calcs >= 4, '休暇計算应至少 4 次（EVALUATE 全 WHEN）' + assert lt_seen >= {'01', '02', '03', '04'}, f'休暇種別应全覆盖, 实际 {lt_seen}' + assert pat['A'] >= 1, f'全匹配(PATTERN-A)应覆盖, 实际 {pat}' + assert pat['B'] >= 1, f'仅R02匹配(PATTERN-B)应覆盖, 实际 {pat}' + assert pat['C'] >= 1, f'仅R03匹配(PATTERN-C)应覆盖, 实际 {pat}' + + +def test_coordinate_tertiary_fd_long_leave_for_cap(): + """PATTERN-C 用判别块应注入长休暇时长（覆盖 8h 上限 T 分支）。""" + fields = _make_data_fields() + records = _make_records(12) + _coordinate_tertiary_fd(records, ['SW', 'SR', 'SL'], fields) + + # 块2（索引 max(3,5)=5..9）：长时长 0600-2200 + for i in range(5, 10): + assert records[i]['SLSTART-TIME'] == '0600', f'rec{i} 长休暇开始时间' + assert records[i]['SLEND-TIME'] == '2200', f'rec{i} 长休暇结束时间' diff --git a/tests/cobol_testgen/test_merge_pipeline.py b/tests/cobol_testgen/test_merge_pipeline.py new file mode 100644 index 0000000..b5a3378 --- /dev/null +++ b/tests/cobol_testgen/test_merge_pipeline.py @@ -0,0 +1,74 @@ +"""MERGE 管道支持：USING 输入识别 + OUTPUT PROCEDURE 内联 + 合并输入记录注入。 + +对应根因： +1. MERGE/SORT 的 USING 文件从不 OPEN → scan_open_statements 不识别为输入 → 无输入数据。 +2. MERGE 的 OUTPUT PROCEDURE 段不被内联进分支树 → 其决策点未被采集。 +3. 即使识别输入，也没生成供 MERGE 消费的 SA*/BO* 记录。 +""" + +import sys, os +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 + +_SRC = r'C:\Users\marye\Desktop\2026技术大赛\cobol-tna-system\src' +_CPY = r'C:\Users\marye\Desktop\2026技术大赛\cobol-tna-system\cpy' + + +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 diff --git a/tests/cobol_testgen/test_read_into_layout.py b/tests/cobol_testgen/test_read_into_layout.py new file mode 100644 index 0000000..3590252 --- /dev/null +++ b/tests/cobol_testgen/test_read_into_layout.py @@ -0,0 +1,138 @@ +"""RI-01~05: opaque FD + READ INTO — WS 子字段值序列化回 FD 记录字节 + 跨记录 key 差异保证""" + +import sys, os, json, tempfile +from pathlib import Path +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +from cobol_testgen import _serialize_read_into_records +from cobol_testgen.core import build_branch_tree + + +def _mk_fields(): + return [ + {'name': 'R01INNREC', 'level': 1, 'pic': 'X(080)', + 'pic_info': {'type': 'alphanumeric', 'digits': 0, 'decimal': 0, 'length': 80, 'signed': False}, + 'is_filler': False, 'redefines': None, 'is_88': False}, + {'name': 'WRK-ABSENCE-REC', 'level': 1, 'pic': '', + 'pic_info': {'type': 'unknown', 'digits': 0, 'decimal': 0, 'length': 0, 'signed': False}, + 'is_filler': False, 'redefines': None, 'is_88': False}, + {'name': 'WRK-EMP-ID', 'level': 5, 'pic': 'X(008)', + 'pic_info': {'type': 'alphanumeric', 'digits': 0, 'decimal': 0, 'length': 8, 'signed': False}, + 'is_filler': False, 'redefines': None, 'is_88': False}, + {'name': 'WRK-YEAR-MONTH', 'level': 5, 'pic': 'X(006)', + 'pic_info': {'type': 'alphanumeric', 'digits': 0, 'decimal': 0, 'length': 6, 'signed': False}, + 'is_filler': False, 'redefines': None, 'is_88': False}, + {'name': 'WRK-ABSENT-HOURS', 'level': 5, 'pic': '9(004)V9(001)', + 'pic_info': {'type': 'numeric', 'digits': 4, 'decimal': 1, 'length': 0, 'signed': False}, + 'is_filler': False, 'redefines': None, 'is_88': False}, + {'name': 'FILLER', 'level': 5, 'pic': 'X(062)', + 'pic_info': {'type': 'alphanumeric', 'digits': 0, 'decimal': 0, 'length': 62, 'signed': False}, + 'is_filler': True, 'redefines': None, 'is_88': False}, + {'name': 'WRK-PREV-EMP-ID', 'level': 1, 'pic': 'X(008)', + 'pic_info': {'type': 'alphanumeric', 'digits': 0, 'decimal': 0, 'length': 8, 'signed': False}, + 'is_filler': False, 'redefines': None, 'is_88': False}, + ] + + +def _mk_records(): + return [ + {'R01INNREC': 'A' + '0' * 79, + 'WRK-EMP-ID': 'G0000001', 'WRK-YEAR-MONTH': 'H00001', 'WRK-ABSENT-HOURS': '01501'}, + {'R01INNREC': 'A' + '0' * 79, + 'WRK-EMP-ID': 'G0000007', 'WRK-YEAR-MONTH': 'H00007', 'WRK-ABSENT-HOURS': '01507'}, + {'R01INNREC': 'A' + '0' * 79, + 'WRK-EMP-ID': 'G0000007', 'WRK-YEAR-MONTH': 'H00008', 'WRK-ABSENT-HOURS': '01508'}, + ] + + +def test_serialize_into_fd_bytes(): + """RI-01: INTO 组子字段值按 PIC 布局 pack 回 R01INNREC 字节""" + records = _mk_records() + file_sec = {'R01INNFIL': ['R01INNREC']} + assignments = {'WRK-ABSENCE-REC': [{'type': 'read_into', 'file': 'R01INNFIL', 'source_vars': []}]} + _serialize_read_into_records(records, assignments, _mk_fields(), file_sec) + + r0 = records[0]['R01INNREC'] + assert r0[0:8] == 'G0000001', r0[0:8] + assert r0[8:14] == 'H00001', r0[8:14] + assert r0[14:19] == '01501', r0[14:19] + assert len(r0) == 80 + assert r0[19:].strip(' ') == '', r0[19:] + + +def test_key_variation_both_pairs(): + """RI-02: 相邻记录 key 既有相同又有差异（同 key 累加 + 新 key flush 两路径均可达）""" + records = _mk_records() + file_sec = {'R01INNFIL': ['R01INNREC']} + assignments = {'WRK-ABSENCE-REC': [{'type': 'read_into', 'file': 'R01INNFIL', 'source_vars': []}]} + _serialize_read_into_records(records, assignments, _mk_fields(), file_sec) + + keys = [r['R01INNREC'][0:8] for r in records] + has_same = any(keys[i] == keys[i + 1] for i in range(len(keys) - 1)) + has_diff = any(keys[i] != keys[i + 1] for i in range(len(keys) - 1)) + assert has_same, f"no same-key pair: {keys}" + assert has_diff, f"no diff-key pair: {keys}" + + +def test_all_same_key_gets_break(): + """RI-03: 全记录同 key 时自动注入差异，保证新 key 分支可达""" + records = [ + {'R01INNREC': 'A' + '0' * 79, + 'WRK-EMP-ID': 'G0000001', 'WRK-YEAR-MONTH': 'H00001', 'WRK-ABSENT-HOURS': '01501'}, + {'R01INNREC': 'A' + '0' * 79, + 'WRK-EMP-ID': 'G0000001', 'WRK-YEAR-MONTH': 'H00002', 'WRK-ABSENT-HOURS': '01502'}, + {'R01INNREC': 'A' + '0' * 79, + 'WRK-EMP-ID': 'G0000001', 'WRK-YEAR-MONTH': 'H00003', 'WRK-ABSENT-HOURS': '01503'}, + ] + file_sec = {'R01INNFIL': ['R01INNREC']} + assignments = {'WRK-ABSENCE-REC': [{'type': 'read_into', 'file': 'R01INNFIL', 'source_vars': []}]} + _serialize_read_into_records(records, assignments, _mk_fields(), file_sec) + + keys = [r['R01INNREC'][0:8] for r in records] + has_diff = any(keys[i] != keys[i + 1] for i in range(len(keys) - 1)) + assert has_diff, f"no diff-key pair: {keys}" + + +def test_noop_without_read_into(): + """RI-04: 无 READ INTO 或 FD 有子字段时不改动记录""" + records = _mk_records() + file_sec = {'R01INNFIL': ['R01INNREC']} + before = [dict(r) for r in records] + _serialize_read_into_records(records, {}, _mk_fields(), file_sec) + assert records == before + + fields_with_children = [ + {'name': 'R01INNREC', 'level': 1, 'pic': 'X(080)', + 'pic_info': {'type': 'alphanumeric', 'digits': 0, 'decimal': 0, 'length': 80, 'signed': False}, + 'is_filler': False, 'redefines': None, 'is_88': False}, + {'name': 'R01EMP-ID', 'level': 5, 'pic': 'X(008)', + 'pic_info': {'type': 'alphanumeric', 'digits': 0, 'decimal': 0, 'length': 8, 'signed': False}, + 'is_filler': False, 'redefines': None, 'is_88': False}, + ] + records2 = _mk_records() + before2 = [dict(r) for r in records2] + assignments = {'WRK-ABSENCE-REC': [{'type': 'read_into', 'file': 'R01INNFIL', 'source_vars': []}]} + _serialize_read_into_records(records2, assignments, fields_with_children, file_sec) + assert records2 == before2 + + +def test_multiline_read_into_parsed(): + """RI-05: READ / INTO 分行书写仍能解析为 read_into 赋值""" + proc = """PROCEDURE DIVISION. +MAIN. + PERFORM UNTIL WRK-EOF = 'Y' + READ R01INNFIL + INTO WRK-ABSENCE-REC + AT END + MOVE 'Y' TO WRK-EOF + END-READ + IF WRK-EOF = 'Y' + EXIT PERFORM + END-IF + END-PERFORM. + GOBACK. +""" + tree, asgn = build_branch_tree(proc, []) + read_keys = [k for k in asgn if any( + isinstance(a, dict) and a.get('type') == 'read_into' for a in asgn[k])] + assert read_keys == ['WRK-ABSENCE-REC'], read_keys + assert asgn['WRK-ABSENCE-REC'][0]['file'] == 'R01INNFIL' diff --git a/tests/cobol_testgen/test_search_phase3.py b/tests/cobol_testgen/test_search_phase3.py new file mode 100644 index 0000000..d54b094 --- /dev/null +++ b/tests/cobol_testgen/test_search_phase3.py @@ -0,0 +1,108 @@ +"""SEARCH 分支路径生成 + 二元 COMPUTE 链反演（Phase 3 — TDD RED） + +目标：KYU05DED SEARCH WHEN 分支（#9）从不可达变为可达， +配合二元 `-` COMPUTE（WRK-TAXABLE-INCOME = WRK-GROSS-PAYMENT - CNS-TAX-BASIC-DEDUCTION） +反演到输入 R01GROSS-PAYMENT。 +""" + +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) + +from cobol_testgen.design_mcdc import enum_paths +from cobol_testgen.core import trace_to_root, invert_through_chain +from cobol_testgen.coverage import collect_decision_points, mark_coverage +from cobol_testgen.models import BrSearch, BrSeq, CondLeaf, CondAnd + + +def _make_search_node(): + node = BrSearch("IW-ENTRY") + node.when_list = [("A <= SUBJ", BrSeq())] + node.cond_trees = [CondLeaf("A", "<=", "SUBJ")] + node.has_at_end = True + node.at_end_seq = BrSeq() + return node + + +FIELDS = [ + {"name": "A(1)", "pic_info": {"type": "numeric", "digits": 9}}, + {"name": "SUBJ", "pic_info": {"type": "numeric", "digits": 9}}, +] + + +# ── 1. enum_paths 生成 SEARCH WHEN 路径 ── + +def test_enum_paths_search_marks_when_branch(): + """enum_paths 对 BrSearch 应生成 WHEN 路径，使 _mark_search 标记 WHEN 分支""" + node = _make_search_node() + paths = enum_paths(node, FIELDS) + assert len(paths) >= 1 + dps, _ = collect_decision_points(node, FIELDS, [0]) + mark_coverage(dps, [], paths, FIELDS) + dp = dps[0] + assert dp.kind == "SEARCH" + assert "WHEN A <= SUBJ" in dp.active_branches + + +def test_enum_paths_search_marks_at_end(): + """enum_paths 应生成 AT END 路径（带不匹配 WHEN 叶的约束）""" + node = _make_search_node() + paths = enum_paths(node, FIELDS) + dps, _ = collect_decision_points(node, FIELDS, [0]) + mark_coverage(dps, [], paths, FIELDS) + dp = dps[0] + assert "AT END" in dp.active_branches + + +# ── 2. trace_to_root / invert_through_chain 二元 `-` 反演 ── + +def test_trace_minus_constant_field_to_input(): + """2 源 `-` COMPUTE：SUBJ = GROSS - DED（DED 常量 VALUE）→ 追溯到 INPUT-GROSS""" + assignments = { + "SUBJ": [{"type": "compute", "source_vars": ["GROSS", "DED"], + "op": "-", "expr": "GROSS - DED"}], + "GROSS": [{"type": "move", "source_vars": ["INPUT-GROSS"]}], + } + fields = [ + {"name": "DED", "pic_info": {"type": "numeric", "digits": 6}, "value": "50000"}, + {"name": "INPUT-GROSS", "pic_info": {"type": "numeric", "digits": 9}}, + ] + root, chain = trace_to_root("SUBJ", assignments, fields) + assert root == "INPUT-GROSS" + # 2 源 `-` COMPUTE 常量侧被折叠为单源 compute + const + assert any(a.get("const") == float("50000") for _, a in chain) + + +def test_invert_minus_constant_field_boundary(): + """SUBJ >= 4701 → GROSS - 50000 >= 4701 → INPUT-GROSS >= 54701""" + assignments = { + "SUBJ": [{"type": "compute", "source_vars": ["GROSS", "DED"], + "op": "-", "expr": "GROSS - DED"}], + "GROSS": [{"type": "move", "source_vars": ["INPUT-GROSS"]}], + } + fields = [ + {"name": "DED", "pic_info": {"type": "numeric", "digits": 6}, "value": "50000"}, + {"name": "INPUT-GROSS", "pic_info": {"type": "numeric", "digits": 9}}, + ] + root, chain = trace_to_root("SUBJ", assignments, fields) + new_root, op, val = invert_through_chain(root, chain, ">=", "4701") + assert new_root == "INPUT-GROSS" + assert op == ">=" + assert float(val) >= 54701 + + +# ── 3. _mark_search 单 CondLeaf 字段对字段（应已绿，防止回归）── + +def test_mark_search_field_to_field_constraint(): + """SUBJ <= A(1) 形态约束命中 WHEN 叶（value 侧基准字段匹配）""" + from cobol_testgen.coverage import DecisionPoint + dp = DecisionPoint(id=1, kind="SEARCH", label="IW-ENTRY", + branch_names=["WHEN A <= SUBJ", "AT END"]) + dp.when_list = [("A <= SUBJ", BrSeq())] + dp.cond_trees = [CondLeaf("A", "<=", "SUBJ")] + dp.has_other = True + branch_paths = [ + ([("SUBJ", "<=", "A(1)", True)], []), + ] + mark_coverage([dp], [], branch_paths, FIELDS) + assert "WHEN A <= SUBJ" in dp.active_branches + assert "AT END" not in dp.active_branches diff --git a/tests/cobol_testgen/test_skill_key_suffix.py b/tests/cobol_testgen/test_skill_key_suffix.py new file mode 100644 index 0000000..bf46f40 --- /dev/null +++ b/tests/cobol_testgen/test_skill_key_suffix.py @@ -0,0 +1,148 @@ +"""JIN02SKL 型"明细键─主表键"后缀配对测试。 + +复现根因：跨文件键协调 _find_key_pairs 只做同名精确匹配 / WORK- 变体 / +EMP 族回退。当明细键 base='SKILL-CODE' 与主表键 base='MST-SKILL-CODE' +（多 'MST-' 限定词）时匹配不到，导致 SEARCH ALL 永未命中、 +TBL-EMP-COUNT 恒 0，#7/#8（> ZERO）TRUE 分支不可达。 + +修复：增加"后缀匹配"层级——一方 base 是另一方真后缀且 key 类别一致 +（同为 -CODE/-ID/-NO/DATE/EMP-ID）时配对。通用实现、无程序名硬编码。 +""" + +import sys +import os + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) + +import pytest + +from cobol_testgen.__init__ import ( + _coordinate_multi_file_keys, + _find_key_pairs_inner, +) + + +def _group(name): + return {'name': name, 'level': 1, 'pic': None, + 'pic_info': {'type': 'group', 'digits': 0, 'decimal': 0, 'length': 200, 'signed': False}, + 'section': 'FILE', 'is_filler': False, 'redefines': None, 'usage': 'DISPLAY', + 'occurs': 0, 'occurs_depending': None, 'value': None, 'values': None} + + +def _field(name, level, pic, ftype, length): + return {'name': name, 'level': level, 'pic': pic, + 'pic_info': {'type': ftype, 'digits': 0, 'decimal': 0, 'length': length, 'signed': False}, + 'section': 'FILE', 'is_filler': False, 'redefines': None, 'usage': 'DISPLAY', + 'occurs': 0, 'occurs_depending': None, 'value': None, 'values': None} + + +def _skill_fields(): + # 字段顺序与源文件一致：01 记录紧跟其子字段（COPY REPLACING R01/R02） + return [ + _group('R01INNREC'), + _field('R01SKILL-CODE', 3, 'X(004)', 'alphanumeric', 4), + _field('R01EMP-ID', 3, 'X(008)', 'alphanumeric', 8), + _field('R01EMP-NAME', 3, 'X(040)', 'alphanumeric', 40), + _field('R01EVAL-DATE', 3, 'X(008)', 'alphanumeric', 8), + _field('R01SCORE', 3, '9(003)', 'numeric', 3), + _group('R02INNREC'), + _field('R02MST-SKILL-CODE', 3, 'X(004)', 'alphanumeric', 4), + _field('R02MST-LEVEL', 3, '9(002)', 'numeric', 2), + _field('R02MST-SCORE-FROM', 3, '9(003)', 'numeric', 3), + _field('R02MST-SCORE-TO', 3, '9(003)', 'numeric', 3), + _field('R02MST-LEVEL-NAME', 3, 'X(020)', 'alphanumeric', 20), + ] + + +def _skill_records(n=6): + recs = [] + for i in range(n): + recs.append({ + 'R01SKILL-CODE': f'A{i + 1:03d}', + 'R01EMP-ID': f'B{i + 1:07d}', + 'R01EMP-NAME': 'N' * 40, + 'R01EVAL-DATE': f'D{i + 1:08d}', + 'R01SCORE': f'{100 + i:03d}', + 'R02MST-SKILL-CODE': f'F{i + 1:03d}', + 'R02MST-LEVEL': f'{20 + i:02d}', + 'R02MST-SCORE-FROM': f'{300 + i:03d}', + 'R02MST-SCORE-TO': f'{400 + i:03d}', + 'R02MST-LEVEL-NAME': 'G' * 20, + }) + return recs + + +FILE_SEC = { + 'R01INNFIL': ['R01INNREC'], + 'R02INNFIL': ['R02INNREC'], + 'W01OUTFIL': ['W01OUTREC'], + 'W02OUTFIL': ['W02OUTREC'], +} +OPEN_DIR = { + 'R01INNFIL': 'INPUT', + 'R02INNFIL': 'INPUT', + 'W01OUTFIL': 'OUTPUT', + 'W02OUTFIL': 'OUTPUT', +} + + +def test_find_key_pairs_suffix_skill_code(): + """明细键 SKILL-CODE 应与主表键 MST-SKILL-CODE 后缀配对。""" + fields = _skill_fields() + pairs = _find_key_pairs_inner('R01', 'R02', fields) + assert ('R01SKILL-CODE', 'R02MST-SKILL-CODE') in pairs, f'SKILL-CODE↔MST-SKILL-CODE 应配对, 实际 {pairs}' + + +def test_find_key_pairs_suffix_no_cross_class(): + """类别不一致的后缀不应误配：DATE 类键不与 CODE 类键配对。""" + fields = _skill_fields() + pairs = _find_key_pairs_inner('R01', 'R02', fields) + date_keys = [fa for fa, _ in pairs if 'DATE' in fa.upper()] + code_keys = [fb for _, fb in pairs if '-CODE' in fb.upper()] + # R01EVAL-DATE 不应与 R02MST-SKILL-CODE 配对 + assert not any('R01EVAL-DATE' == fa and fb == 'R02MST-SKILL-CODE' for fa, fb in pairs), f'DATE 不应与 CODE 后缀配对, 实际 {pairs}' + + +def test_coordinate_multi_file_keys_skill_match(): + """协调后应使部分记录 R01SKILL-CODE == R02MST-SKILL-CODE，运行时 SEARCH 可达。""" + fields = _skill_fields() + records = _skill_records(6) + _coordinate_multi_file_keys( + records, [], fields, {}, FILE_SEC, + term_types=['normal'] * len(records), open_dir=OPEN_DIR, + ) + matched = [r for r in records + if str(r['R01SKILL-CODE']).strip() == str(r['R02MST-SKILL-CODE']).strip()] + assert matched, '至少一条记录明细键应与主表键一致' + + +def test_coordinate_exact_match_still_pairs(): + """既有同名精确匹配（EMP-ID↔EMP-ID）不受后缀层级影响。""" + fields = [ + _group('R01INNREC'), + _field('R01EMP-ID', 3, 'X(008)', 'alphanumeric', 8), + _group('R02INNREC'), + _field('R02EMP-ID', 3, 'X(008)', 'alphanumeric', 8), + ] + pairs = _find_key_pairs_inner('R01', 'R02', fields) + assert ('R01EMP-ID', 'R02EMP-ID') in pairs, f'同名 EMP-ID 应仍精确配对, 实际 {pairs}' + + +def test_coordinate_last_record_matches_table_key(): + """末条 normal 记录明细键应对齐表内键（首条 R02 键）。 + + 社员工键中断集约（JIN09RPT 型）：最终社员工组在 EOF 後の终了处理で出力される + （`IF WRK-EMP-EVAL-CNT > ZERO PERFORM 3000OUTPUSOR`）。協調が先頭のみ整列だと + 最終组に照合データがなく最终组输出分支不可達。末条记录も表内键に揃えることで + 該分支が到達可能になる。 + """ + fields = _skill_fields() + records = _skill_records(12) + _coordinate_multi_file_keys( + records, [], fields, {}, FILE_SEC, + term_types=['normal'] * len(records), open_dir=OPEN_DIR, + ) + table_key = str(records[0]['R02MST-SKILL-CODE']).strip() + last = records[-1] + assert str(last['R01SKILL-CODE']).strip() == table_key, \ + f'末条记录明细键应对齐表内键 {table_key}, 实际 {last["R01SKILL-CODE"]!r}' diff --git a/tests/cobol_testgen/test_sqlca_byteorder.py b/tests/cobol_testgen/test_sqlca_byteorder.py new file mode 100644 index 0000000..6a86202 --- /dev/null +++ b/tests/cobol_testgen/test_sqlca_byteorder.py @@ -0,0 +1,38 @@ +"""SQLCA 字节序对齐测试（T1）。 + +gixsql 以 native 小端 int 写入 SQLCA.SQLCODE。GnuCOBOL 的 COMP 按大端存储， +导致非零 SQLCODE 被错读（如 -1555 → -302383105）。必须改用 COMP-5（native）， +否则 `IF SQLCODE = -803` 类具体值判断永远无法匹配。 +""" + +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +from cobol_testgen.read import resolve_sql_includes, _BUILTIN_SQLCA +from runners.gixsql_runner import GixsqlCobolRunner + + +def test_builtin_sqlca_uses_comp5_for_numeric_fields(): + """内置 SQLCA 的 SQLCABC/SQLCODE 必须用 COMP-5（native 字节序）""" + assert "SQLCABC PIC S9(9) COMP-5" in _BUILTIN_SQLCA + assert "SQLCODE PIC S9(9) COMP-5" in _BUILTIN_SQLCA + assert "SQLERRML PIC S9(4) COMP-5" in _BUILTIN_SQLCA + assert "PIC S9(9) COMP-5" in _BUILTIN_SQLCA + + +def test_resolve_sql_includes_injects_comp5_sqlca(): + """EXEC SQL INCLUDE SQLCA 解析后注入的 SQLCA 用 COMP-5""" + src = " EXEC SQL INCLUDE SQLCA END-EXEC.\n" + out = resolve_sql_includes(src, ".") + assert "SQLCODE PIC S9(9) COMP-5" in out + + +def test_runner_inline_sqlca_uses_comp5(): + """gixsql_runner 内联 SQLCA 的数值字段用 COMP-5""" + r = GixsqlCobolRunner(gixpp_path="gixpp", lib_path="lib") + r._copybook_dirs = [] + # _expand_all_copies 内联 SQLCA 分支（copybook 不存在时） + text = " COPY SQLCA.\n" + out = r._expand_all_copies(text, []) + assert "SQLCODE PIC S9(9) COMP-5" in out + assert "SQLCABC PIC S9(9) COMP-5" in out + assert "SQLERRD PIC S9(9) COMP-5 OCCURS 6" in out diff --git a/tests/cobol_testgen/test_to_sql_between.py b/tests/cobol_testgen/test_to_sql_between.py new file mode 100644 index 0000000..586476e --- /dev/null +++ b/tests/cobol_testgen/test_to_sql_between.py @@ -0,0 +1,86 @@ +"""BETWEEN WHERE 子句解析测试。 + +根因：KYU05DED 的 `WHERE :WRK-TAXABLE-INCOME BETWEEN TAX-FROM AND TAX-TO`： +1. `_split_on_AND` 把 BETWEEN 内部的 AND 当成分隔符切断，导致约束解析为空； +2. `_RE_BETWEEN` 要求首操作数为列名，不支持 `:主机变量 BETWEEN ...`。 +修复后应能解析出 subject / lo / hi 三端（每端可为列名 / 主机变量 / 字面量）。 +""" + +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +from cobol_testgen.to_sql import _split_on_AND, sql_extract_constraints + + +def test_split_on_and_preserves_between(): + """BETWEEN 内部的 AND 不应被当作条件分隔符切断。""" + parts = _split_on_AND(':WRK-TAXABLE-INCOME BETWEEN TAX-FROM AND TAX-TO') + assert parts == [':WRK-TAXABLE-INCOME BETWEEN TAX-FROM AND TAX-TO'] + + +def test_split_on_and_between_with_following_and(): + """BETWEEN 后的其他 AND 条件仍应正确切分。""" + parts = _split_on_AND( + 'EMP-ID = :DBV-EMP-ID AND :WRK-TAXABLE-INCOME BETWEEN TAX-FROM AND TAX-TO AND STATUS = :S' + ) + assert parts == [ + 'EMP-ID = :DBV-EMP-ID', + ':WRK-TAXABLE-INCOME BETWEEN TAX-FROM AND TAX-TO', + 'STATUS = :S', + ] + + +def test_sql_extract_constraints_between_hostvar_subject(): + """`：主机变量 BETWEEN 列1 AND 列2`：subject 为主机变量，lo/hi 为列名。""" + cons = sql_extract_constraints( + ':WRK-TAXABLE-INCOME BETWEEN TAX-FROM AND TAX-TO', 'TAX_TABLE', {}, {}, {}) + assert len(cons) == 1 + c = cons[0] + assert c['type'] == 'between' + assert c['neg'] is False + assert c['subject'] == {'kind': 'host_var', 'name': 'WRK-TAXABLE-INCOME'} + assert c['lo'] == {'kind': 'column', 'name': 'TAX-FROM'} + assert c['hi'] == {'kind': 'column', 'name': 'TAX-TO'} + + +def test_sql_extract_constraints_between_column_subject(): + """`列 BETWEEN :低 AND :高`：subject 为列名，lo/hi 为主机变量。""" + cons = sql_extract_constraints( + 'SALARY BETWEEN :LO AND :HI', 'EMP', {}, {}, {}) + assert len(cons) == 1 + c = cons[0] + assert c['type'] == 'between' + assert c['subject'] == {'kind': 'column', 'name': 'SALARY'} + assert c['lo'] == {'kind': 'host_var', 'name': 'LO'} + assert c['hi'] == {'kind': 'host_var', 'name': 'HI'} + + +def test_sql_extract_constraints_between_compound(): + """BETWEEN 与其他条件组合时，全部解析且顺序保持。""" + cons = sql_extract_constraints( + 'EMP-ID = :DBV-EMP-ID AND :WRK-TAXABLE-INCOME BETWEEN TAX-FROM AND TAX-TO', + 'TAX_TABLE', {}, {}, {}) + assert len(cons) == 2 + assert cons[0]['type'] == 'host_var' + assert cons[0]['host_var'] == 'DBV-EMP-ID' + assert cons[1]['type'] == 'between' + + +def test_sql_extract_constraints_not_between(): + """NOT BETWEEN 正确解析 neg 标志。""" + cons = sql_extract_constraints(':X NOT BETWEEN A AND B', 'T', {}, {}, {}) + assert len(cons) == 1 + assert cons[0]['type'] == 'between' + assert cons[0]['neg'] is True + assert cons[0]['subject'] == {'kind': 'host_var', 'name': 'X'} + + +def test_sql_extract_constraints_between_three_part(): + """三段式：BETWEEN 之后跟后续 AND 条件。""" + cons = sql_extract_constraints( + ':X BETWEEN A AND B AND :Y > 5', 'T', {}, {}, {}) + assert len(cons) == 2 + assert cons[0]['type'] == 'between' + assert cons[0]['lo'] == {'kind': 'column', 'name': 'A'} + assert cons[0]['hi'] == {'kind': 'column', 'name': 'B'} + assert cons[1]['type'] == 'literal' + assert cons[1]['op'] == '>' diff --git a/tests/cobol_testgen/test_to_sql_between_dbinput.py b/tests/cobol_testgen/test_to_sql_between_dbinput.py new file mode 100644 index 0000000..2a481c8 --- /dev/null +++ b/tests/cobol_testgen/test_to_sql_between_dbinput.py @@ -0,0 +1,188 @@ +"""SELECT BETWEEN WHERE 约束的 DB 预置行生成测试（Phase 2）。 + +根因：KYU05DED 的 `WHERE :WRK-TAXABLE-INCOME BETWEEN TAX-FROM AND TAX-TO` +在修复解析后成为 between 约束，但 build_db_input 不处理 between 类型， +TAX_FROM/TAX_TO 仍落入默认序号值 → 运行时 SELECT 查不到行 → SQLCODE<>0。 +修复：build_db_input 对 between 约束，用记录中的 subject 值播种 lo/hi 列， +使运行时 `:hv BETWEEN TAX-FROM AND TAX-TO` 能命中。 +""" + +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +from cobol_testgen.to_sql import build_db_input + + +def _sql_meta_select_between(): + return [{ + "type": "exec_sql_select", + "table": "TAX-TABLE", + "select_list": "TAX-RATE, DEDUCTION", + "into_vars": ["DBV-TAX-RATE", "DBV-TAX-DEDUCTION"], + "where": ":WRK-TAXABLE-INCOME BETWEEN TAX-FROM AND TAX-TO", + "where_constraints": [ + {"type": "between", "neg": False, "op": "BETWEEN", + "subject": {"kind": "host_var", "name": "WRK-TAXABLE-INCOME"}, + "lo": {"kind": "column", "name": "TAX-FROM"}, + "hi": {"kind": "column", "name": "TAX-TO"}}, + ], + }] + + +def _declared_tax_table(): + return { + "TAX-TABLE": [ + {"name": "TAX_FROM", "db_type": "DECIMAL"}, + {"name": "TAX_TO", "db_type": "DECIMAL"}, + {"name": "TAX_RATE", "db_type": "DECIMAL"}, + {"name": "DEDUCTION", "db_type": "DECIMAL"}, + ], + } + + +def test_select_between_hostvar_subject_seeds_lo_hi_from_record(): + """`:hv BETWEEN TAX-FROM AND TAX-TO`：TAX_FROM/TAX_TO 播成记录中 + WRK-TAXABLE-INCOME 的值，保证运行时 BETWEEN 命中。""" + records = [{"WRK-TAXABLE-INCOME": "000000000"}] + db_input = build_db_input( + [([('SQLCODE', '=', '0', True)], {})], [], {}, + _sql_meta_select_between(), _declared_tax_table(), records=records) + rows = db_input.get('TAX-TABLE', []) + assert len(rows) == 1 + assert int(rows[0]['TAX_FROM']) == 0 + assert int(rows[0]['TAX_TO']) == 0 + + +def test_select_between_hostvar_subject_nonzero(): + """subject 非零时，lo/hi 取记录值。""" + records = [{"WRK-TAXABLE-INCOME": "000102400"}] + db_input = build_db_input( + [([('SQLCODE', '=', '0', True)], {})], [], {}, + _sql_meta_select_between(), _declared_tax_table(), records=records) + rows = db_input.get('TAX-TABLE', []) + assert len(rows) == 1 + assert int(rows[0]['TAX_FROM']) == 102400 + assert int(rows[0]['TAX_TO']) == 102400 + + +def test_select_between_hostvar_subject_moved_from_input(): + """subject 主机变量经 MOVE 链到输入键时，优先用输入键值。""" + sql_meta = [{ + "type": "exec_sql_select", + "table": "TAX-TABLE", + "select_list": "TAX-RATE, DEDUCTION", + "into_vars": ["DBV-TAX-RATE", "DBV-TAX-DEDUCTION"], + "where": ":WRK-TAXABLE-INCOME BETWEEN TAX-FROM AND TAX-TO", + "where_constraints": [ + {"type": "between", "neg": False, "op": "BETWEEN", + "subject": {"kind": "host_var", "name": "WRK-TAXABLE-INCOME"}, + "lo": {"kind": "column", "name": "TAX-FROM"}, + "hi": {"kind": "column", "name": "TAX-TO"}}, + ], + }] + records = [{"R01GROSS-PAYMENT": "000102400", + "WRK-TAXABLE-INCOME": "000000000"}] + assignments = { + 'WRK-TAXABLE-INCOME': [{'type': 'move', 'source_vars': ['R01GROSS-PAYMENT']}], + } + db_input = build_db_input( + [([('SQLCODE', '=', '0', True)], {})], [], assignments, + sql_meta, _declared_tax_table(), records=records) + rows = db_input.get('TAX-TABLE', []) + assert len(rows) == 1 + assert int(rows[0]['TAX_FROM']) == 102400 + + +def test_select_between_column_subject_seeds_subject_col(): + """`SALARY BETWEEN :LO AND :HI`：subject 是列时，把该列播成 lo 值 + （满足 lo <= col <= hi）。""" + sql_meta = [{ + "type": "exec_sql_select", + "table": "EMP", + "select_list": "EMP_NAME", + "into_vars": ["DBV-EMPLOYEE-NAME"], + "where": "SALARY BETWEEN :LO AND :HI", + "where_constraints": [ + {"type": "between", "neg": False, "op": "BETWEEN", + "subject": {"kind": "column", "name": "SALARY"}, + "lo": {"kind": "host_var", "name": "LO"}, + "hi": {"kind": "host_var", "name": "HI"}}, + ], + }] + declared = { + "EMP": [ + {"name": "SALARY", "db_type": "DECIMAL"}, + {"name": "EMP_NAME", "db_type": "CHAR", "size": 40}, + ], + } + records = [{"LO": "000005000", "HI": "000009000"}] + db_input = build_db_input( + [([('SQLCODE', '=', '0', True)], {})], [], {}, + sql_meta, declared, records=records) + rows = db_input.get('EMP', []) + assert len(rows) == 1 + sal = int(rows[0]['SALARY']) + assert 5000 <= sal <= 9000 + + +def test_select_between_sql_error_path_skips_row(): + """SQL 失败路径（SQLCODE<>0）仍不生成预置行。""" + records = [{"WRK-TAXABLE-INCOME": "000102400"}] + db_input = build_db_input( + [([('SQLCODE', '<>', '0', True)], {})], [], {}, + _sql_meta_select_between(), _declared_tax_table(), records=records) + rows = db_input.get('TAX-TABLE', []) + assert not rows + + +def _fdict_with_ws_constants(): + """KYU05DED 简化的字段定义：CNS-TAX-BASIC-DEDUCTION 常量 + MOVE/COMPUTE 链。""" + return [ + {'name': 'R01GROSS-PAYMENT', 'pic_info': {'type': 'numeric', 'digits': 9, 'decimal': 0, 'length': 0, 'signed': False}, 'section': 'INPUT', 'value': None}, + {'name': 'WRK-GROSS-PAYMENT', 'pic_info': {'type': 'numeric', 'digits': 9, 'decimal': 0, 'length': 0, 'signed': False}, 'section': 'WORKING-STORAGE', 'value': None}, + {'name': 'WRK-TAXABLE-INCOME', 'pic_info': {'type': 'numeric', 'digits': 9, 'decimal': 0, 'length': 0, 'signed': False}, 'section': 'WORKING-STORAGE', 'value': None}, + {'name': 'CNS-TAX-BASIC-DEDUCTION', 'pic_info': {'type': 'numeric', 'digits': 6, 'decimal': 0, 'length': 0, 'signed': False}, 'section': 'WORKING-STORAGE', 'value': '50000'}, + ] + + +def test_select_between_derives_runtime_value_from_input_key(): + """记录中 WRK-TAXABLE-INCOME 为合成值时，按运行时推导值播种带。 + + 输入 gross=000100000 → 运行时 taxable = gross - 50000 = 50000， + 预置行 TAX_FROM/TAX_TO 应为 50000，而非记录的合成值。""" + assignments = { + 'WRK-GROSS-PAYMENT': [{'type': 'move', 'source_vars': ['R01GROSS-PAYMENT']}], + 'WRK-TAXABLE-INCOME': [ + {'type': 'move_literal', 'literal': 'ZERO'}, + {'type': 'compute', 'source_vars': ['WRK-GROSS-PAYMENT', 'CNS-TAX-BASIC-DEDUCTION'], 'op': '-'}, + ], + } + records = [{"R01GROSS-PAYMENT": "000100000", + "WRK-TAXABLE-INCOME": "000003701"}] + db_input = build_db_input( + [([('SQLCODE', '=', '0', True)], {})], _fdict_with_ws_constants(), assignments, + _sql_meta_select_between(), _declared_tax_table(), records=records) + rows = db_input.get('TAX-TABLE', []) + assert len(rows) == 1 + assert int(rows[0]['TAX_FROM']) == 50000 + assert int(rows[0]['TAX_TO']) == 50000 + + +def test_select_between_derives_zero_for_low_gross(): + """gross < 基本控除（50000）时，运行时 taxable = 0 → 带为 [0,0]。""" + assignments = { + 'WRK-GROSS-PAYMENT': [{'type': 'move', 'source_vars': ['R01GROSS-PAYMENT']}], + 'WRK-TAXABLE-INCOME': [ + {'type': 'move_literal', 'literal': 'ZERO'}, + {'type': 'compute', 'source_vars': ['WRK-GROSS-PAYMENT', 'CNS-TAX-BASIC-DEDUCTION'], 'op': '-'}, + ], + } + records = [{"R01GROSS-PAYMENT": "000000501", + "WRK-TAXABLE-INCOME": "000003701"}] + db_input = build_db_input( + [([('SQLCODE', '=', '0', True)], {})], _fdict_with_ws_constants(), assignments, + _sql_meta_select_between(), _declared_tax_table(), records=records) + rows = db_input.get('TAX-TABLE', []) + assert len(rows) == 1 + assert int(rows[0]['TAX_FROM']) == 0 + assert int(rows[0]['TAX_TO']) == 0 + diff --git a/tests/cobol_testgen/test_to_sql_dispatch_seed.py b/tests/cobol_testgen/test_to_sql_dispatch_seed.py new file mode 100644 index 0000000..0137488 --- /dev/null +++ b/tests/cobol_testgen/test_to_sql_dispatch_seed.py @@ -0,0 +1,126 @@ +"""build_db_input 事务调度感知种子测试（JIN05UPD 根因）。 + +复现根因：UPDATE/DELETE 种子对每个 sql_ok 路径都用该路径记录的主键建行， +不区分记录运行时是否真的执行 UPDATE。JIN05UPD 的 4 条 INSERT 记录主键因此 +全部被预置进 EMPLOYEE → 运行时 INSERT 全部 -803，INSERT 成功分支不可达。 +另外：非 sql_ok INSERT 路径的记录可能不是 INSERT 类（运行时走 WHEN OTHER）， +碰撞行必须回退到真实 INSERT 记录的主键。 + +修复：按 88 级事务类型（INSERT/UPDATE/DELETE 语义标记）沿 MOVE 链追溯调度 +字段，分类每条记录；UPDATE/DELETE 种子仅对 UPDATE/DELETE 记录；INSERT 碰撞 +回退到首条 INSERT 类记录。 +""" + +import sys +import os + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) + +import pytest + +from cobol_testgen.to_sql import build_db_input + + +def _fields(): + def f(name, level, pic, ftype, length): + return {'name': name, 'level': level, 'pic': pic, + 'pic_info': {'type': ftype, 'digits': 0, 'decimal': 0, + 'length': length, 'signed': False}, + 'section': 'WS', 'is_filler': False, 'redefines': None, + 'usage': 'DISPLAY', 'occurs': 0, 'occurs_depending': None, + 'value': None, 'values': None} + base = [ + f('WRK-TRAN-TYPE', 3, 'X(001)', 'alphanumeric', 1), + f('R01TRAN-TYPE', 3, 'X(001)', 'alphanumeric', 1), + f('R01TRAN-EMP-ID', 3, 'X(008)', 'alphanumeric', 8), + f('HV-EMP-ID', 3, 'X(008)', 'alphanumeric', 8), + ] + for name, parent, vals in [('WRK-TRAN-INSERT', 'WRK-TRAN-TYPE', ['A', 'B', 'C']), + ('WRK-TRAN-UPDATE', 'WRK-TRAN-TYPE', ['D', 'E', 'F', 'G']), + ('WRK-TRAN-DELETE', 'WRK-TRAN-TYPE', ['Z'])]: + e = f(name, 88, None, 'alphanumeric', 1) + e['is_88'] = True + e['parent'] = parent + e['values'] = vals + e['value'] = vals[0] + base.append(e) + return base + + +_ASSIGN = { + 'WRK-TRAN-TYPE': [{'type': 'move', 'source_vars': ['R01TRAN-TYPE']}], + 'HV-EMP-ID': [{'type': 'move', 'source_vars': ['R01TRAN-EMP-ID']}], +} + + +def _declared(): + return {'EMPLOYEE': [{'name': 'EMP_ID', 'db_type': 'CHAR', 'size': 8}]} + + +def _sql_meta(): + return [ + {'type': 'exec_sql_insert', 'table': 'EMPLOYEE', + 'columns': ['EMP_ID'], 'host_vars': ['HV-EMP-ID'], + 'sql_text': 'INSERT INTO EMPLOYEE (EMP_ID) VALUES (:HV-EMP-ID)'}, + {'type': 'exec_sql_update', 'table': 'EMPLOYEE', + 'set_clause': 'STATUS = :HV-STATUS', + 'where': 'EMP_ID = :HV-EMP-ID', 'host_vars': ['HV-EMP-ID', 'HV-STATUS'], + 'where_constraints': [{'col': 'EMP_ID', 'type': 'host_var', 'host_var': 'HV-EMP-ID'}], + 'sql_text': 'UPDATE EMPLOYEE SET STATUS = :HV-STATUS WHERE EMP_ID = :HV-EMP-ID'}, + ] + + +def _records(): + return [ + {'R01TRAN-TYPE': 'A', 'R01TRAN-EMP-ID': 'B0000001', 'HV-EMP-ID': 'B0000001'}, # INSERT + {'R01TRAN-TYPE': 'D', 'R01TRAN-EMP-ID': 'B0000005', 'HV-EMP-ID': 'B0000005'}, # UPDATE + {'R01TRAN-TYPE': 'H', 'R01TRAN-EMP-ID': 'B0000008', 'HV-EMP-ID': 'B0000008'}, # OTHER + {'R01TRAN-TYPE': 'I', 'R01TRAN-EMP-ID': 'B0000009', 'HV-EMP-ID': 'B0000009'}, # OTHER (非sql_ok INSERT 路径) + ] + + +def _paths(): + return [ + ([], {}), # path0 sql_ok → rec0 (INSERT) + ([], {}), # path1 sql_ok → rec1 (UPDATE) + ([], {}), # path2 sql_ok → rec2 (OTHER) + ([('SQLCODE', '<>', '0', True)], {}), # path3 INSERT 错误 → rec3 (OTHER) + ] + + +_INSERT_PK = {'EMPLOYEE': ['EMP_ID']} + + +def test_update_seed_only_for_update_records(): + """INSERT/OTHER 记录不应被 UPDATE 种子注入；OTHER 记录主键不得入表。 + + B0000001 允许作为 INSERT 碰撞行出现一次（供 -803 分支），但不得因 + UPDATE 种子而额外注入；B0000008/B0000009（OTHER 记录）绝不能入表。 + """ + di = build_db_input(_paths(), _fields(), _ASSIGN, _sql_meta(), _declared(), + records=_records(), insert_pk=_INSERT_PK) + keys = [r.get('EMP_ID', '').strip() for r in di.get('EMPLOYEE', [])] + assert keys.count('B0000001') <= 1, 'INSERT 记录主键仅允许作为碰撞行出现一次' + assert 'B0000005' in keys, 'UPDATE 记录主键应被种子注入' + assert 'B0000008' not in keys, 'OTHER 记录主键不应被种子注入' + assert 'B0000009' not in keys, '非 sql_ok OTHER 记录主键不应被 UPDATE 种子注入' + + +def test_insert_collision_falls_back_to_insert_record(): + """非 sql_ok INSERT 路径的记录若非 INSERT 类，碰撞行应回退到首条 INSERT 类记录主键。""" + di = build_db_input(_paths(), _fields(), _ASSIGN, _sql_meta(), _declared(), + records=_records(), insert_pk=_INSERT_PK) + keys = [r.get('EMP_ID', '').strip() for r in di.get('EMPLOYEE', [])] + assert 'B0000001' in keys, f'碰撞行应回退到首条 INSERT 记录主键, 实际 {keys}' + + +def test_insert_success_path_skips_seed(): + """sql_ok INSERT 路径不应生成任何种子行（INSERT 成功需主键空闲）。""" + paths = [ + ([('SQLCODE', '=', '0', True)], {}), # INSERT 成功 + ] + recs = [{'R01TRAN-TYPE': 'A', 'R01TRAN-EMP-ID': 'B0000001', 'HV-EMP-ID': 'B0000001'}] + di = build_db_input(paths, _fields(), _ASSIGN, _sql_meta(), _declared(), + records=recs, insert_pk=_INSERT_PK) + keys = [r.get('EMP_ID', '').strip() for r in di.get('EMPLOYEE', [])] + assert 'B0000001' not in keys, f'INSERT 成功路径不应生成种子, 实际 {keys}' diff --git a/tests/cobol_testgen/test_to_sql_insert_pk.py b/tests/cobol_testgen/test_to_sql_insert_pk.py new file mode 100644 index 0000000..1f0df46 --- /dev/null +++ b/tests/cobol_testgen/test_to_sql_insert_pk.py @@ -0,0 +1,177 @@ +"""INSERT 表主键碰撞预置行生成测试（T2）。 + +gixsql+SQLite 下，`IF SQLCODE = -803` 分支要求运行时 INSERT 失败。 +让 INSERT 失败的唯一途径是 DB 中已存在同主键行。因此 build_db_input +必须为"要求 SQLCODE≠0 的路径"生成主键碰撞预置行（主键值取自该记录）。 +""" + +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +from cobol_testgen.to_sql import build_db_input + + +def _sql_meta_insert(table="EMP-MASTER", columns=("EMP-ID", "EMP-NAME"), + host_vars=("DBV-EMP-ID", "DBV-EMP-NAME")): + return [{ + "type": "exec_sql_insert", + "table": table, + "columns": list(columns), + "host_vars": list(host_vars), + "sql_text": f"INSERT INTO {table} ({', '.join(columns)}) VALUES ({', '.join(':'+h for h in host_vars)})", + }] + + +def test_insert_sql_ok_path_skips_collision_row(): + """要求 SQLCODE=0（INSERT 成功）的路径 → 不生成碰撞预置行""" + branch_paths = [ + ([('SQLCODE', '=', '0', True)], {}), # SQL ok → skip + ] + records = [{'DBV-EMP-ID': 'A0000001', 'DBV-EMP-NAME': 'NAME'}] + db_input = build_db_input( + branch_paths, [], {}, _sql_meta_insert(), + {'EMP-MASTER': [{'name': 'EMP-ID', 'db_type': 'CHAR', 'size': 8}]}, + records=records, + insert_pk={'EMP-MASTER': ['EMP-ID']}, + ) + assert 'EMP-MASTER' not in db_input or db_input['EMP-MASTER'] == [] + + +def test_insert_error_path_generates_pk_collision_row(): + """要求 SQLCODE<>0（INSERT 失败）的路径 → 生成主键碰撞预置行""" + branch_paths = [ + ([('SQLCODE', '<>', '0', True)], {}), # SQL fails → collision row + ] + records = [{'DBV-EMP-ID': 'A0000001', 'DBV-EMP-NAME': 'NAME'}] + db_input = build_db_input( + branch_paths, [], {}, _sql_meta_insert(), + {'EMP-MASTER': [{'name': 'EMP-ID', 'db_type': 'CHAR', 'size': 8}, + {'name': 'EMP-NAME', 'db_type': 'CHAR', 'size': 40}]}, + records=records, + insert_pk={'EMP-MASTER': ['EMP-ID']}, + ) + rows = db_input.get('EMP-MASTER', []) + assert len(rows) == 1 + assert rows[0]['EMP-ID'] == 'A0000001' + + +def test_insert_error_path_uses_record_host_var_value(): + """碰撞行主键取自记录中对应 host_var 的值""" + branch_paths = [ + ([('SQLCODE', '=', '-803', True)], {}), + ] + records = [{'DBV-EMP-ID': 'Z9999999', 'DBV-EMP-NAME': 'DUPE'}] + db_input = build_db_input( + branch_paths, [], {}, _sql_meta_insert(), + {'EMP-MASTER': [{'name': 'EMP-ID', 'db_type': 'CHAR', 'size': 8}]}, + records=records, + insert_pk={'EMP-MASTER': ['EMP-ID']}, + ) + rows = db_input.get('EMP-MASTER', []) + assert rows and rows[0]['EMP-ID'] == 'Z9999999' + + +def test_insert_collision_uses_input_record_pk_not_hostvar(): + """碰撞行主键用输入记录主键（R02EMP-ID），而非 expected host_var 值。 + + generate_records 的 rec 中 DBV-EMP-ID 可能是 DB 预置值（W前缀）或 + expected 值，而运行时实际 INSERT 的是输入记录主键（R02EMP-ID）。 + 碰撞行必须与输入主键一致，运行时 INSERT 才会失败。 + """ + branch_paths = [ + ([('SQLCODE', '=', '-803', True)], {}), + ] + records = [{'R02EMP-ID': 'A0000001', 'DBV-EMP-ID': 'W0000001'}] + db_input = build_db_input( + branch_paths, [], {}, _sql_meta_insert(), + {'EMP-MASTER': [{'name': 'EMP-ID', 'db_type': 'CHAR', 'size': 8}]}, + records=records, + insert_pk={'EMP-MASTER': ['EMP-ID']}, + ) + rows = db_input.get('EMP-MASTER', []) + assert rows and rows[0]['EMP-ID'] == 'A0000001' + + +def _sql_meta_insert_salary_results(): + """KYU06UPD 的 INSERT 语句形态（复合主键 EMP-ID + YEAR-MONTH）。""" + return [{ + "type": "exec_sql_insert", + "table": "SALARY-RESULTS", + "columns": ["EMP-ID", "YEAR-MONTH", "GROSS-PAYMENT"], + "host_vars": ["DBV-EMP-ID", "DBV-YEAR-MONTH", "DBV-GROSS-PAYMENT"], + "sql_text": ("INSERT INTO SALARY-RESULTS (EMP-ID, YEAR-MONTH, GROSS-PAYMENT) " + "VALUES (:DBV-EMP-ID, :DBV-YEAR-MONTH, :DBV-GROSS-PAYMENT)"), + }] + + +def _declared_salary_results(): + return { + "SALARY-RESULTS": [ + {"name": "EMP-ID", "db_type": "CHAR", "size": 8}, + {"name": "YEAR-MONTH", "db_type": "CHAR", "size": 6}, + {"name": "GROSS-PAYMENT", "db_type": "DECIMAL", "size": 9}, + ], + } + + +def _kyu06_assignments(): + """KYU06UPD 的主键 MOVE 链： + DBV-EMP-ID ← R02EMP-ID（输入键） + DBV-YEAR-MONTH ← CNS-YEAR-MONTH-PARM ← WRK-YEAR-MONTH（ACCEPT + + IF = SPACES 默认 '202605'）""" + return { + 'DBV-EMP-ID': [{'type': 'move', 'source_vars': ['R02EMP-ID']}], + 'DBV-YEAR-MONTH': [{'type': 'move', 'source_vars': ['CNS-YEAR-MONTH-PARM']}], + 'CNS-YEAR-MONTH-PARM': [{'type': 'move', 'source_vars': ['WRK-YEAR-MONTH']}], + 'WRK-YEAR-MONTH': [{'type': 'accept', 'from': 'USER'}, + {'type': 'move_literal', 'literal': '202605'}], + } + + +def test_insert_collision_composite_pk_matches_runtime_keys(): + """复合主键碰撞行必须等于运行时 INSERT 键： + EMP-ID 用输入记录键（R02EMP-ID）；YEAR-MONTH 用 ACCEPT 后 + IF = SPACES 的默认字面量（'202605'），而非记录中的合成值（'L00007'）。 + + 运行时 INSERT 键是 (A0000007, '202605')，碰撞行若用合成 YEAR-MONTH + 则永不碰撞，-803 分支永远不可达。 + """ + branch_paths = [ + ([('SQLCODE', '=', '-803', True)], {}), + ] + records = [{ + 'R02EMP-ID': 'A0000007', + 'DBV-EMP-ID': 'M0000007', + 'DBV-YEAR-MONTH': 'N00007', + 'WRK-YEAR-MONTH': 'L00007', + }] + db_input = build_db_input( + branch_paths, [], _kyu06_assignments(), _sql_meta_insert_salary_results(), + _declared_salary_results(), + records=records, + insert_pk={'SALARY-RESULTS': ['EMP_ID', 'YEAR_MONTH'], + 'SALARY_RESULTS': ['EMP_ID', 'YEAR_MONTH']}, + ) + rows = db_input.get('SALARY-RESULTS', []) + assert len(rows) == 1 + assert rows[0]['EMP-ID'] == 'A0000007' + assert rows[0]['YEAR-MONTH'] == '202605' + + +def test_insert_collision_year_month_not_falling_back_to_emp_id(): + """无 MOVE 链/ACCEPT 默认时，YEAR-MONTH 碰撞列不应回退到 EMP-ID 输入字段 + （旧 bug 会取 R02EMP-ID 截断成 'A00000'），而应使用宿主变量自身记录值。""" + branch_paths = [ + ([('SQLCODE', '=', '-803', True)], {}), + ] + records = [{'R02EMP-ID': 'A0000007', 'DBV-EMP-ID': 'M0000007', + 'DBV-YEAR-MONTH': 'N00007'}] + db_input = build_db_input( + branch_paths, [], {}, _sql_meta_insert_salary_results(), + _declared_salary_results(), + records=records, + insert_pk={'SALARY-RESULTS': ['EMP_ID', 'YEAR_MONTH'], + 'SALARY_RESULTS': ['EMP_ID', 'YEAR_MONTH']}, + ) + rows = db_input.get('SALARY-RESULTS', []) + assert len(rows) == 1 + assert rows[0]['YEAR-MONTH'] == 'N00007' diff --git a/tests/cobol_testgen/test_to_sql_schema_table.py b/tests/cobol_testgen/test_to_sql_schema_table.py new file mode 100644 index 0000000..4c9e1d6 --- /dev/null +++ b/tests/cobol_testgen/test_to_sql_schema_table.py @@ -0,0 +1,135 @@ +r"""DB2 schema 限定表名（SCHEMA.TABLE）在数据生成层的归一化测试。 + +根因：`FROM SALARYDB.EMP-MASTER` 中 `SALARYDB.` 限定符使 core.py 表名正则 +`(\w[\w-]*)` 遇 `.` 即停，把种子表解析为 `SALARYDB`；build_db_input 把 24 条 +EMP 种子行挂到不存在的 `SALARYDB` 表键下，插入时被丢弃 → EMP_MASTER 未播种， +运行时 EMP 正常路径（SEARCH/EVALUATE）无法覆盖。 + +修复（分层，通用）： +- core.py `_parse_sql` 各表名正则捕获完整限定名（SCHEMA.TABLE 不丢失）。 +- to_sql.py `collect_sql_meta` 用 `_norm_table` 剥掉限定符取末段（EMP-MASTER）， + 与 YAML schema 表名一致；种子键归一，`_populate_database` 正常插入。 +""" + +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +from cobol_testgen.core import extract_sql_assignments +from cobol_testgen.to_sql import collect_sql_meta, build_db_input + +_SRC_DECLARE = ( + " PROCEDURE DIVISION.\n" + " EXEC SQL\n" + " DECLARE C2 CURSOR FOR\n" + " SELECT EMP-ID, EMP-NAME, DEPT-CODE, BASE-SALARY\n" + " FROM SALARYDB.EMP-MASTER\n" + " ORDER BY EMP-ID\n" + " END-EXEC.\n" +) + + +def test_extract_sql_assignments_keeps_full_qualified_table(): + """core.py 表名正则须捕获完整限定名（不丢失 .EMP-MASTER）。""" + assignments = extract_sql_assignments(_SRC_DECLARE) + tables = [a.get("table") for a in assignments.values() for a in a] + assert "SALARYDB.EMP-MASTER" in tables, f"qualified table lost: {tables}" + + +def test_collect_sql_meta_normalizes_schema_qualified_table(): + """collect_sql_meta 把 SCHEMA.TABLE 归一化为末段表名。""" + assignments = { + "__SQL_CURSOR_C2": [{ + "type": "exec_sql_select", + "table": "SALARYDB.EMP-MASTER", + "select_list": "EMP-ID, EMP-NAME, DEPT-CODE, BASE-SALARY", + "into_vars": ["DBV-EMP-ID", "DBV-EMP-NAME", "DBV-DEPT-CODE", "DBV-BASE-SALARY"], + "where": "", + "sql_text": "DECLARE C2 CURSOR FOR SELECT EMP-ID FROM SALARYDB.EMP-MASTER", + }], + } + declared = {"EMP-MASTER": [ + {"name": "EMP-ID", "db_type": "CHAR", "size": 8}, + {"name": "EMP-NAME", "db_type": "CHAR", "size": 40}, + {"name": "DEPT-CODE", "db_type": "CHAR", "size": 2}, + {"name": "BASE-SALARY", "db_type": "CHAR", "size": 9}, + ]} + sql_meta = collect_sql_meta(assignments, declared) + assert len(sql_meta) == 1 + assert sql_meta[0]["table"] == "EMP-MASTER", sql_meta[0]["table"] + + +def test_build_db_input_seeds_emp_master_not_salarydb(): + """种子行键必须是 EMP-MASTER（而非 SALARYDB），否则插入时被丢弃。""" + branch_paths = [ + ([('SQLCODE', '=', '0', True)], {}), + ] + records = [{ + "DBV-EMP-ID": "A0000001", + "DBV-EMP-NAME": "TANAKA", + "DBV-DEPT-CODE": "01", + "DBV-BASE-SALARY": "429010", + }] + declared = {"EMP-MASTER": [ + {"name": "EMP-ID", "db_type": "CHAR", "size": 8}, + {"name": "EMP-NAME", "db_type": "CHAR", "size": 40}, + {"name": "DEPT-CODE", "db_type": "CHAR", "size": 2}, + {"name": "BASE-SALARY", "db_type": "CHAR", "size": 9}, + ]} + assignments = { + "__SQL_CURSOR_C2": [{ + "type": "exec_sql_select", + "table": "SALARYDB.EMP-MASTER", + "select_list": "EMP-ID, EMP-NAME, DEPT-CODE, BASE-SALARY", + "into_vars": ["DBV-EMP-ID", "DBV-EMP-NAME", "DBV-DEPT-CODE", "DBV-BASE-SALARY"], + "where": "", + "sql_text": "DECLARE C2 CURSOR FOR SELECT EMP-ID FROM SALARYDB.EMP-MASTER", + }], + } + db_input = build_db_input( + branch_paths, [], assignments, + collect_sql_meta(assignments, declared), + declared, + records=records, + ) + assert "SALARYDB" not in db_input, f"wrong table key: {list(db_input.keys())}" + assert "EMP-MASTER" in db_input, f"missing EMP-MASTER key: {list(db_input.keys())}" + + +def test_select_where_accept_fed_hostvar_uses_runtime_default(): + """WHERE 宿主变量由 ACCEPT FROM COMMAND-LINE 馈入时，种子必须用运行时 + 缺省值（如 SHA02MNC RATES 的 WRK-YEAR-MONTH='202605'），而非记录里的 + 合成值（'I00001'）。否则 EFFECTIVE-FROM/TO 不满足 WHERE，RATES 不装载， + SEARCH/EVALUATE 无法覆盖。""" + branch_paths = [ + ([('SQLCODE', '=', '0', True)], {}), + ] + records = [{"WRK-YEAR-MONTH": "I00001"}] + assignments = { + "WRK-YEAR-MONTH": [ + {"type": "accept", "from": "COMMAND-LINE"}, + {"type": "move_literal", "literal": "202605"}, + ], + } + sql_meta = [{ + "type": "exec_sql_select", + "table": "INSURANCE-RATES", + "select_list": "EFFECTIVE-FROM, EFFECTIVE-TO", + "into_vars": [], + "where": "EFFECTIVE-FROM <= :WRK-YEAR-MONTH AND EFFECTIVE-TO >= :WRK-YEAR-MONTH", + "where_constraints": [ + {"col": "EFFECTIVE-FROM", "type": "host_var", "host_var": "WRK-YEAR-MONTH", "op": "<="}, + {"col": "EFFECTIVE-TO", "type": "host_var", "host_var": "WRK-YEAR-MONTH", "op": ">="}, + ], + }] + declared = {"INSURANCE-RATES": [ + {"name": "EFFECTIVE-FROM", "db_type": "CHAR", "size": 6}, + {"name": "EFFECTIVE-TO", "db_type": "CHAR", "size": 6}, + ]} + db_input = build_db_input( + branch_paths, [], assignments, sql_meta, declared, records=records, + ) + rows = db_input.get("INSURANCE-RATES", []) + assert rows, "no INSURANCE-RATES seed rows" + assert rows[0]["EFFECTIVE-FROM"].strip() == "202605", ( + f"seed EFFECTIVE-FROM={rows[0]['EFFECTIVE-FROM']!r} not runtime default" + ) + assert rows[0]["EFFECTIVE-TO"].strip() == "202605" diff --git a/tests/cobol_testgen/test_to_sql_select_hostvar.py b/tests/cobol_testgen/test_to_sql_select_hostvar.py new file mode 100644 index 0000000..850c842 --- /dev/null +++ b/tests/cobol_testgen/test_to_sql_select_hostvar.py @@ -0,0 +1,279 @@ +"""SELECT 的 WHERE 宿主变量经 MOVE 链解析到输入记录键的测试。 + +根因：`MOVE R02EMP-ID TO DBV-EMP-ID` 在程序里先执行，运行时 SQL 查询键是 +输入记录键（R02EMP-ID）。但 build_db_input 直接读记录中的 DBV-EMP-ID +（生成器的独立合成值，如 J 前缀），与输入键不一致，导致 DB 预置行查不到、 +正常路径（SQLCODE=0）永远无法覆盖。修复：WHERE 宿主变量若有 MOVE 链， +应优先使用输入记录键。 +""" + +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +from cobol_testgen.to_sql import build_db_input + + +def _sql_meta_select(table="EMP-MASTER"): + return [{ + "type": "exec_sql_select", + "table": table, + "select_list": "EMP_NAME", + "into_vars": ["DBV-EMPLOYEE-NAME"], + "where": "EMP_ID = :DBV-EMP-ID", + "where_constraints": [ + {"col": "EMP-ID", "type": "host_var", "host_var": "DBV-EMP-ID", "op": "="} + ], + }] + + +def _declared_columns(): + return { + "EMP-MASTER": [ + {"name": "EMP-ID", "db_type": "CHAR", "size": 8}, + {"name": "EMP_NAME", "db_type": "CHAR", "size": 40}, + ], + } + + +def test_select_where_hostvar_moved_from_input_uses_input_key(): + """WHERE 宿主变量 DBV-EMP-ID 由 MOVE R02EMP-ID 赋值时， + DB 预置行主键必须用输入记录键（R02EMP-ID），而非 DBV-EMP-ID 的合成值。""" + branch_paths = [ + ([('SQLCODE', '=', '0', True)], {}), # SQL 成功路径 + ] + records = [{'R02EMP-ID': 'A0000001', 'DBV-EMP-ID': 'J0000001'}] + assignments = { + 'DBV-EMP-ID': [{'type': 'move', 'source_vars': ['R02EMP-ID']}], + } + db_input = build_db_input( + branch_paths, [], assignments, _sql_meta_select(), + _declared_columns(), + records=records, + ) + rows = db_input.get('EMP-MASTER', []) + assert len(rows) == 1 + assert rows[0]['EMP-ID'].strip() == 'A0000001' + + +def test_select_where_hostvar_no_move_keeps_record_value(): + """WHERE 宿主变量没有 MOVE 链时，维持原行为（用记录中宿主变量的值）。""" + branch_paths = [ + ([('SQLCODE', '=', '0', True)], {}), + ] + records = [{'R02EMP-ID': 'A0000001', 'DBV-EMP-ID': 'J0000001'}] + db_input = build_db_input( + branch_paths, [], {}, _sql_meta_select(), + _declared_columns(), + records=records, + ) + rows = db_input.get('EMP-MASTER', []) + assert len(rows) == 1 + assert rows[0]['EMP-ID'].strip() == 'J0000001' + + +def test_select_where_hostvar_chain_through_two_moves(): + """MOVE 链可跨多跳解析（A <- B <- C），最终用输入记录键。""" + branch_paths = [ + ([('SQLCODE', '=', '0', True)], {}), + ] + records = [{'R02EMP-ID': 'A0000001', 'WK-EMP-ID': 'M0000001', 'DBV-EMP-ID': 'J0000001'}] + assignments = { + 'WK-EMP-ID': [{'type': 'move', 'source_vars': ['R02EMP-ID']}], + 'DBV-EMP-ID': [{'type': 'move', 'source_vars': ['WK-EMP-ID']}], + } + db_input = build_db_input( + branch_paths, [], assignments, _sql_meta_select(), + _declared_columns(), + records=records, + ) + rows = db_input.get('EMP-MASTER', []) + assert len(rows) == 1 + assert rows[0]['EMP-ID'].strip() == 'A0000001' + + +def test_select_where_hostvar_skip_sql_error_path(): + """SQL 失败路径（SQLCODE<>0）仍不生成预置行。""" + branch_paths = [ + ([('SQLCODE', '<>', '0', True)], {}), + ] + records = [{'R02EMP-ID': 'A0000001', 'DBV-EMP-ID': 'J0000001'}] + assignments = { + 'DBV-EMP-ID': [{'type': 'move', 'source_vars': ['R02EMP-ID']}], + } + db_input = build_db_input( + branch_paths, [], assignments, _sql_meta_select(), + _declared_columns(), + records=records, + ) + rows = db_input.get('EMP-MASTER', []) + assert not rows + + +def test_select_uses_declared_columns_with_underscore_key(): + """declared_columns 以下划线键（EMP_MASTER）声明时，SELECT 行生成 + 也要命中，使用声明的 CHAR(8) 列宽，避免回退到 10 字符推断导致 + 存储值带尾随空格、与运行时 8 字符键失配。""" + branch_paths = [ + ([('SQLCODE', '=', '0', True)], {}), + ] + records = [{'R02EMP-ID': 'A0000001', 'DBV-EMP-ID': 'A0000001'}] + assignments = { + 'DBV-EMP-ID': [{'type': 'move', 'source_vars': ['R02EMP-ID']}], + } + declared = { + "EMP_MASTER": [ + {"name": "EMP_ID", "db_type": "CHAR", "size": 8}, + {"name": "EMP_NAME", "db_type": "CHAR", "size": 40}, + ], + } + db_input = build_db_input( + branch_paths, [], assignments, _sql_meta_select(), + declared, + records=records, + ) + rows = db_input.get('EMP-MASTER', []) + assert len(rows) == 1 + assert rows[0]['EMP_ID'] == 'A0000001' + assert len(rows[0]['EMP_ID']) == 8 + + +def test_update_where_hostvar_moved_from_input_uses_input_key(): + """UPDATE/DELETE 的 WHERE 宿主变量同样经 MOVE 链解析到输入记录键。""" + sql_meta = [{ + "type": "exec_sql_update", + "table": "EMP-MASTER", + "columns": ["STATUS"], + "host_vars": ["DBV-STATUS"], + "where": "EMP_ID = :DBV-EMP-ID", + "where_constraints": [ + {"col": "EMP-ID", "type": "host_var", "host_var": "DBV-EMP-ID", "op": "="} + ], + }] + branch_paths = [ + ([('SQLCODE', '=', '0', True)], {}), + ] + records = [{'R02EMP-ID': 'A0000001', 'DBV-EMP-ID': 'J0000001', 'DBV-STATUS': 'M'}] + assignments = { + 'DBV-EMP-ID': [{'type': 'move', 'source_vars': ['R02EMP-ID']}], + } + db_input = build_db_input( + branch_paths, [], assignments, sql_meta, + _declared_columns(), + records=records, + ) + rows = db_input.get('EMP-MASTER', []) + assert len(rows) == 1 + assert rows[0]['EMP-ID'].strip() == 'A0000001' + + +def _sql_meta_two_selects(): + """两个 SELECT：EMP-MASTER 在前（运行时先查），OVT-MONTHLY 在后。""" + return [ + { + "type": "exec_sql_select", "table": "EMP-MASTER", + "select_list": "EMP_NAME", "into_vars": ["DBV-EMPLOYEE-NAME"], + "where": "EMP_ID = :DBV-EMP-ID", + "where_constraints": [ + {"col": "EMP-ID", "type": "host_var", "host_var": "DBV-EMP-ID", "op": "="} + ], + "pos": 100, + }, + { + "type": "exec_sql_select", "table": "OVT-MONTHLY", + "select_list": "SUM(OVT_HOURS)", "into_vars": ["DBV-OVT-HOURS"], + "where": "EMP_ID = :DBV-EMP-ID AND YEAR_MONTH = :WRK-YEAR-MONTH", + "where_constraints": [ + {"col": "EMP-ID", "type": "host_var", "host_var": "DBV-EMP-ID", "op": "="}, + {"col": "YEAR-MONTH", "type": "host_var", "host_var": "WRK-YEAR-MONTH", "op": "="}, + ], + "pos": 200, + }, + ] + + +def _declared_two_tables(): + return { + "EMP-MASTER": [ + {"name": "EMP-ID", "db_type": "CHAR", "size": 8}, + {"name": "EMP_NAME", "db_type": "CHAR", "size": 40}, + ], + "OVT-MONTHLY": [ + {"name": "EMP-ID", "db_type": "CHAR", "size": 8}, + {"name": "YEAR-MONTH", "db_type": "CHAR", "size": 6}, + ], + } + + +def test_multiple_selects_drop_last_table_for_last_ok_path(): + """两个 SELECT 时，为最后一条 SQL 成功路径丢弃最后一张表（OVT）的行， + 使该记录到达下游查询时无匹配行 → 覆盖下游 SELECT 的 no-data 分支 + （IF SQLCODE = 0 的 ELSE 分支）。""" + branch_paths = [ + ([('SQLCODE', '=', '0', True)], {}), # path0 sql ok + ([('SQLCODE', '=', '0', True)], {}), # path1 sql ok (last) + ] + records = [ + {'R02EMP-ID': 'A0000001', 'DBV-EMP-ID': 'A0000001', + 'WRK-YEAR-MONTH': 'B00001', 'DBV-OVT-HOURS': '00101'}, + {'R02EMP-ID': 'A0000002', 'DBV-EMP-ID': 'A0000002', + 'WRK-YEAR-MONTH': 'B00002', 'DBV-OVT-HOURS': '00102'}, + ] + assignments = { + 'DBV-EMP-ID': [{'type': 'move', 'source_vars': ['R02EMP-ID']}], + } + db_input = build_db_input( + branch_paths, [], assignments, _sql_meta_two_selects(), + _declared_two_tables(), + records=records, + ) + emp = db_input.get('EMP-MASTER', []) + ovt = db_input.get('OVT-MONTHLY', []) + assert len(emp) == 2, emp # 两条路径都种 EMP（前置查询必须命中） + assert len(ovt) == 1, ovt # 最后一条路径的 OVT 行被丢弃 + assert emp[0]['EMP-ID'].strip() == 'A0000001' + assert ovt[0]['EMP-ID'].strip() == 'A0000001' + + +def test_single_select_keeps_all_rows(): + """只有一个 SELECT 时不触发丢弃（保持原行为）。""" + branch_paths = [ + ([('SQLCODE', '=', '0', True)], {}), + ([('SQLCODE', '=', '0', True)], {}), + ] + records = [ + {'R02EMP-ID': 'A0000001', 'DBV-EMP-ID': 'A0000001'}, + {'R02EMP-ID': 'A0000002', 'DBV-EMP-ID': 'A0000002'}, + ] + assignments = { + 'DBV-EMP-ID': [{'type': 'move', 'source_vars': ['R02EMP-ID']}], + } + db_input = build_db_input( + branch_paths, [], assignments, _sql_meta_select(), + _declared_columns(), + records=records, + ) + rows = db_input.get('EMP-MASTER', []) + assert len(rows) == 2, rows + + +def test_drop_requires_two_sql_ok_paths(): + """只有一条 SQL 成功路径时不丢弃最后一张表（避免该表行被清空）。""" + branch_paths = [ + ([('SQLCODE', '=', '0', True)], {}), # 唯一的 sql ok 路径 + ([('SQLCODE', '<>', '0', True)], {}), # 失败路径 + ] + records = [ + {'R02EMP-ID': 'A0000001', 'DBV-EMP-ID': 'A0000001', + 'WRK-YEAR-MONTH': 'B00001', 'DBV-OVT-HOURS': '00101'}, + {'R02EMP-ID': 'A0000002', 'DBV-EMP-ID': 'A0000002', + 'WRK-YEAR-MONTH': 'B00002', 'DBV-OVT-HOURS': '00102'}, + ] + assignments = { + 'DBV-EMP-ID': [{'type': 'move', 'source_vars': ['R02EMP-ID']}], + } + db_input = build_db_input( + branch_paths, [], assignments, _sql_meta_two_selects(), + _declared_two_tables(), + records=records, + ) + assert len(db_input.get('EMP-MASTER', [])) == 1 + assert len(db_input.get('OVT-MONTHLY', [])) == 1, db_input.get('OVT-MONTHLY') diff --git a/tests/config/test_program_schema_drop_tables.py b/tests/config/test_program_schema_drop_tables.py new file mode 100644 index 0000000..d3f39d0 --- /dev/null +++ b/tests/config/test_program_schema_drop_tables.py @@ -0,0 +1,37 @@ +"""Scenario `drop_tables` 配置解析测试。 + +用于覆盖「表缺失导致 OPEN CURSOR 失败」的错误分支（如 SHA02MNC #5 T： +EMP-OPEN SQLCODE≠0）。机制通用：任一程序可在场景里声明要 DROP 的表， +运行时 OPEN 该表即失败 → SQL 错误分支可达。 +""" + +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +from config.program_schema import ProgramSchema + +_YAML = """ +program_id: SHA02MNC +db_type: SQLite +db_name: INSURANCEDB.db +db_tables: + - name: EMP_MASTER + columns: + - name: EMP_ID + type: CHAR(8) + primary_key: true +runs: + - id: normal + - id: emp_open_fail + drop_tables: [EMP_MASTER] + inject_duplicate_pk: false +""" + + +def test_drop_tables_parsed_from_yaml(tmp_path): + p = tmp_path / "SHA02MNC.yaml" + p.write_text(_YAML, encoding="utf-8") + schema = ProgramSchema.from_yaml(p) + assert len(schema.runs) == 2 + fail = schema.runs[1] + assert fail.id == "emp_open_fail" + assert fail.drop_tables == ["EMP_MASTER"] diff --git a/tests/runners/test_gixsql_current_timestamp.py b/tests/runners/test_gixsql_current_timestamp.py new file mode 100644 index 0000000..4907f20 --- /dev/null +++ b/tests/runners/test_gixsql_current_timestamp.py @@ -0,0 +1,58 @@ +"""源级 CURRENT TIMESTAMP → CURRENT_TIMESTAMP 归一化测试（T4）。 + +gixpp 把长 SQL 折行，可能把 DB2 的 `CURRENT TIMESTAMP` 拆成 `CURRENT TIMES` / +`TAMP` 跨两行，导致 pp 级同行正则漏掉 → 生成的 UPDATE 保留空格版， +SQLite 报 `near "TIMESTAMP": syntax error`，UPDATE 成功分支（SQLCODE=0）不可达。 + +修复：在 gixpp 之前对源里的 EXEC SQL 块做归一化，使生成的 SQL 字面量拼接后 +为 SQLite 合法的 CURRENT_TIMESTAMP。仅作用于 EXEC SQL 块，不影响其它文本。 +""" + +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +from runners.gixsql_runner import _normalize_current_timestamp + + +def test_normalize_inside_exec_sql_block(): + src = (" EXEC SQL\n" + " UPDATE SALARY-RESULTS SET\n" + " UPDATED-AT = CURRENT TIMESTAMP\n" + " END-EXEC.") + out = _normalize_current_timestamp(src) + assert "CURRENT_TIMESTAMP" in out + assert "CURRENT TIMESTAMP" not in out + + +def test_normalize_multiple_exec_sql_blocks(): + src = (" EXEC SQL\n" + " INSERT INTO T1 (UPDATED-AT) VALUES (CURRENT TIMESTAMP)\n" + " END-EXEC.\n" + " EXEC SQL\n" + " UPDATE T2 SET UPDATED-AT = CURRENT TIMESTAMP\n" + " END-EXEC.") + out = _normalize_current_timestamp(src) + assert out.count("CURRENT_TIMESTAMP") == 2 + assert "CURRENT TIMESTAMP" not in out + + +def test_normalize_case_insensitive(): + src = " EXEC SQL\n SELECT CURRENT timestamp FROM T\n END-EXEC." + out = _normalize_current_timestamp(src) + assert "CURRENT_TIMESTAMP" in out + + +def test_normalize_idempotent_on_already_underscore(): + src = " EXEC SQL\n UPDATE T SET UPDATED-AT = CURRENT_TIMESTAMP\n END-EXEC." + out = _normalize_current_timestamp(src) + assert out == src + + +def test_normalize_does_not_touch_outside_exec_sql(): + src = (" MOVE 'CURRENT TIMESTAMP' TO WS-LIT\n" + " EXEC SQL\n" + " UPDATE T SET UPDATED-AT = CURRENT TIMESTAMP\n" + " END-EXEC.") + out = _normalize_current_timestamp(src) + # 块内被替换，块外（数据字符串）保持不变 + assert out.startswith(" MOVE 'CURRENT TIMESTAMP' TO WS-LIT\n") + assert "CURRENT_TIMESTAMP" in out diff --git a/tests/runners/test_gixsql_schema_qualifier.py b/tests/runners/test_gixsql_schema_qualifier.py new file mode 100644 index 0000000..9a4b1bc --- /dev/null +++ b/tests/runners/test_gixsql_schema_qualifier.py @@ -0,0 +1,99 @@ +"""EXEC SQL 内 DB2 schema 限定表名（SCHEMA.TABLE）剥离测试。 + +gixsql 会把 `FROM SALARYDB.EMP-MASTER` 原样归一化为 `FROM SALARYDB.EMP_MASTER` +注入 GIXSQL SQL 字符串；SQLite 无 schema 对象，OPEN CURSOR 报 +'no such table: SALARYDB.EMP_MASTER' → SQLCODE≠0 → EMP-OPEN 错误分支 → ABEND， +主循环 2000MAJSOR 永不执行（SHA02MNC 覆盖率 66.7% 根因）。 + +修复：gixpp 前把 EXEC SQL 块内 FROM/INTO/UPDATE/JOIN 后的 SCHEMA.TABLE +剥掉限定符（通用规则，处理任意 schema 名，无程序硬编码）。 +""" + +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +from runners.gixsql_runner import ( + _strip_schema_qualifiers, + _normalize_schema_qualifiers, +) + + +def test_from_schema_table_stripped(): + src = ( + " EXEC SQL\n" + " DECLARE C2 CURSOR FOR\n" + " SELECT EMP-ID, EMP-NAME\n" + " FROM SALARYDB.EMP-MASTER\n" + " ORDER BY EMP-ID\n" + " END-EXEC.\n" + ) + out = _strip_schema_qualifiers(src) + assert "FROM SALARYDB.EMP-MASTER" not in out + assert "FROM EMP-MASTER" in out + + +def test_from_three_part_qualified_stripped(): + src = "EXEC SQL SELECT A FROM DB2INST1.PAYROLL.TIMESHEET END-EXEC." + out = _strip_schema_qualifiers(src) + assert "FROM TIMESHEET" in out + assert "DB2INST1.PAYROLL" not in out + + +def test_update_schema_table_stripped(): + src = "EXEC SQL UPDATE SALARYDB.EMP-MASTER SET DEPT-CODE = '01' WHERE EMP-ID = :X END-EXEC." + out = _strip_schema_qualifiers(src) + assert "UPDATE EMP-MASTER" in out + + +def test_into_schema_table_stripped(): + src = "EXEC SQL INSERT INTO SALARYDB.EMP-MASTER (EMP-ID) VALUES (:X) END-EXEC." + out = _strip_schema_qualifiers(src) + assert "INSERT INTO EMP-MASTER" in out + + +def test_join_schema_table_stripped(): + src = ("EXEC SQL SELECT A.ID FROM A " + "JOIN SALARYDB.EMP-MASTER B ON A.ID = B.EMP-ID END-EXEC.") + out = _strip_schema_qualifiers(src) + assert "JOIN EMP-MASTER B" in out + + +def test_column_alias_qualifier_untouched(): + """列限定（E.EMP-ID / B.DEPT-CODE，非表名位置）不得剥离。""" + src = ("EXEC SQL SELECT E.EMP-ID FROM EMP-MASTER E " + "WHERE E.DEPT-CODE = '01' END-EXEC.") + out = _strip_schema_qualifiers(src) + assert "SELECT E.EMP-ID" in out + assert "E.DEPT-CODE" in out + assert "FROM EMP-MASTER E" in out + + +def test_hostvar_into_untouched(): + """SELECT ... INTO :DBV-X（宿主变量，非 SCHEMA.TABLE）不得剥离。""" + src = "EXEC SQL SELECT EMP-NAME INTO :DBV-EMP-NAME FROM EMP-MASTER END-EXEC." + out = _strip_schema_qualifiers(src) + assert "INTO :DBV-EMP-NAME" in out + + +def test_unqualified_table_untouched(): + src = "EXEC SQL SELECT EMP-ID FROM EMP-MASTER ORDER BY EMP-ID END-EXEC." + assert _strip_schema_qualifiers(src) == src + + +def test_wrapper_only_touches_exec_sql_blocks(): + text = " MOVE 'FROM A.B' TO WS-X.\n" + assert _normalize_schema_qualifiers(text) == text + + +def test_wrapper_strips_inside_exec_sql_only(): + text = ( + " MOVE 'X' TO WS-A.\n" + " EXEC SQL\n" + " SELECT EMP-ID\n" + " FROM SALARYDB.EMP-MASTER\n" + " END-EXEC.\n" + " MOVE 'FROM C.D' TO WS-B.\n" + ) + out = _normalize_schema_qualifiers(text) + assert "FROM SALARYDB.EMP-MASTER" not in out + assert "FROM EMP-MASTER" in out + assert "MOVE 'FROM C.D' TO WS-B" in out diff --git a/tests/runners/test_sqlcode_normalize.py b/tests/runners/test_sqlcode_normalize.py new file mode 100644 index 0000000..91b3293 --- /dev/null +++ b/tests/runners/test_sqlcode_normalize.py @@ -0,0 +1,66 @@ +"""SQLCODE 归一化注入测试（T3）。 + +gixsql+SQLite 对主键冲突返回 SQLCODE=-1555（SQLITE_CONSTRAINT_PRIMARYKEY）， +而 DB2 语义是 -803。在预处理后的 COBOL 中注入通用归一化代码，把 +SQLITE 约束错误码映射为 -803，使 `IF SQLCODE = -803` 分支可达。 +""" + +import sys, os +from pathlib import Path +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +from runners.gixsql_runner import GixsqlCobolRunner + + +def _make_runner(): + return GixsqlCobolRunner(gixpp_path="gixpp", lib_path="lib") + + +def _pp_text_with_if(): + return ( + " CALL \"GIXSQLEndSQL\"\n" + " END-CALL.\n" + " IF SQLCODE = 0\n" + " ADD 1 TO CUN-DB-INS\n" + " END-IF.\n" + " IF SQLCODE = -803\n" + " PERFORM 2000MAJSOR-UPD\n" + " END-IF.\n" + ) + + +def test_normalize_injects_mapping_after_endsql(tmp_path): + """GIXSQLEndSQL 后注入 SQLITE 约束错误 → -803 映射""" + r = _make_runner() + pp = tmp_path / "test_pp.cbl" + pp.write_text(_pp_text_with_if(), encoding="utf-8") + r._patch_sqlcode_normalize(pp) + out = pp.read_text(encoding="utf-8") + assert "SQLCODE = -1555" in out + assert "MOVE -803 TO SQLCODE" in out + assert "SQLCODE = -2067" in out + + +def test_normalize_mapping_before_if_sqlcode(): + """映射代码注入在 IF SQLCODE = -803 判断之前""" + r = _make_runner() + pp = Path(r'C:\Users\marye\AppData\Local\Temp\opencode\t3_pp.cbl') + pp.parent.mkdir(parents=True, exist_ok=True) + pp.write_text(_pp_text_with_if(), encoding="utf-8") + r._patch_sqlcode_normalize(pp) + out = pp.read_text(encoding="utf-8") + # 映射代码必须在第一个 IF SQLCODE 之前 + mapping_idx = out.find("MOVE -803 TO SQLCODE") + if_idx = out.find("IF SQLCODE = -803") + assert mapping_idx != -1 + assert if_idx != -1 + assert mapping_idx < if_idx + + +def test_normalize_does_not_change_ok_path(): + """归一化不影响 SQLCODE=0 成功路径""" + r = _make_runner() + pp = Path(r'C:\Users\marye\AppData\Local\Temp\opencode\t3b_pp.cbl') + pp.write_text(_pp_text_with_if(), encoding="utf-8") + r._patch_sqlcode_normalize(pp) + out = pp.read_text(encoding="utf-8") + assert "IF SQLCODE = 0" in out diff --git a/tests/test_kin08dbu_fixes.py b/tests/test_kin08dbu_fixes.py new file mode 100644 index 0000000..f912ee1 --- /dev/null +++ b/tests/test_kin08dbu_fixes.py @@ -0,0 +1,169 @@ +"""KIN08DBU DB 管道修复测试。 + +覆盖: +- Fix A: _scan_assign_to 引号 ASSIGN 捕获（阻断运行回归） +- Fix B: _inject_sql_error_rows 目标实际 INSERT 记录（9100DBERRSOR 可达） +- Fix D: 聚合边界数据注入（overflow / agg table full） +""" + +import os +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) + +import pytest + +from orchestrator_db import GixsqlOrchestrator + + +def _make_orch(src_text: str): + orch = object.__new__(GixsqlOrchestrator) + tmp = Path(tempfile.mkdtemp()) + orch.src_path = tmp / "prog.cbl" + orch.src_path.write_text(src_text, encoding="utf-8-sig") + orch.work_dir = tmp / "work" + return orch, tmp + + +KIN08_SRC = """ IDENTIFICATION DIVISION. + PROGRAM-ID. KIN08DBU. + ENVIRONMENT DIVISION. + INPUT-OUTPUT SECTION. + FILE-CONTROL. + SELECT R01INNFIL ASSIGN TO "KIN08R01". + SELECT SYSINFILE ASSIGN TO "KIN08S01". + SELECT W01OUTFIL ASSIGN TO "KIN08W01". + DATA DIVISION. + FILE SECTION. + FD R01INNFIL. + 01 R01INNREC. + PROCEDURE DIVISION. + OPEN INPUT R01INNFIL. + OPEN INPUT SYSINFILE. + OPEN OUTPUT W01OUTFIL. +""" + + +def test_scan_assign_to_does_not_capture_quote(): + """带引号 ASSIGN TO 应返回不带引号的文件名（Fix A 回归）。""" + orch, _ = _make_orch(KIN08_SRC) + assign_map = orch._scan_assign_to() + assert assign_map.get('KIN08S01') == 'INPUT', f'KIN08S01 应为 INPUT, 实际: {assign_map}' + assert assign_map.get('KIN08R01') == 'INPUT' + assert assign_map.get('KIN08W01') == 'OUTPUT' + # 禁止出现带前导引号的键 + assert not any(k.startswith('"') for k in assign_map), f'键不应含引号: {assign_map}' + + +def test_make_synthetic_error_rows_skips_blanked_first_record(): + """合成错误行应跳过 records[0]（其 EMP-ID 会被清空、运行时不会 INSERT）。""" + orch, _ = _make_orch(KIN08_SRC) + table = type('T', (), { + 'name': 'DAILY_RECORDS', + 'columns': [ + type('C', (), {'name': 'EMP_ID', 'primary_key': True}), + type('C', (), {'name': 'TARGET_DATE', 'primary_key': True}), + type('C', (), {'name': 'TIME_IN', 'primary_key': False}), + ], + })() + records = [ + {'R01EMP-ID': ' ' * 8, 'R01DATE': '20000101', 'R01TIME-IN': '0900', 'HV-EMP-ID': ' ' * 8, 'HV-TARGET-DATE': '20000101'}, + {'R01EMP-ID': '00000102', 'R01DATE': '20000102', 'R01TIME-IN': '0900', 'HV-EMP-ID': '00000102', 'HV-TARGET-DATE': '20000102'}, + {'R01EMP-ID': '00000103', 'R01DATE': '20000103', 'R01TIME-IN': '0900', 'HV-EMP-ID': '00000103', 'HV-TARGET-DATE': '20000103'}, + ] + rows = orch._make_synthetic_error_rows(table, records) + assert rows, '应生成合成错误行' + # 第一行不应是 records[0]（空 EMP-ID 记录） + assert rows[0][0].strip() != '', f'首行 EMP_ID 不应为空白: {rows[0]}' + assert rows[0][0] == '00000102', f'首行应为 records[1] 的 EMP_ID: {rows[0]}' + + +def test_inject_sql_error_rows_uses_per_row_values(): + """inject 循环内应逐行使用 row 值（而非固定 rows[0]）。""" + orch, tmp = _make_orch(KIN08_SRC) + table = type('T', (), { + 'name': 'DAILY_RECORDS', + 'columns': [ + type('C', (), {'name': 'EMP_ID', 'primary_key': True}), + type('C', (), {'name': 'TARGET_DATE', 'primary_key': True}), + type('C', (), {'name': 'TIME_IN', 'primary_key': False}), + ], + })() + orch.schema = type('S', (), {'db_tables': [table]})() + db = tmp / 'kin.db' + + import sqlite3 + conn = sqlite3.connect(str(db)) + conn.execute('CREATE TABLE [DAILY_RECORDS] ([EMP_ID] CHAR(8), [TARGET_DATE] CHAR(8), [TIME_IN] CHAR(4), PRIMARY KEY ([EMP_ID], [TARGET_DATE]))') + conn.commit() + conn.close() + + records = [ + {'R01EMP-ID': '00000101', 'R01DATE': '20000101', 'R01TIME-IN': '0900', 'HV-EMP-ID': '00000101', 'HV-TARGET-DATE': '20000101'}, + {'R01EMP-ID': '00000102', 'R01DATE': '20000102', 'R01TIME-IN': '0900', 'HV-EMP-ID': '00000102', 'HV-TARGET-DATE': '20000102'}, + {'R01EMP-ID': '00000103', 'R01DATE': '20000103', 'R01TIME-IN': '0900', 'HV-EMP-ID': '00000103', 'HV-TARGET-DATE': '20000103'}, + ] + orch._inject_sql_error_rows(db, records) + + conn = sqlite3.connect(str(db)) + rows = conn.execute('SELECT EMP_ID FROM [DAILY_RECORDS]').fetchall() + conn.close() + emp_ids = {r[0] for r in rows} + # 至少注入 1 条与 records[1] 匹配的 PK（运行时将被 INSERT → 冲突） + assert '00000102' in emp_ids, f'应注入 records[1] 的 PK, 实际: {emp_ids}' + + +def test_format_value_numeric_field_non_numeric_writes_spaces(): + """数字字段收到非数字值不应被静默转成 0（会造成 '00000000' PK 冲突），应写 SPACE。""" + from cobol_testgen.flatfile import _format_value + field = {'name': 'R01EMP-ID', 'type': 'numeric', 'length': 8, + 'pic': '9(008)', 'usage': None, + 'pic_info': {'type': 'numeric', 'digits': 8, 'decimal': 0, + 'length': 8, 'signed': False}} + # 非数字值 → SPACE（程序空社員チェックでスキップ） + assert _format_value('U0000031', field) == b' ' * 8, \ + f'非数字 EMP 应写 SPACE, 实际 {_format_value("U0000031", field)!r}' + # 数字值 → 正常 zfill + assert _format_value('102', field) == b'00000102' + # 全ゼロ → 保持（由 sanitize 负责 blank） + assert _format_value('00000000', field) == b'00000000' + + +def test_inject_aggregation_boundaries(): + """聚合边界注入：overflow 同键大小时数 + agg-full 超 100 组合（数字 EMP 字段）。""" + orch, tmp = _make_orch(KIN08_SRC) + recs = [] + for i in range(60): + recs.append({ + 'R01EMP-ID': f'00000{i + 100}', 'R01DATE': f'{20000101 + i:08d}', + 'R01TIME-IN': '0900', 'R01TIME-OUT': '1800', + 'R01ANNUAL-H': '00010', 'R01PERSONAL-H': '00000', + 'R01OFFICIAL-H': '00000', 'R01SICK-H': '00000', 'R01ABSENT-H': '00000', + 'R01FILLER': ' ' * 151, + }) + # 字母型 EMP（数字字段的非法值）不应被选为 dup_eid + recs.append({'R01EMP-ID': 'U0000031', 'R01DATE': '20001231', + 'R01TIME-IN': '0900', 'R01TIME-OUT': '1800', + 'R01ANNUAL-H': '00010', 'R01PERSONAL-H': '00000', + 'R01OFFICIAL-H': '00000', 'R01SICK-H': '00000', 'R01ABSENT-H': '00000', + 'R01FILLER': ' ' * 151}) + orch._inject_aggregation_boundaries(recs) + # overflow: 至少 2 条同 (EMP, 月) 且 ANNUAL-H 为最大值 + pairs = {} + for r in recs: + if r.get('R01ANNUAL-H'): + ym = r['R01DATE'][:6] + pairs.setdefault((r['R01EMP-ID'], ym), []).append(r['R01ANNUAL-H']) + max_seen = max('9' * len(v[0]) for v in pairs.values() if v) + over = [(k, v) for k, v in pairs.items() if len(v) >= 2 and all(h == max_seen for h in v)] + assert over, f'应存在同键 overflow 组, pairs={pairs}' + # 追加记录的 EMP 必须为数字（否则 9(008) 字段写文件会变 SPACE） + for r in recs[60:]: + assert str(r['R01EMP-ID']).isdigit(), f'追加记录 EMP 应为数字: {r["R01EMP-ID"]!r}' + # agg-full: 不同 (EMP, 月) 组合 >= 101 + distinct = set() + for r in recs: + distinct.add((r['R01EMP-ID'], r['R01DATE'][:6])) + assert len(distinct) >= 101, f'应有 >=101 个聚合键, 实际 {len(distinct)}' diff --git a/tests/test_kin09csv_fixes.py b/tests/test_kin09csv_fixes.py new file mode 100644 index 0000000..1038019 --- /dev/null +++ b/tests/test_kin09csv_fixes.py @@ -0,0 +1,97 @@ +"""KIN09CSV 修复测试。 + +覆盖: +- Fix A: PARM 场景配置（missing_mode / reordered_parm / missing_ym / empty_daily） +- Fix B: seed_extra_rows 大结果集注入（表头重出分支可达） +""" + +import os +import sqlite3 +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) + +import pytest + +from orchestrator_db import GixsqlOrchestrator + + +def _make_orch(): + orch = object.__new__(GixsqlOrchestrator) + table = type('T', (), { + 'name': 'DAILY_RECORDS', + 'columns': [ + type('C', (), {'name': 'EMP_ID', 'primary_key': True}), + type('C', (), {'name': 'TARGET_DATE', 'primary_key': True}), + type('C', (), {'name': 'TIME_IN', 'primary_key': False}), + type('C', (), {'name': 'UPDATED_AT', 'primary_key': False}), + ], + })() + orch.schema = type('S', (), {'db_tables': [table]})() + return orch + + +def _make_db(path: Path): + conn = sqlite3.connect(str(path)) + conn.execute('CREATE TABLE [DAILY_RECORDS] (' + '[EMP_ID] CHAR(8), [TARGET_DATE] CHAR(8), ' + '[TIME_IN] CHAR(4), [UPDATED_AT] TIMESTAMP, ' + 'PRIMARY KEY ([EMP_ID], [TARGET_DATE]))') + conn.execute("INSERT INTO [DAILY_RECORDS] VALUES ('Y0000001', '20260701', '0900', '2026-07-01')") + conn.commit() + conn.close() + + +def test_seed_extra_rows_injects_rows(): + """seed_extra_rows 应从 seed 行推导月份并注入唯一 (EMP, 日期) 行。""" + orch = _make_orch() + tmp = Path(tempfile.mkdtemp()) + db = tmp / 'kin.db' + _make_db(db) + scenario = type('S', (), {'seed_extra_rows': {'DAILY_RECORDS': 10}})() + orch._inject_extra_seed_rows(db, scenario) + + conn = sqlite3.connect(str(db)) + rows = conn.execute('SELECT EMP_ID, TARGET_DATE FROM [DAILY_RECORDS]').fetchall() + conn.close() + emps = {r[0] for r in rows} + assert len(rows) >= 11, f'应注入 10 条额外行, 实际 {len(rows)}' + assert 'SEED0001' in emps and 'SEED0010' in emps, f'应含 SEED 行, 实际 {emps}' + # 注入行日期应在查询月份内 + seed_dates = {r[1] for r in rows if r[0].startswith('SEED')} + assert seed_dates == {'20260701'}, f'注入日期应为 20260701, 实际 {seed_dates}' + + +def test_seed_extra_rows_noop_when_not_configured(): + """seed_extra_rows 未配置时不应注入。""" + orch = _make_orch() + tmp = Path(tempfile.mkdtemp()) + db = tmp / 'kin.db' + _make_db(db) + scenario = type('S', (), {'seed_extra_rows': {}})() + orch._inject_extra_seed_rows(db, scenario) + conn = sqlite3.connect(str(db)) + n = conn.execute('SELECT COUNT(*) FROM [DAILY_RECORDS]').fetchone()[0] + conn.close() + assert n == 1, f'不应注入额外行, 实际 {n}' + + +def test_kin09csv_scenarios_parse(): + """KIN09CSV.yaml 场景（含 seed_extra_rows）应正确解析。""" + from config.program_schema import load_schema + schema = load_schema('KIN09CSV', [str(Path(__file__).parent.parent / 'config' / 'programs')]) + by_id = {r.id: r for r in schema.runs} + assert 'missing_mode' in by_id, f'应含 missing_mode 场景: {list(by_id)}' + assert 'reordered_parm' in by_id + assert 'missing_ym' in by_id + assert 'empty_daily' in by_id + assert by_id['missing_mode'].command_line == 'YEARMONTH=202607' + assert by_id['reordered_parm'].command_line == 'MODE=FULL,YEARMONTH=202607' + assert by_id['missing_ym'].command_line == 'MODE=FULL' + assert by_id['empty_daily'].drop_tables == ['DAILY_RECORDS'], \ + f'empty_daily 应 drop DAILY_RECORDS, 实际 {by_id["empty_daily"].drop_tables}' + full = by_id['FULL'] + assert full.seed_extra_rows.get('DAILY_RECORDS') == 100, \ + f'FULL 场景 seed_extra_rows 应为 100, 实际 {full.seed_extra_rows}' diff --git a/tests/test_orchestrator_db_gcov_merge.py b/tests/test_orchestrator_db_gcov_merge.py new file mode 100644 index 0000000..79df104 --- /dev/null +++ b/tests/test_orchestrator_db_gcov_merge.py @@ -0,0 +1,65 @@ +r"""子程序 gcov 行号不得并入主程序 gcov 字典的回归测试。 + +Bug：`generate_coverage_report` 用 `gcov_data.update(sub_merged)` 把子程序 +(SUB*.cbl) 的 gcov 行计数并入主程序行号字典。主程序与子程序的行号都是纯 +整数，直接碰撞——SUB04CHK 的第 167 行（count=0）覆盖了 SHA02MNC 主程序 +第 167 行（count=25，主循环 PERFORM 已进入），导致 #10 Enter 分支被误判为 +未覆盖（SHA02MNC 覆盖率从应有的水平被拉低）。 + +修复：子程序 gcov 按子程序名独立保存（`_sub_gcov_data`），不并入主程序 +`gcov_data`。`_merge_run_dirs_gcov` 对单个程序跨 run 目录合并。 +""" + +import sys, os +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) +from pathlib import Path +from cobol_testgen.gcov import parse_cbl_gcov +from orchestrator_db import _merge_run_dirs_gcov + + +def _write_fake_gcov(dirpath: Path, name: str, counts: dict[int, int]): + lines = [] + for ln in sorted(counts): + cnt = counts[ln] + marker = "#####" if cnt == 0 else f"{cnt}*" + lines.append(f"{marker}:{ln}: dummy source line\n") + (dirpath / f"{name}.cbl.gcov").write_text("".join(lines), encoding="utf-8") + + +def _fake_gcov_func(name: str, d: str): + return parse_cbl_gcov(os.path.join(d, f"{name}.cbl.gcov")) + + +def test_sub_gcov_line_collision_does_not_corrupt_main(tmp_path): + run_normal = tmp_path / "run_normal"; run_normal.mkdir() + run_fail = tmp_path / "run_fail"; run_fail.mkdir() + + # 主程序：line 167 在 normal 场景执行 25 次（主循环进入） + _write_fake_gcov(run_normal, "MAIN", {167: 25, 265: 1}) + _write_fake_gcov(run_fail, "MAIN", {167: 0, 265: 1}) + # 子程序：line 167 从未执行（count=0）——与主程序行号碰撞 + _write_fake_gcov(run_normal, "SUB", {167: 0, 10: 1}) + _write_fake_gcov(run_fail, "SUB", {167: 0, 10: 1}) + + main = _merge_run_dirs_gcov(tmp_path, "MAIN", gcov_func=_fake_gcov_func) + sub = _merge_run_dirs_gcov(tmp_path, "SUB", gcov_func=_fake_gcov_func) + + # 合并后主程序 167 必须保持 25 + assert main[167] == 25, f"main line 167 corrupted: {main[167]}" + assert sub[167] == 0 + + # 旧 bug 行为演示：若把子程序并入主程序字典，167 会被覆盖为 0 + buggy = dict(main) + buggy.update(sub) + assert buggy[167] == 0, "demonstrates the corruption the fix avoids" + # 修复不变量：子程序数据独立保存，主程序字典完好 + assert main[167] == 25 + + +def test_merge_run_dirs_takes_max_count(tmp_path): + r1 = tmp_path / "run_a"; r1.mkdir() + r2 = tmp_path / "run_b"; r2.mkdir() + _write_fake_gcov(r1, "MAIN", {167: 25}) + _write_fake_gcov(r2, "MAIN", {167: 3}) + merged = _merge_run_dirs_gcov(tmp_path, "MAIN", gcov_func=_fake_gcov_func) + assert merged[167] == 25