test: cover between/hostvar/gcov-merge, class conditions, schema drop-tables, gixsql fixes

This commit is contained in:
hangshuo652
2026-08-09 17:43:16 +08:00
parent 273a3f8211
commit ff51bda962
22 changed files with 2559 additions and 0 deletions
+63
View File
@@ -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')
@@ -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}"
@@ -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 用小端 '<',但本 GnuCOBOLGC32-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'
@@ -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} 长休暇结束时间'
@@ -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
@@ -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 <FD> / INTO <GRP> 分行书写仍能解析为 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'
+108
View File
@@ -0,0 +1,108 @@
"""SEARCH 分支路径生成 + 二元 COMPUTE 链反演(Phase 3 — TDD RED
目标:KYU05DED SEARCH WHEN 分支(#9)从不可达变为可达,
配合二元 `-` COMPUTEWRK-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 源 `-` COMPUTESUBJ = GROSS - DEDDED 常量 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
@@ -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> ZEROTRUE 分支不可达。
修复:增加"后缀匹配"层级——一方 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}'
@@ -0,0 +1,38 @@
"""SQLCA 字节序对齐测试(T1)。
gixsql 以 native 小端 int 写入 SQLCA.SQLCODE。GnuCOBOL 的 COMP 按大端存储,
导致非零 SQLCODE 被错读(如 -1555 → -302383105)。必须改用 COMP-5native),
否则 `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-5native 字节序)"""
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
@@ -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'] == '>'
@@ -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
@@ -0,0 +1,126 @@
"""build_db_input 事务调度感知种子测试(JIN05UPD 根因)。
复现根因:UPDATE/DELETE 种子对每个 sql_ok 路径都用该路径记录的主键建行,
不区分记录运行时是否真的执行 UPDATE。JIN05UPD 的 4 条 INSERT 记录主键因此
全部被预置进 EMPLOYEE → 运行时 INSERT 全部 -803INSERT 成功分支不可达。
另外:非 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/B0000009OTHER 记录)绝不能入表。
"""
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}'
@@ -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=0INSERT 成功)的路径 → 不生成碰撞预置行"""
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<>0INSERT 失败)的路径 → 生成主键碰撞预置行"""
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-MONTHACCEPT +
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'
@@ -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 不满足 WHERERATES 不装载,
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"
@@ -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():
"""两个 SELECTEMP-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')
@@ -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"]
@@ -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
@@ -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
+66
View File
@@ -0,0 +1,66 @@
"""SQLCODE 归一化注入测试(T3)。
gixsql+SQLite 对主键冲突返回 SQLCODE=-1555SQLITE_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
+169
View File
@@ -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)}'
+97
View File
@@ -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}'
+65
View File
@@ -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