"""COBOL Test Data Generator — 模块化版入口 公开 API: extract_structure() — 解析 COBOL 控制流 → dict generate_data() — 生成测试数据 → list[dict] incremental_supplement — 差分补充数据 → list[dict] """ import os import sys import json import re import shutil import logging from datetime import datetime from pathlib import Path # ── 配置（必须放在本地模块导入之前，避免循环导入） ── CONFIG = { 'abend_programs': ['SUB03END'], } from .read import preprocess, extract_data_division, extract_procedure_division from .read import resolve_copybooks, parse_data_division, parse_file_section, scan_open_statements, scan_all_file_directions from .read import parse_file_control, resolve_sql_includes, strip_exec_sql_from_data_div from .core import build_branch_tree, classify_field_roles, _init_child_names, sql_register_virtual_fields, _find_multi_write_fds from .cond import parse_single_condition, is_field, collect_leaves from .pipeline_bridge import build_branch_tree_fallback from .design_mcdc import enum_paths as mcdc_enum_paths, _filter_stop from .design import enum_paths, generate_records, get_term_type, extend_abend_programs, make_base_record from .output import output_json, output_input_files from .coverage import run_coverage, generate_coverage_index, collect_decision_points, mark_coverage from japanese_data import generate_fullwidth_text, generate_halfwidth_katakana, generate_wareki_date try: from .runner import run_and_compare, run_all, GroupInfo, GroupResult _HAVE_RUNNER = True except ImportError: _HAVE_RUNNER = False try: from .gcov import run_gcov _HAVE_GCOV = True except ImportError: _HAVE_GCOV = False try: from .to_sql import collect_sql_meta, build_db_input _HAVE_TOSQL = True except ImportError: _HAVE_TOSQL = False logger = logging.getLogger(__name__) __all__ = [ "extract_structure", "generate_data", "incremental_supplement", "CONFIG", "generate_fullwidth_text", "generate_halfwidth_katakana", "generate_wareki_date", ] # ── OCCURS 展开 ── def _add_subscript(name, occ): """追加或扩展下标：WS-CELL → WS-CELL(1), WS-CELL(1) → WS-CELL(1,2)""" if name.endswith(')'): return name[:-1] + f',{occ})' return name + f'({occ})' def expand_occurs(fields): """展开 OCCURS 字段为下标副本。递归处理嵌套 OCCURS。""" result = [] i = 0 while i < len(fields): f = fields[i] if f.get('occurs', 0) > 0 and not f.get('is_88'): children = [] j = i + 1 while j < len(fields): child = fields[j] if child.get('is_88'): children.append(child) j += 1 continue if child['level'] <= f['level'] or child.get('level') == 77: break children.append(child) j += 1 if children: group = dict(f) group['occurs'] = 0 result.append(group) for occ in range(1, f['occurs'] + 1): for child in children: copy = dict(child) if child.get('occurs', 0) == 0: copy['occurs'] = 0 copy['occurs_depending'] = f.get('occurs_depending') if child.get('is_88'): parent = child.get('parent') or f['name'] copy['parent'] = _add_subscript(parent, occ) copy['name'] = _add_subscript(child['name'], occ) else: copy['name'] = _add_subscript(child['name'], occ) result.append(copy) else: for occ in range(1, f['occurs'] + 1): copy = dict(f) copy['name'] = _add_subscript(f['name'], occ) copy['occurs'] = 0 copy['occurs_depending'] = f.get('occurs_depending') result.append(copy) i = j else: result.append(f) i += 1 if any(f.get('occurs', 0) > 0 for f in result): return expand_occurs(result) return result # ── PREV 连锁 ── def _constraint_in(cons, field, op, value, want): for c in cons: if len(c) == 4 and c[0] == field and c[1] == op and c[2] == value and c[3] == want: return True return False def _inc_str(s, length): try: return str(int(s) + 1).zfill(length) except ValueError: c = list(str(s).ljust(length)[:length]) for i in range(len(c) - 1, -1, -1): if c[i] not in ' 9Zz\xff': c[i] = chr(ord(c[i]) + 1) break if c[i] == ' ': c[i] = '0' break if c[i] == '9': c[i] = '0' return ''.join(c) def _dec_str(s, length): try: n = max(0, int(s) - 1) return str(n).zfill(length) except ValueError: c = list(str(s).ljust(length)[:length]) for i in range(len(c) - 1, -1, -1): if c[i] not in ' 0Aa\x00': c[i] = chr(ord(c[i]) - 1) break if c[i] == ' ': break if c[i] == '0': c[i] = '9' return ''.join(c) def _field_length(fname, fields): for f in fields: if f['name'] == fname: pi = f.get('pic_info', {}) return pi.get('digits', 0) + pi.get('decimal', 0) or pi.get('length', 0) or 1 return 1 def _chain_prev(records, path_infos, fields, fd_fields, field_to_fd, open_dir): """跨记录 PREV 连锁。修改 records 使批次执行的路径与实际比较一致。 每个路径 k-1 的约束（PREV OP CURRENT）对应批次中 loop iter k-1 的实际比较： PREV = records[prev_src].R01 (程序内部保持的前值) CURRENT = records[k].R01 (当前读入值) 本函数调整 records[k] 的字段以保证交叉记录比较满足路径约束。 """ N = len(records) if N < 2: return key_fields = [] time_start_field = None time_end_field = None for fname in records[0]: if fname.startswith('R01') and not fname.startswith('R01INNREC'): base = fname[3:] prev_name = 'WRK-PREV-' + base if prev_name in records[0]: if 'EMP-ID' in fname or 'APPL-DATE' in fname: key_fields.append(fname) if 'END-TIME' in fname: time_end_field = fname if 'START-TIME' in fname: time_start_field = fname prev_src = 0 for k in range(1, N): if k - 1 >= len(path_infos): break cons = path_infos[k - 1][0] is_same_key = all( _constraint_in(cons, f'WRK-PREV-{fn[3:]}', '=', fn, True) for fn in key_fields ) if key_fields else False is_overlap = is_same_key and time_end_field and time_start_field and \ _constraint_in(cons, f'WRK-PREV-{time_end_field[3:]}', '>', time_start_field, True) is_normal = is_same_key and time_end_field and time_start_field and \ (_constraint_in(cons, f'WRK-PREV-{time_end_field[3:]}', '<=', time_start_field, True) or _constraint_in(cons, f'WRK-PREV-{time_end_field[3:]}', '>', time_start_field, False)) for fname in records[prev_src]: if fname.startswith('R01') and not fname.startswith('R01INNREC'): base = fname[3:] prev_name = 'WRK-PREV-' + base if prev_name in records[k]: records[k][prev_name] = records[prev_src][fname] if is_same_key: for kf in key_fields: if kf in records[k] and kf in records[prev_src]: records[k][kf] = records[prev_src][kf] if is_normal and time_end_field and time_start_field: prev_end = records[prev_src].get(time_end_field, '') curr_start = records[k].get(time_start_field, '') if prev_end >= curr_start: length = _field_length(time_start_field, fields) records[k][time_start_field] = _inc_str(prev_end, length) if is_overlap and time_end_field and time_start_field: prev_end = records[prev_src].get(time_end_field, '') curr_start = records[k].get(time_start_field, '') if prev_end <= curr_start: length = _field_length(time_start_field, fields) records[k][time_start_field] = _dec_str(prev_end, length) if prev_end else '0' * length else: for kf in key_fields: if kf in records[k] and kf in records[prev_src]: if records[k][kf] == records[prev_src][kf]: length = _field_length(kf, fields) records[k][kf] = _inc_str(str(records[k][kf]), length) records[k]['_w02_path'] = is_same_key and time_end_field and time_start_field and not is_overlap records[k]['_overlap_path'] = is_overlap for fn in list(records[k].keys()): if fn.startswith('R01') and not fn.startswith('R01INNREC'): wfn = 'W01' + fn[3:] if wfn in records[k]: records[k][wfn] = records[k][fn] if is_overlap: pass else: prev_src = k def _inject_empty_emp_rec(records, fields): """Insert a record with empty EMP-ID to trigger SPACE comparison paths.""" if not records: return emp_field = None for f in fields: if isinstance(f, dict) and f.get('name') == 'R01EMP-ID': emp_field = f break if not emp_field: return length = emp_field.get('pic_info', {}).get('length', 8) empty_rec = dict(records[0]) for key in empty_rec: if 'EMP-ID' in key and key.startswith('R01'): empty_rec[key] = ' ' * length records.insert(0, empty_rec) logger.info(f" injected empty-EMP-ID record at position 0") def _is_match_key_field(base: str) -> bool: """跨文件照合キー判定：EMP-ID / DATE に加え、末尾が -NO/-CODE のキー （OFFICE-NO / INSURER-CODE 等）も対象。非キー（NAME/KANA/FILLER/SEX/ TYPE/STATUS/PREF）は除外。共通ロジック、プログラム名ハードコードなし。""" u = base.upper() if any(kw in u for kw in ('NAME', 'KANA', 'FILLER', 'SEX', 'TYPE', 'STATUS', 'PREF', 'ADDR', 'DETAIL', 'REMARK')): return False if 'EMP-ID' in u or 'DATE' in u: return True if u.endswith('-NO') or u.endswith('-CODE') or u.endswith('-ID'): return True return False def _common_prefix(names): """计算一组字段名的最长公共前缀（限定以字母开头）。 用于从 FD 的实际字段名推导字段前缀。COPY REPLACING 程序（如 KIN07DAI 的 SWEMP-ID/SWDATE… → 'SW'）与字段名即 FD 前缀的程序（如 JIN01EMP-ID → 'JIN01'） 均适用。 """ cands = [n for n in names if n and n[0].isalpha()] if not cands: return '' base = cands[0] for n in cands[1:]: i = 0 while i < len(base) and i < len(n) and base[i] == n[i]: i += 1 base = base[:i] if not base: break return base def _fd_leaf_field_names(fd_name, file_sec, data_fields): """返回指定 FD 的叶子字段名列表（排除 88 级 / FILLER / 组项）。""" names = [] for rn in file_sec.get(fd_name, []): for child in _init_child_names(rn, data_fields): names.append(child) return names def _coordinate_multi_file_keys(records, kept_path_cons, data_fields, assignments, file_sec, term_types=None, open_dir=None): """协调跨文件匹配程序的キー值，为部分记录设置相同キー以触发照合路径。 数据驱动方式：扫描记录中的跨文件キー字段，强制部分记录キー一致。 通用实现：从 data_fields 中发现不同 FD 的キー类字段（EMP-ID/DATE）， 配对后修改记录値使キー一致。不依赖路径约束。 前缀推导：从 FD 的叶子字段名求公共前缀（SW/SR/SL 或 JIN01 等）， 而非从 01 记录名推导（R01INNREC→'R01'）。COPY REPLACING 场景下 记录名前缀与字段前缀不一致，旧逻辑导致键对匹配不到。 """ if not records or not file_sec or len(file_sec) < 2: return # Step 1: 自动发现 INPUT FD 的字段前缀和キー字段 # 保留 file_sec 顺序（源文件顺序）：第一个 INPUT FD 即主驱动（master） fd_prefixes = [] seen_prefix = set() for fd_name in file_sec: # 只保留 INPUT FD（open_dir 有方向信息时按方向过滤，避免把输出 FD # 的キー也纳入协调导致输出记录被误改） if open_dir: direction = str(open_dir.get(fd_name, open_dir.get(fd_name.upper(), ''))).upper() if direction not in ('INPUT', 'I-O'): continue prefix = _common_prefix(_fd_leaf_field_names(fd_name, file_sec, data_fields)) if prefix and prefix not in seen_prefix: seen_prefix.add(prefix) fd_prefixes.append(prefix) if len(fd_prefixes) < 2: return # 3+ 输入 FD：使用前置匹配序列（全匹配 + 各单文件匹配 + 判别字段注入） if len(fd_prefixes) > 2: _coordinate_tertiary_fd(records, fd_prefixes, data_fields, term_types) return # 按前缀分组找到キー字段（EMP-ID / DATE / 事務所・保険者コード類）。 # 复用共享实现 _find_key_pairs_inner（精确 / WORK- / EMP 族 / 后缀匹配）， # 避免两处维护同一逻辑。 key_pairs = _find_key_pairs_inner(fd_prefixes[0], fd_prefixes[1], data_fields) if not key_pairs: return # Step 2: 修改部分记录的キー值 modified = 0 max_modify = max(1, min(5, len(records) // 5)) modified_indices = [] for i, rec in enumerate(records): if modified >= max_modify: break # 确认双方都有非空値 has_both = True for fa, fb in key_pairs: va = str(rec.get(fa, '')).strip() vb = str(rec.get(fb, '')).strip() if not va or not vb: has_both = False break if not has_both: continue # 已有部分キー一致时跳过（避免破坏已有匹配） already_matched = all(str(rec.get(fa, '')).strip() == str(rec.get(fb, '')).strip() for fa, fb in key_pairs) if already_matched: continue # 将 prefix_a 的キー设为 prefix_b 的値 for fa, fb in key_pairs: rec[fa] = rec[fb] modified_indices.append(i) modified += 1 # Step 3: 创建重复 R02 キー，触发 2030PRIOSOR（マッチング選定の優先度判定） # 需要连续 2 条 R02 记录拥有相同キー，使 2020MATCHSOR 的内部循环进入 ELSE 分支 # 跳过 'abend' 记录（不会写入二进制文件），只考虑 'normal'/'eof' 记录 if len(modified_indices) >= 2 and term_types: normal_pairs = [] for idx in range(1, len(modified_indices)): i_prev = modified_indices[idx - 1] i_next = modified_indices[idx] t_prev = term_types[i_prev] if i_prev < len(term_types) else 'normal' t_next = term_types[i_next] if i_next < len(term_types) else 'normal' if t_prev not in ('abend',) and t_next not in ('abend',): normal_pairs.append((i_prev, i_next)) if normal_pairs: i_prev, i_next = normal_pairs[0] for _, fb in key_pairs: records[i_next][fb] = records[i_prev][fb] logger.info(f" 重复 R02 キー: 记录 {i_prev} → 记录 {i_next}（触发 2030PRIOSOR）") elif len(modified_indices) >= 2 and not term_types: # 无 term_types 时使用第一个 pair i_prev, i_next = modified_indices[0], modified_indices[1] for _, fb in key_pairs: records[i_next][fb] = records[i_prev][fb] logger.info(f" 重复 R02 キー: 记录 {i_prev} → 记录 {i_next}（触发 2030PRIOSOR，无 term_types）") if modified: logger.info(f" 跨文件键值协同: 将 {modified} 条记录的キー设为一致 (前缀 {fd_prefixes[0]}={fd_prefixes[1]})") else: # 回退：强制修改第一条有两方都非空キーの记录 for i, rec in enumerate(records): has_both = all(str(rec.get(fa, '')).strip() and str(rec.get(fb, '')).strip() for fa, fb in key_pairs) if has_both: for fa, fb in key_pairs: rec[fa] = rec[fb] logger.info(f" 跨文件键值协同(回退): 记录 {i} キー已设为一致") break # Step 4: 末条 normal 记录を表内键に整列（キーブレイク集約の最终组输出分支可达） # 最終社员工组は EOF 後の終了処理（`IF WRK-EMP-EVAL-CNT > ZERO ...`）で出力されるため、 # 末条记录が主表键と一致しないと最終组输出分支不可達。先頭のみ整列したままだと # 最終组に照合データが無い。首条 R02 键（必ず内部表ロード範囲内）を使用する。 if records and key_pairs: table_key = str(records[0].get(key_pairs[0][1], '')).strip() if table_key: for i in range(len(records) - 1, -1, -1): rec = records[i] if term_types and i < len(term_types) and term_types[i] == 'abend': continue if str(rec.get(key_pairs[0][1], '')).strip(): for fa, fb in key_pairs: rec[fa] = table_key logger.info(f" 最終キーブレイク组照合: 记录 {i} 明细键={table_key}") break def _coordinate_range_matching(records, data_fields, file_sec, open_dir, term_types=None): """金额-区间对齐：让部分记录的 *MONTHLY-AMOUNT 落入另一输入 FD 的 *MONTHLY-FROM/TO 区间。 适用：N:1 キーブレイク集約後、平均標準報酬月額を GRAD 区间でマッチング するプログラム（SHA05TWN 等）。生成データの金額と区間は独立 idx で合成 され数量级がずれるため、AVG が区间に命中せず GRADE 判定/出力経路が 不可達になる。本関数は部分记录の金額を区間中値に揃え、恰 3 条同 EMP-ID の记录も注入して MOVE 'B' 分支を覆う。 通用実装：フィールド名パターン（MONTHLY-AMOUNT / MONTHLY-FROM / MONTHLY-TO）＋入力 FD 判定（open_dir）で検出。プログラム名ハードコードなし。 """ if not records or not data_fields or not file_sec: return # 输入 FD 名 → 前缀 input_fd_names = [] if open_dir: for fd_name in file_sec: direction = str(open_dir.get(fd_name, open_dir.get(fd_name.upper(), ''))).upper() if direction in ('INPUT', 'I-O'): input_fd_names.append(fd_name) if not input_fd_names: return # 输入 FD 前缀集合（从 file_sec 的 FD 记录名提取前缀） input_prefixes = set() for fd_name in input_fd_names: for rn in file_sec.get(fd_name, []): m = re.match(r'^([A-Z]+\d*)', rn) if m: input_prefixes.add(m.group(1)) # 收集输入 FD 的字段（按前缀，从 data_fields 扫描实际字段名） amount_fields = [] # (fd_prefix, field_name) from_fields = [] to_fields = [] for f in data_fields: if not isinstance(f, dict) or not f.get('pic'): continue name = f['name'] if f.get('is_88'): continue prefix = None for p in input_prefixes: if name.startswith(p): prefix = p break if prefix is None: continue base = name[len(prefix):].lstrip('-') u = base.upper() if 'MONTHLY-AMOUNT' in u: amount_fields.append((prefix, name)) elif u.endswith('MONTHLY-FROM'): from_fields.append((prefix, name)) elif u.endswith('MONTHLY-TO'): to_fields.append((prefix, name)) if not amount_fields or not from_fields or not to_fields: return # 找跨 FD 前缀的 金额↔区间 配对（金额在 R01，区间在 R02） range_pair = None for af_pref, af in amount_fields: for ff_pref, ff in from_fields: for tf_pref, tf in to_fields: if af_pref != ff_pref and ff_pref == tf_pref: range_pair = (af_pref, af, ff_pref, ff, tf) break if range_pair: break if range_pair: break if not range_pair: return af_pref, af, ff_pref, ff, tf = range_pair logger.info(f" 金额-区间对齐: 金额 {af} ({af_pref}) ↔ 区间 {ff}/{tf} ({ff_pref})") # EMP-ID 字段（金额所在 FD 前缀） emp_field = None for f in data_fields: if not isinstance(f, dict) or not f.get('pic'): continue nm = f['name'] if nm.startswith(af_pref) and 'EMP-ID' in nm.upper(): emp_field = nm break if not emp_field: return # 从记录中提取有效区间（取第一条有值的 R02 区间） from_val = None to_val = None for rec in records: fv = str(rec.get(ff, '')).strip() tv = str(rec.get(tf, '')).strip() if fv and tv and fv.isdigit() and tv.isdigit(): from_val = int(fv) to_val = int(tv) break if from_val is None: return mid_val = (from_val + to_val) // 2 mid_str = str(mid_val).zfill(9) # 选定 normal 记录（非 abend） normal_indices = [i for i in range(len(records)) if not term_types or i >= len(term_types) or term_types[i] != 'abend'] if len(normal_indices) < 4: return # 1) 前 4 条记录：金额设为区间中值，前 4 条同 EMP-ID（→ COUNT>=4 → MOVE 'A'） hit_indices = normal_indices[:4] hit_emp = str(records[hit_indices[0]].get(emp_field, '')).strip() or 'A0000001' for i in hit_indices: rec = records[i] rec[af] = mid_str rec[emp_field] = hit_emp logger.info(f" 金额-区间对齐: {len(hit_indices)} 条记录金额→{mid_str}（区间命中，COUNT>=4 → TYPE='A'）") # 2) 构造恰 3 条同 EMP-ID（→ COUNT=3 → MOVE 'B'）：从 remaining 取 3 条 remaining = [i for i in normal_indices if i not in set(hit_indices)] if len(remaining) >= 3: b_indices = remaining[:3] b_emp = str(records[b_indices[0]].get(emp_field, '')).strip() or 'B0000002' for i in b_indices: rec = records[i] rec[af] = mid_str rec[emp_field] = b_emp logger.info(f" 金额-区间对齐: {len(b_indices)} 条记录同 EMP-ID={b_emp}（COUNT=3 → TYPE='B'）") # 3) 其余记录保留原值：区间外 / 记录数<3 → NO-GRADE / INVALID 错误分支维持 def _serialize_read_into_records(records, assignments, fields, file_sec): """不透明 FD + READ INTO：将求解出的 INTO 组子字段值序列化回 FD 记录字节。 问题：FD 记录不透明（如 R01INNREC PIC X(080) 无子字段）且程序用 `READ FD INTO WS-GRP` 读取时，Pass 3.5 中 fd_children 为空 → 求解出的 WRK-* 值从未写回 FD 记录字节 → 物理输入文件全是占位符、key 全同， 新 key 分支（flush + WRITE）不可达。 本函数： 1) 找到 READ INTO 赋值（fd 名, 目标组） 2) 按目标组子字段 PIC 长度计算布局偏移 3) 将求解出的子字段值按偏移 pack 回 FD 记录字节 4) 保证相邻记录 key 既有相同又有差异（同 key 累加 + 新 key flush 两路径） """ if not records or not assignments or not file_sec: return records name_to_field = {f['name']: f for f in fields} read_intos = [] for tgt, asgn_list in assignments.items(): for a in asgn_list: if isinstance(a, dict) and a.get('type') == 'read_into': read_intos.append((a.get('file', ''), tgt)) if not read_intos: return records for fname, tgt in read_intos: if fname not in file_sec or not file_sec[fname]: continue fd_rec = file_sec[fname][0] fd_field = name_to_field.get(fd_rec) if not fd_field: continue if _init_child_names(fd_rec, fields): continue pi = fd_field.get('pic_info', {}) fd_len = pi.get('length') or (pi.get('digits', 0) + pi.get('decimal', 0)) or 0 if fd_len <= 0: continue ws_children = _init_child_names(tgt, fields) layout = [] offset = 0 for c in ws_children: cf = name_to_field.get(c) if not cf: continue cpi = cf.get('pic_info', {}) if not cpi or cpi.get('type') == 'unknown': continue clen = cpi.get('length') or (cpi.get('digits', 0) + cpi.get('decimal', 0)) or 0 if clen <= 0: continue layout.append((c, offset, clen, cpi)) offset += clen if not layout: continue key_children = [] for c, off, clen, cpi in layout: base = c[4:] if c.startswith('WRK-') else c if c.startswith('WRK-') and ('WRK-PREV-' + base) in name_to_field: key_children.append((c, off, clen, cpi)) if not key_children: for c, off, clen, cpi in layout: if 'EMP-ID' in c or 'APPL-DATE' in c: key_children.append((c, off, clen, cpi)) if key_children and len(records) >= 2: kc, koff, klen, kpi = key_children[0] keys = [str(r.get(kc, '')).strip() 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)) if not has_diff: records[-1][kc] = _inc_str(str(records[-1].get(kc, '')), klen) logger.info(f" READ INTO {tgt}: 注入 key 差异（新 key flush 路径）") if not has_same: records[1][kc] = records[0][kc] logger.info(f" READ INTO {tgt}: 注入相同 key（累加路径）") for rec in records: buf = bytearray(b' ' * fd_len) for c, off, clen, cpi in layout: if off >= fd_len: continue val = str(rec.get(c, '')) if cpi.get('type') in ('numeric', 'numeric-edited'): seg = val.zfill(clen) else: seg = val.ljust(clen)[:clen] seg_bytes = seg.encode('utf-8', 'replace')[:clen] end = min(off + len(seg_bytes), fd_len) buf[off:end] = seg_bytes[:end - off] rec[fd_rec] = buf.decode('utf-8', 'replace') logger.info(f" READ INTO 序列化: {fd_rec} ← {tgt} ({len(layout)} 字段)") return records def _inject_merge_input_records(records, fields, file_sec, fd_fields, proc_div, term_types=None): """MERGE 程序：为 USING 输入文件注入可执行合并逻辑的 S/B 记录。 根因：MERGE ... USING 的输入文件从不 OPEN，原管道不识别为输入 → 无输入 数据 → 运行时 MERGE 读到空文件 → 2000MRGOUTSOR 全部不执行。 本函数： 1) 解析 MERGE 语句的 USING 文件列表（SALARY-FILE/BONUS-FILE） 2) 识别各 USING 文件 FD 的 REC-TYPE / EMP-ID / PAY-AMOUNT / GROSS 字段 3) 从 OUTPUT PROCEDURE 的 `IF = ` 解析判别常量值 （如 CNS-PAY-TYPE-SALARY='S'），其余 USING 文件取相异值 4) 注入一条含全部 USING 文件字段的记录（同 EMP-ID 跨文件匹配， 触发 2000MRGOUTSOR 同社員統合 + REC-TYPE T/F 分支） 通用实现：不依赖程序名/字段名硬编码，对所有 MERGE/SORT 程序适用。 """ if not records or not proc_div or not file_sec or len(fd_fields) < 2: return records _KEYWORDS = {'USING', 'KEY', 'ASCENDING', 'DESCENDING', 'ON', 'OUTPUT', 'INPUT', 'PROCEDURE'} def _resolve_const(name, fields_list): for f in fields_list: if f.get('name') == name and f.get('value') is not None: return str(f['value']).strip("'\"").strip() return None def _find_field(fd_name, suffix): for fname in fd_fields.get(fd_name, []): if fname.upper().endswith(suffix.upper()) and fname not in (fd_name,): return fname return None injected = False for m in re.finditer( r'\bMERGE\s+\w[\w-]*\s+.*?\bUSING\s+(.+?)\s*(?:OUTPUT|INPUT)?\s*PROCEDURE\s+\w[\w-]*\s*\.', proc_div, re.IGNORECASE | re.DOTALL ): stmt = m.group(0) files_raw = m.group(1) using = [f.upper() for f in re.findall(r'\b(\w[\w-]*)\b', files_raw) if f.upper() not in _KEYWORDS and f.upper() in file_sec] if len(using) < 2: continue out_sec = None om = re.search(r'\bOUTPUT\s+PROCEDURE\s+(\w[\w-]*)', stmt, re.IGNORECASE) if om: out_sec = om.group(1).upper() # 判别字段（REC-TYPE）与判别常量值 type_field = None const_val = None if out_sec: for f in fields: fn = f.get('name', '') if fn.upper().endswith('REC-TYPE'): type_field = fn break if type_field: cm = re.search( r'\bIF\s+%s\b\s*(?:=|NOT\s*=|<>|>|<|>=|<=)\s*(\w[\w-]*)' % re.escape(type_field), proc_div, re.IGNORECASE ) if not cm: cm = re.search( r'\b%s\s*(?:=|NOT\s*=|<>)\s*(\w[\w-]*)' % re.escape(type_field), proc_div, re.IGNORECASE ) if cm: const_val = _resolve_const(cm.group(1).upper(), fields) if const_val is None: const_val = 'S' # 通用回退：第一个 USING 文件用 'S' # 第二文件判别值：优先取 *PAY-TYPE-BONUS* 常量，否则取与 const_val 相异值 other_val = None for f in fields: fn = f.get('name', '') if 'PAY-TYPE' in fn.upper() and 'BONUS' in fn.upper() and f.get('value') is not None: other_val = str(f['value']).strip("'\"").strip() break if other_val is None: other_val = (chr(ord(const_val) + 1) if len(const_val) == 1 else 'B') if other_val == const_val: other_val = 'B' # 各 USING 文件的字段映射 emp_fields, pay_fields, gross_fields, type_fields = [], [], [], [] for fd in using: emp_fields.append(_find_field(fd, 'EMP-ID')) pay_fields.append(_find_field(fd, 'PAY-AMOUNT')) gross_fields.append(_find_field(fd, 'GROSS')) type_fields.append(_find_field(fd, 'REC-TYPE')) if any(f is None for f in emp_fields) or any(f is None for f in type_fields): continue # 每条记录：所有 USING 文件字段都设置，同 EMP-ID，REC-TYPE 按文件相异。 # 注：output_input_files 对次要 FD 会丢弃末尾 2 条 → 注入记录必须放最前。 base = dict(records[0]) if records else {} n = len(using) merge_recs = [] for emp_i in range(2): rec = dict(base) emp_id = f'A{emp_i + 1:07d}' for i, fd in enumerate(using): type_val = const_val if i == 0 else other_val rec[emp_fields[i]] = emp_id rec[type_fields[i]] = type_val if pay_fields[i]: rec[pay_fields[i]] = str(100000 + emp_i * 1000 + i * 100).zfill(9) if gross_fields[i]: rec[gross_fields[i]] = str(200000 + emp_i * 1000 + i * 100).zfill(9) merge_recs.append(rec) records[0:0] = merge_recs if term_types is not None: term_types[0:0] = ['normal', 'normal'] injected = True logger.info(f" 注入 MERGE 输入记录: USING={using} REC-TYPE={const_val}/{other_val} 共 2 条（同 EMP-ID 跨文件，置顶）") break return records def _inject_end_of_page_records(records, full_source, proc_div, file_sec, open_dir, term_types=None): """LINAGE 分页（WRITE ... AT END-OF-PAGE）プログラム：补充分页体积记录。 根因：V3 路径枚举只生成覆盖决策点的最小记录集。对 END-OF-PAGE（分页）分支， 需要足够多的明细记录触发 LINAGE 分页才能覆盖（例：JIN04PRT 输入仅 1 条 → AT END-OF-PAGE 换页重打分支不可达，gcov 仅 80/96）。 本函数（通用・程序名硬编码なし）： 1) PROCEDURE DIVISION 中检测 `WRITE ... AT END-OF-PAGE` 2) 从全量源码解析输出 FD 的 LINAGE 总行数 - TOP - BOTTOM 作为正文容量 （解析失败回退 60） 3) 将一条有效 normal 记录复制补足到「容量+1」条，使最后一条 WRITE 触发 END-OF-PAGE """ if not records or not proc_div or not full_source: return records if not re.search(r'\bEND-OF-PAGE\b', proc_div, re.IGNORECASE): return records # LINAGE 容量：LINAGE IS n LINES（TOP t / BOTTOM b）→ 正文行数 capacity = None m = re.search( r'LINAGE\s+IS\s+(\d+)\s+LINES([^.]*?)(?=\.)', full_source, re.IGNORECASE | re.DOTALL) if m: total = int(m.group(1)) rest = m.group(2) top_m = re.search(r'\bTOP\s+(\d+)', rest, re.IGNORECASE) bot_m = re.search(r'\bBOTTOM\s+(\d+)', rest, re.IGNORECASE) top = int(top_m.group(1)) if top_m else 0 bottom = int(bot_m.group(1)) if bot_m else 0 capacity = total - top - bottom if not capacity or capacity <= 0: capacity = 60 target = capacity + 1 def _is_normal(i): if term_types is None or i >= len(term_types): return True return term_types[i] != 'abend' template = None for i in range(len(records) - 1, -1, -1): if _is_normal(i) and records[i]: template = dict(records[i]) break if template is None: return records current = sum(1 for i in range(len(records)) if _is_normal(i)) if current >= target: return records need = target - current for _ in range(need): records.append(dict(template)) if term_types is not None: term_types.extend(['normal'] * need) logger.info( f" END-OF-PAGE 分页注入: 复制 {need} 条记录至 {len(records)} 条" f"（LINAGE 容量 {capacity}，触发分页）") return records def _inject_sort_limit_records(records, proc_div, fields, term_types=None): """SORT INPUT 计数守卫（RELEASE + IF <计数器> <= <上限>）：补充分页体积记录。 根因：V3 路径枚举只生成覆盖决策点的最小记录集。对 SORT INPUT PROCEDURE 中的 记录数上限守卫（如 `ADD 1 TO WRK-SORT-CNT ... IF WRK-SORT-CNT <= CNS-MAX-SORT-REC(999) RELEASE ... ELSE 上限エラー`），需超过上限条数的输入 记录才能覆盖 ELSE（SORT LIMIT OVER）分支（例：JIN06SRT 输入仅 6 条，上限 999）。 本函数（通用・程序名硬编码なし）： 1) PROC 中检测 SORT + RELEASE + 同一段落内 `IF (<=|<) ` 2) 解析 的常量值（字段 VALUE 子句或字面值） 3) 将一条 valid normal 记录复制到「上限+1」条，使最后一条 READ 触发上限分支 """ if not records or not proc_div: return records up = proc_div.upper() if 'SORT' not in up or 'RELEASE' not in up: return records # 计数守卫：IF (<=|<) （同段落内必须有 RELEASE） m = re.search(r'\bIF\s+(\w[\w-]*)\s*(?:<=|<)\s*(\w[\w-]*)\b', proc_div, re.IGNORECASE) if not m: return records after = proc_div[m.start():] para_end = re.search(r'\n\s*\w[\w-]*-EXT\.', after) scope = after[:para_end.start()] if para_end else after if 'RELEASE' not in scope.upper(): return records # 解析上限值：字段 VALUE 子句或字面值 limit_ref = m.group(2) limit = None raw = str(limit_ref).strip().strip("'\"").strip() if raw.isdigit(): limit = int(raw) else: for f in (fields or []): if f.get('name') == raw and f.get('value') is not None: try: limit = int(str(f['value']).strip("'\"").strip()) except ValueError: limit = None break if limit is None or limit <= 0: return records target = limit + 1 def _is_normal(i): if term_types is None or i >= len(term_types): return True return term_types[i] != 'abend' template = None for i in range(len(records) - 1, -1, -1): if _is_normal(i) and records[i]: template = dict(records[i]) break if template is None: return records current = sum(1 for i in range(len(records)) if _is_normal(i)) if current >= target: return records need = target - current for _ in range(need): records.append(dict(template)) if term_types is not None: term_types.extend(['normal'] * need) logger.info( f" SORT 上限体积注入: 复制 {need} 条记录至 {len(records)} 条" f"（上限 {limit}，触发 SORT LIMIT OVER）") return records def _provision_sub_input_files(main_source, source_dir, outdir): """子程序输入供给：为 CALL 目标子程序的 INPUT FD 生成输入文件。 根因：V3 只为主程序生成输入文件。テストドライバ（主程序无 FILE 节）调用 读取文件的自程序时（如 JIN08TST → JIN07SUB 读 JIN07R01），自程序 OPEN 输入文件失败（status 35）→ ABEND → 后续决策点不可达。 本函数（通用・程序名硬编码なし）： 1) 主程序 PROC 中 `CALL 'NAME'` 收集子程序名 2) 在 source_dir 的 sub/ 找到子程序源码 3) 解析其 FILE-CONTROL 的 INPUT/I-O FD（ASSIGN 名・记录布局） 4) 按子程序 FD 布局写一条基础记录到 outdir/ 返回已生成的 assign 名列表。 """ src_dir = Path(source_dir) sub_dir = src_dir.parent / 'sub' if not sub_dir.is_dir(): sub_dir = src_dir / 'sub' cpy_dir = src_dir.parent / 'cpy' if not cpy_dir.is_dir(): cpy_dir = src_dir / 'cpy' calls = set() for m in re.finditer(r"\bCALL\s+['\"](\w+)['\"]", main_source, re.IGNORECASE): calls.add(m.group(1).upper()) if not calls: return [] from .flatfile import analyze_fd_layout, write_flat_file written = [] for sub_name in sorted(calls): sp = sub_dir / f"{sub_name}.cbl" if not sp.exists(): continue sub_src = sp.read_text(encoding='utf-8') try: sub_src = resolve_copybooks( sub_src, str(sub_dir), extra_search_paths=[str(cpy_dir)] if cpy_dir.is_dir() else None) layouts = analyze_fd_layout( sub_src, copybook_dirs=[str(cpy_dir)] if cpy_dir.is_dir() else None) except Exception as e: logger.debug(f" 子程序 {sub_name} 布局解析失败: {e}") continue for assign, layout in layouts.items(): if layout.get('direction') not in ('INPUT', 'I-O'): continue base = {} for rec in layout.get('records', []): for f in rec.get('fields', []): pi = f.get('pic_info') or {} if pi.get('type') == 'numeric': ln = (pi.get('digits', 0) + pi.get('decimal', 0)) or 1 base[f['name']] = '0' * ln else: ln = f.get('length') or pi.get('length') or 1 base[f['name']] = 'A' * ln outpath = Path(outdir) / assign outpath.parent.mkdir(parents=True, exist_ok=True) try: write_flat_file([base], layout, outpath) except Exception as e: logger.debug(f" 子程序 {sub_name} 输入文件写入失败: {e}") continue rec_len = layout.get('records', [{}])[0].get('record_length', 0) written.append(assign) logger.info(f" 子程序输入供给: {sub_name} → {assign}（记录 {rec_len}B）") return written def _coordinate_tertiary_fd(records, fd_prefixes, data_fields, term_types=None): """构造多文件照合程序的前置匹配序列（通用，不依赖程序名）。 fd_prefixes = [主FD字段前缀, 明细FD字段前缀1, ...]（如 SW/SR/SL）。 在记录头部构造单调递增键序列，使 merge-join 指针自然对齐： - rec0: 主 + 全部明细键一致 → 全匹配（照合ループ + 判别 EVALUATE 全 WHEN） - rec(k): 主 + 第 k 个明细键一致、其余不同 → 单明细匹配（各照合组合） 判别字段（*LEAVE-TYPE）所在明细构造同键重复块： - 块1: 短时长 + 判别值 01-04 → 全匹配内循环消费（EVALUATE 全 WHEN） - 块2: 长时长（0600-2200）+ 判别值 01-04 → 单明细匹配内循环消费， 覆盖时长上限检查 T 分支与午餐重叠判定。 """ master = fd_prefixes[0] details = fd_prefixes[1:] if not details: return normal_indices = [i for i in range(len(records)) if not term_types or i >= len(term_types) or term_types[i] != 'abend'] dpairs = {} for d in details: pairs = _find_key_pairs_inner(master, d, data_fields) if pairs: dpairs[d] = pairs if not dpairs: return # 判别字段：取"最后一个"含 *LEAVE-TYPE 的明细 FD（深层休暇展开文件）。 # 注意 R02 打刻文件也可能含 LEAVE-TYPE 字段（如 SRLEAVE-TYPE），但 # 内循环 EVALUATE 判别的是休暇展开明细（R03），故从后向前选。 disc_detail = None disc_field = None for d in reversed(details): if d not in dpairs: continue for f in data_fields: if not isinstance(f, dict) or not f.get('pic'): continue nm = f.get('name', '') if nm.startswith(d) and 'LEAVE-TYPE' in nm.upper(): disc_detail, disc_field = d, nm break if disc_detail: break disc_values = ['01', '02', '03', '04', '99'] if disc_field else [] m = len(disc_values) if disc_field else 2 n = len(details) match_cnt = n + 1 need = max(match_cnt, m) + m if len(normal_indices) < max(need, 6): return def _synth(idx): # 合成键：EMP-ID 'A0000001' 起递增、日期 20000101 起递增（字典序单调递增） return f'A{idx + 1:07d}', f'{20000101 + idx:08d}' def _apply(rec, prefix, pairs, idx): emp, date = _synth(idx) for fa, fb in pairs: f = fa if prefix == master else fb val = date if 'DATE' in f.upper() else emp rec[f] = val p = normal_indices # rec0: 全匹配（主 + 全部明细键 = K0） rec = records[p[0]] for d, pairs in dpairs.items(): _apply(rec, master, pairs, 0) _apply(rec, d, pairs, 0) # rec1..n: 单明细匹配（主 + 该明细 = K(k+1)，其余明细用大偏移避免误配） for k, d in enumerate(details): if k + 1 >= len(p): continue rec = records[p[k + 1]] _apply(rec, master, dpairs[d], k + 1) _apply(rec, d, dpairs[d], k + 1) for d2, pairs2 in dpairs.items(): if d2 != d: _apply(rec, d2, pairs2, k + 1 + 50) if disc_detail and disc_field: pairs = dpairs[disc_detail] # 块1：位置 p[0..m-1]，键 = K0，短时长 + 判别值 for j in range(m): if j >= len(p): break rec = records[p[j]] _apply(rec, disc_detail, pairs, 0) rec[disc_field] = disc_values[j] # 块2：位置 p[max(match_cnt,m)..+m]，键 = K(disc_idx)，长时长 + 判别值 disc_idx = details.index(disc_detail) + 1 b2_start = max(match_cnt, m) st_f = en_f = None for f in data_fields: if not isinstance(f, dict) or not f.get('pic'): continue nm = f.get('name', '') if nm.startswith(disc_detail): u = nm.upper() if 'START-TIME' in u: st_f = nm elif 'END-TIME' in u: en_f = nm for j in range(m): idx = b2_start + j if idx >= len(p): break rec = records[p[idx]] _apply(rec, disc_detail, pairs, disc_idx) rec[disc_field] = disc_values[j] if st_f and en_f: rec[st_f] = '0600' rec[en_f] = '2200' logger.info( f" 前置匹配序列: 全匹配+{n}单明细匹配, {disc_detail} 判别块 {disc_values} " f"(长时长 0600-2200 覆盖上限/午餐判定)" ) else: # 无判别字段：仍制造同键重复（触发明细内循环） d0, pairs0 = next(iter(dpairs.items())) if len(p) >= 2: rec0, rec1 = records[p[0]], records[p[1]] _apply(rec0, master, pairs0, 0) _apply(rec0, d0, pairs0, 0) _apply(rec1, master, pairs0, 0) _apply(rec1, d0, pairs0, 0) logger.info(f" 前置匹配序列: 全匹配+{n}单明细匹配 ({d0} 同键重复)") def _find_key_pairs_inner(prefix_a, prefix_b, data_fields): """Find BEST matching key field pairs between two FD prefixes. Isolated helper for _coordinate_tertiary_fd. """ a_candidates = {} b_candidates = {} for f in data_fields: if not isinstance(f, dict) or not f.get('pic'): continue name = f['name'] if name.startswith(prefix_a): base = name[len(prefix_a):].lstrip('-') if _is_match_key_field(base): a_candidates[name] = base elif name.startswith(prefix_b): base = name[len(prefix_b):].lstrip('-') if _is_match_key_field(base): b_candidates[name] = base pairs = [] used_b = set() for fa, ba in sorted(a_candidates.items()): for fb, bb in sorted(b_candidates.items()): if fb in used_b: continue if ba == bb or ba.replace('WORK-', '') == bb or ba == bb.replace('WORK-', ''): pairs.append((fa, fb)) used_b.add(fb) break for fa, ba in sorted(a_candidates.items()): if any(p[0] == fa for p in pairs): continue ua = ba.upper() if 'EMP' not in ua: continue for fb, bb in sorted(b_candidates.items()): if fb in used_b: continue ub = bb.upper() if 'EMP' in ub: pairs.append((fa, fb)) used_b.add(fb) break # Suffix tier: 明细键 base 是主表键 base 的真后缀（如 SKILL-CODE ⊂ # MST-SKILL-CODE）且 key 类别一致（同为 -CODE/-ID/-NO/DATE/EMP-ID）时配对。 # 通用实现、无程序名硬编码，覆盖"主表键带限定词（MST- 等）"的照合形态。 def _same_key_class(x, y): ux, uy = x.upper(), y.upper() for cls in ('EMP-ID', '-CODE', '-ID', '-NO', 'DATE'): if cls in ux and cls in uy: return True return False for fa, ba in sorted(a_candidates.items()): if any(p[0] == fa for p in pairs): continue for fb, bb in sorted(b_candidates.items()): if fb in used_b: continue if (len(ba) >= 3 and len(bb) >= 3 and ba != bb and _same_key_class(ba, bb) and (bb.upper().endswith(ba.upper()) or ba.upper().endswith(bb.upper()))): pairs.append((fa, fb)) used_b.add(fb) break return pairs def _inject_leave_type_scenarios(records, term_types): """Inject R02LEAVE-TYPE / R02APPL-ID 覆盖 2030PRIOSOR 所有 WHEN + IF-ELSE 分支 + WHEN OTHER。 在 _coordinate_multi_file_keys 之后运行, 取 16 条连续 normal 记录 (1 anchor + 14 scenario + 1 WHEN OTHER), 设其 R01/R02 キー相同, 然后依次设 R02LEAVE-TYPE 为 '99'→'04'→'04'→'03'→'03'→'04'→'02'→'02'→ '03'→'04'→'01'→'01'→'02'→'03'→'04'→'99', 配合 APPL-ID 递减, 以单一 2020MATCHSOR 调用遍历 EVALUATE 内部所有路径。 """ if not records: return # 守卫：仅当记录中存在 LEAVE-TYPE 字段（2030PRIOSOR 系）时执行。 # 避免对无 LEAVE-TYPE 的 N:1 集約程序（如 SHA05TWN）产生副作用—— # 该函数原会无差别把前 16 条 normal 记录的 EMP-ID 设为同一值。 if not any('LEAVE-TYPE' in str(k).upper() for k in records[0]): return # 跳过 abend 记录, 找连续 16 条 normal (1 anchor + 14 scenario + 1 OTHER) needed = 16 normal_indices = [] for i in range(len(records)): if term_types and i < len(term_types) and term_types[i] == 'abend': continue normal_indices.append(i) if len(normal_indices) >= needed: break if len(normal_indices) < needed: logger.warning(f" 休暇種別覆盖: normal 记录不足 ({len(normal_indices)}/{needed}), 跳过") return anchor_idx = normal_indices[0] anchor = records[anchor_idx] common_emp = str(anchor.get('R02EMP-ID', '')).strip() or 'E0000999' common_date = str(anchor.get('R02DATE', '')).strip() or '20991231' # 所有记录设相同キー for idx in normal_indices: rec = records[idx] rec['R02EMP-ID'] = common_emp rec['R02DATE'] = common_date rec['R01EMP-ID'] = common_emp rec['R01WORK-DATE'] = common_date # 锚点: LT='99'(CNS-NO-LEAVE), 第一个 MATCH 使 BEST-TYPE=99 anchor['R02LEAVE-TYPE'] = '99' anchor['R02APPL-ID'] = '000000100' anchor['R02START-TIME'] = '0800' anchor['R02END-TIME'] = '1700' # 14 scenario + 1 OTHER 按顺序覆盖各 WHEN 分支 # 路径: BEST 从 99→04→03→02→01 逐级覆盖, 末尾 LT='99' 进 WHEN OTHER scenarios = [ ('04', '000000080', '0800', '1700'), # LE04 ELSE(384-387) ('04', '000000060', '0800', '1700'), # LE04 AID 23) or '0060' (minute=60 > 59) # For EMP-ID fields: SUB04CHK checks spaces + alpha/digit/special → '00000000' is valid field_invalid = { 'WRK-CSV-START-DATE': ' ', 'WRK-CSV-START-TIME': '2500', 'WRK-CSV-END-DATE': ' ', 'WRK-CSV-END-TIME': '2500', } field_sizes = { 'WRK-CSV-START-DATE': 8, 'WRK-CSV-START-TIME': 4, 'WRK-CSV-END-DATE': 8, 'WRK-CSV-END-TIME': 4, } targets = [ (9, 'WRK-CSV-START-DATE', {'WRK-CSV-START-DATE': ' '}), (10, 'WRK-CSV-START-TIME', {'WRK-CSV-START-TIME': '2500', 'WRK-CSV-START-DATE': '20240115', 'WRK-CSV-END-DATE': '20240115'}), (11, 'WRK-CSV-END-DATE', {'WRK-CSV-END-DATE': ' '}), (12, 'WRK-CSV-END-TIME', {'WRK-CSV-END-TIME': '2500', 'WRK-CSV-START-DATE': '20240115', 'WRK-CSV-END-DATE': '20240115'}), ] for dp_num, csv_field, overrides in targets: if csv_field not in base: continue new_rec = dict(base) for k, v in overrides.items(): new_rec[k] = v csv_parts = [str(new_rec.get(fname, '')) for fname in unstring_fields] csv_value = ','.join(csv_parts).ljust(r01_len)[:r01_len] new_rec['R01LINE'] = csv_value records.append(new_rec) logger.info(f" injected DP#{dp_num} T coverage record ({csv_field}={repr(overrides.get(csv_field,''))})") # ── 入口 ── def main(): if len(sys.argv) < 2: print("用法: python -m cobol_testgen [cobol文件2 ...] [输出目录]") sys.exit(1) args = sys.argv[1:] do_run = False gcov_mode = False temp_dir = None if '--run' in args: do_run = True args.remove('--run') if '--gcov' in args: gcov_mode = True args.remove('--gcov') if not _HAVE_RUNNER: logger.warning("--gcov: runner.py not found. Compile/run will be skipped. " "Use --gcov without runner only generates test data + static coverage.") i = 0 while i < len(args): if args[i] == '--temp-dir': if i + 1 < len(args): temp_dir = args[i + 1] args.pop(i + 1) args.pop(i) else: args.pop(i) break elif args[i].startswith('--temp-dir='): temp_dir = args[i].split('=', 1)[1] args.pop(i) break else: i += 1 cobol_files = [] outdir = None user_specified_outdir = False for a in args: p = Path(a) if p.is_dir() or (not p.suffix and p.parent.exists()): outdir = p user_specified_outdir = True elif p.suffix.upper() in ('.CBL', '.COB', '.CPY'): cobol_files.append(p) else: print(f"警告：跳过未知参数 {a}") if not cobol_files: print("错误：未找到任何 COBOL 文件") sys.exit(1) if outdir is None: from pathlib import Path as _Path _v3_root = _Path(__file__).parent.parent outdir = _v3_root / "runtime" user_specified_outdir = False outdir.mkdir(parents=True, exist_ok=True) (outdir / 'logs').mkdir(parents=True, exist_ok=True) log_path = outdir / 'logs' / f"cobol_testgen_{datetime.now():%Y%m%d_%H%M%S}.log" fh = logging.FileHandler(log_path, encoding="utf-8", mode="w") fh.setLevel(logging.DEBUG) fh.setFormatter(logging.Formatter( "%(asctime)s [%(levelname)s] %(name)s: %(message)s" )) sh = logging.StreamHandler() sh.setLevel(logging.INFO) sh.setFormatter(logging.Formatter("%(message)s")) root_logger = logging.getLogger() root_logger.setLevel(logging.DEBUG) root_logger.addHandler(fh) root_logger.addHandler(sh) programs = [] # ── Auto-route: split DB / non-DB per file ── db_files = [] non_db_files = [] for f in cobol_files: if 'EXEC SQL' in f.read_text(encoding='utf-8-sig').upper(): db_files.append(f) else: non_db_files.append(f) if db_files: import sys as _sys _v3_root = str(Path(__file__).parent.parent) if _v3_root not in _sys.path: _sys.path.insert(0, _v3_root) from orchestrator_db import GixsqlOrchestrator from config import Config _db_config = Config() _src_dir = cobol_files[0].parent if cobol_files else Path.cwd() _cpy_dirs = [str(_src_dir / '..' / 'cpy')] for filepath in db_files: pid = filepath.stem prog_outdir = outdir / pid prog_outdir.mkdir(parents=True, exist_ok=True) (prog_outdir / 'logs').mkdir(parents=True, exist_ok=True) (prog_outdir / 'coverage').mkdir(parents=True, exist_ok=True) logger.info(f"\n========== DB: {pid} ==========") orch = GixsqlOrchestrator( config=_db_config, program_id=pid, cobol_src_dir=str(_src_dir), copybook_dirs=[str(d) for d in _cpy_dirs], skip_jvm=True, ) vr = orch.run_all(generate_coverage=False) # Copy output files to outdir (skip if src == dst to avoid self-copy) if orch.runtime_dir.exists() and orch.runtime_dir.resolve() != prog_outdir.resolve(): for item in orch.runtime_dir.iterdir(): if item.name == "gixsql.log": continue dst = prog_outdir / item.name if item.is_dir(): shutil.copytree(str(item), str(dst), dirs_exist_ok=True) else: try: shutil.copy2(str(item), str(dst)) except PermissionError: logger.warning(f" Skipping locked file: {item.name}") logger.info(f" {pid}: rc={vr.exit_code} status={vr.status}") # Coverage report if gcov_mode and '--coverage' in getattr(_db_config, 'gixsql_compile_flags', ''): cov_result = orch.generate_coverage_report(output_dir=str(prog_outdir / 'coverage')) if cov_result.success: cv = cov_result.data.get("coverage", "unknown") logger.info(f" Coverage: {cv}") cov_dict = cov_result.data.get("_cov_dict") if cov_dict: rel = Path(prog_outdir / 'coverage' / f"{pid}_coverage.html") cov_dict['detail_relpath'] = str(rel.relative_to(outdir).as_posix()) programs.append(cov_dict) for filepath in non_db_files: if not filepath.exists(): logger.error(f"错误：文件不存在 {filepath}") continue source = filepath.read_text(encoding='utf-8') orig_source = source # 用于行号定位（与 gcov 对齐） source = resolve_copybooks( source, str(filepath.parent), extra_search_paths=[str(filepath.parent / '..' / 'cpy')], ) source = resolve_sql_includes(source, str(filepath.parent)) preprocessed = preprocess(source) file_sec = parse_file_section(preprocessed) data_div = extract_data_division(preprocessed) if data_div: data_div, declared_columns = strip_exec_sql_from_data_div(data_div) else: declared_columns = {} if not data_div: logger.error(f"错误：{filepath.name} 中没有 DATA DIVISION。") continue data_fields = parse_data_division(data_div) if not data_fields: logger.error(f"错误：{filepath.name} 中没有找到含 PIC 的字段。") continue fields_dict = [] parent_pic = {} filler_counter = 0 for f in data_fields: pi = f.pic_info name = f.name if name == 'FILLER': filler_counter += 1 if filler_counter > 1: name = f'FILLER_{filler_counter}' entry = { 'name': name, 'level': f.level, 'pic': f.pic, 'pic_info': { 'type': pi.type if pi else 'unknown', 'digits': pi.digits if pi else 0, 'decimal': pi.decimal if pi else 0, 'length': pi.length if pi else 0, 'signed': pi.signed if pi else False, }, 'value': f.value, 'values': f.values, 'section': f.section, 'is_filler': f.is_filler, 'redefines': f.redefines, 'usage': f.usage, 'occurs': f.occurs_count, 'occurs_depending': f.occurs_depending, } if f.is_88: entry['is_88'] = True entry['parent'] = f.parent if f.parent and f.parent in parent_pic: entry['pic_info'] = dict(parent_pic[f.parent]) else: parent_pic[name] = entry['pic_info'] fields_dict.append(entry) fields_dict = expand_occurs(fields_dict) sql_register_virtual_fields(fields_dict) fd_fields = {} field_to_fd = {} if file_sec: for fd_name, rec_names in file_sec.items(): fds = [] seen = set() for rec in rec_names: if rec not in seen: fds.append(rec) seen.add(rec) for child in _init_child_names(rec, fields_dict): if child not in seen: fds.append(child) seen.add(child) fd_fields[fd_name] = fds for child in fds: field_to_fd[child] = fd_name # Per-program output directory (always) prog_outdir = outdir / filepath.stem prog_outdir.mkdir(parents=True, exist_ok=True) (prog_outdir / 'logs').mkdir(parents=True, exist_ok=True) (prog_outdir / 'coverage').mkdir(parents=True, exist_ok=True) logger.info(f"\n========== {filepath.name} ==========") logger.info(f"\n字段列表:") logger.info(f"{'层级':<6} {'名称':<25} {'PIC':<15} {'类型':<12} {'长度':<5}") logger.info("-" * 65) for f in fields_dict: pi = f['pic_info'] t = pi.get('type', '?') l = pi.get('digits', 0) + pi.get('decimal', 0) or pi.get('length', 0) pic_display = str(f.get('pic', '')) if f.get('pic') else ('88-level' if f.get('is_88') else '') logger.info(f"{f['level']:<6} {f['name']:<25} {pic_display:<15} {t:<12} {l:<5}") proc_div = extract_procedure_division(preprocessed) branch_paths = [] assignments = {} if proc_div: branch_tree, assignments = build_branch_tree(proc_div, fields_dict, full_source=preprocessed) roles = classify_field_roles(branch_tree, assignments, fields_dict, source=preprocessed, proc_text=proc_div) logger.info(f"\n字段角色（输入/输出/出入/未用）:") for f in fields_dict: if f.get('is_88'): continue logger.info(f" {f['name']:<30} {roles.get(f['name'], '?')}") abend_list = CONFIG.get('abend_programs', []) if abend_list: extend_abend_programs(abend_list) branch_paths_with_assigns = enum_paths(branch_tree, fields_dict) path_infos = [] for c, a in branch_paths_with_assigns: filtered_c, term = get_term_type(c) path_infos.append((filtered_c, a, term)) def _is_skip(cons): eq1_true = 0 other = 0 for c in cons: if len(c) == 4 and c[0] == 'WRK-R01EOF': val = str(c[2]).strip("'\"") if val == '1' and c[1] == '=' and c[3]: eq1_true += 1 else: other += 1 return eq1_true > 0 and other == 0 skip_path_infos = [p for p in path_infos if _is_skip(p[0])] main_path_infos = [p for p in path_infos if not _is_skip(p[0])] _c01_types = {} for pi in path_infos: wants = tuple(c[3] for c in pi[0] if len(c) == 4 and c[0] == 'C01CHKRRC' and c[1] == '<>' and c[2] == 'ZERO') if wants: _c01_types[wants] = _c01_types.get(wants, 0) + 1 logger.info(f" C01 path types: {dict(sorted(_c01_types.items()))}") path_infos = main_path_infos if skip_path_infos: logger.info(f" Skip 路径: {len(skip_path_infos)} 条（将单独生成数据集）") open_dir = scan_all_file_directions(proc_div) if proc_div else {} if proc_div: logger.info(f"\n分支路径数：{len(branch_paths_with_assigns)}") for i, (path_cons, _path_assign) in enumerate(branch_paths_with_assigns): descs = [] for c in path_cons: if len(c) == 4: field, op, val, want = c if op == 'not_in': descs.append(f"{field} not in {val}") else: descs.append(f"{field} {op} {val} ({'T' if want else 'F'})") logger.debug(f" 路径 {i + 1}: {', '.join(descs)}") else: logger.warning("\n没有找到 PROCEDURE DIVISION。") branch_paths_with_assigns = [([], {})] path_infos = [([], {}, 'normal')] roles = {f['name']: 'unused' for f in fields_dict} records, kept_path_cons, term_types = generate_records(path_infos, fields_dict, assignments, file_sec=file_sec) def _is_eof_path(cons): last_eq1_true = -1 for i, c in enumerate(cons): if len(c) == 4 and c[0] == 'WRK-R01EOF': val = str(c[2]).strip("'\"") if val == '1' and c[1] == '=' and c[3]: last_eq1_true = i if last_eq1_true < 0: return False for i in range(last_eq1_true + 1, len(cons)): if len(cons[i]) == 4 and cons[i][0] == 'WRK-R01EOF': return False return True eof_mask = [_is_eof_path(c) for c, a, t in path_infos] eof_count = sum(eof_mask) if eof_count: term_types = ['eof' if e else t for e, t in zip(eof_mask, term_types)] logger.info(f" EOF 路径: {eof_count} 条（将单独执行）") multi_write_fds = _find_multi_write_fds(branch_tree, field_to_fd) if proc_div and branch_tree else set() if multi_write_fds: logger.info(f" 检测到多 WRITE FD: {', '.join(sorted(multi_write_fds))}") _chain_prev(records, path_infos, fields_dict, fd_fields, field_to_fd, open_dir) # P4: inject empty EMP-ID record to trigger R01EMP-ID = SPACE path _inject_empty_emp_rec(records, fields_dict) # P5: inject records for uncovered C01 decision points (DP#9-#12 T) _inject_c01_coverage_records(records, fields_dict, assignments) # P6: coordinate cross-file keys for multi-file matching programs _coordinate_multi_file_keys(records, kept_path_cons, fields_dict, assignments, file_sec, term_types=term_types, open_dir=open_dir) # P6.5: 金额-区间对齐（N:1 集約→GRADE 匹配路径可达） _coordinate_range_matching(records, fields_dict, file_sec, open_dir, term_types) # P7: inject leave-type coverage for 2030PRIOSOR EVALUATE branches _inject_leave_type_scenarios(records, term_types) # P8: opaque FD + READ INTO — 将 WS 子字段值序列化回 FD 记录字节 _serialize_read_into_records(records, assignments, fields_dict, file_sec) # P9: MERGE 程序 — 为 USING 输入文件注入 S/B 记录 _inject_merge_input_records(records, fields_dict, file_sec, fd_fields, proc_div, term_types=term_types) # P10: LINAGE 分页（WRITE AT END-OF-PAGE）— 补充分页体积记录 _inject_end_of_page_records(records, source, proc_div, file_sec, open_dir, term_types=term_types) # P11: SORT 计数守卫（RELEASE + IF <= <上限>）— 补充上限体积记录 _inject_sort_limit_records(records, proc_div, fields_dict, term_types=term_types) if _HAVE_TOSQL: sql_meta = collect_sql_meta(assignments, declared_columns) db_input = build_db_input( branch_paths_with_assigns, fields_dict, assignments, sql_meta, declared_columns, records=records, ) else: db_input = None outpath = prog_outdir / 'main' / 'json' / (filepath.stem + '.json') output_json(records, outpath, roles, fd_fields=fd_fields, field_to_fd=field_to_fd, open_dir=open_dir, term_types=term_types, db_input=db_input if db_input else None, data_fields=fields_dict) select_info = parse_file_control(preprocessed) output_input_files(records, prog_outdir / 'main' / 'input', filepath.stem, roles, fd_fields, field_to_fd, open_dir, term_types=term_types, data_fields=fields_dict, select_info=select_info) # P12: 子程序输入供给（CALL 目标子程序的 INPUT FD 输入文件） _provision_sub_input_files( source, str(filepath.parent), prog_outdir / 'main' / 'input' ) # ── Skip 数据集（主 FD 空文件触发 PERFORM UNTIL 条件即时满足）── skip_records = None skip_term_types = None if skip_path_infos: skip_records, _, skip_term_types = generate_records( skip_path_infos, fields_dict, assignments, file_sec=file_sec) # 剥离主 FD 的输入字段（记录不写入输入文件 → 文件为空） eof_fd = 'R01INNFIL' eof_fd_fields = set(fd_fields.get(eof_fd, [])) eof_fd_dir = (open_dir or {}).get(eof_fd, '') for rec in skip_records: for fname in list(rec.keys()): if fname in eof_fd_fields: r = roles.get(fname, 'unused') if eof_fd_dir in ('INPUT', 'I-O') and r in ('input', 'inout'): del rec[fname] # 写 Skip JSON skip_outpath = prog_outdir / 'skip' / 'json' / (filepath.stem + '.json') output_json(skip_records, skip_outpath, roles, fd_fields=fd_fields, field_to_fd=field_to_fd, open_dir=open_dir, term_types=skip_term_types, data_fields=fields_dict) # 写 Skip 输入文件（主 FD 因字段已剥离而不输出） skip_input_dir = prog_outdir / 'skip' / 'input' output_input_files(skip_records, skip_input_dir, filepath.stem + '_skip', roles, fd_fields, field_to_fd, open_dir, term_types=skip_term_types, data_fields=fields_dict, select_info=select_info) # 强制写空主 FD 输入文件（0 条记录，COBOL 运行时需要文件存在） eof_input_path = skip_input_dir / f'{filepath.stem}_skip_{eof_fd}.json' eof_input_path.parent.mkdir(parents=True, exist_ok=True) with open(eof_input_path, 'w', encoding='utf-8') as f: json.dump([], f) # 空二进制文件（COBOL INPUT 模式需要物理文件存在） eof_assign = select_info.get(eof_fd, {}).get('assign', '') if eof_assign: bin_path = skip_input_dir / eof_assign bin_path.parent.mkdir(parents=True, exist_ok=True) bin_path.write_bytes(b'') logger.info(f" Skip 数据集: {skip_outpath}（空 {eof_fd}）") gcov_data = None if gcov_mode and proc_div and _HAVE_GCOV and _HAVE_RUNNER: _temp = temp_dir or str(prog_outdir / '.gcov_cache') source_dir = str(filepath.parent) expected_records: list[dict] = [{}] * len(records) if file_sec and os.path.exists(outpath): with open(outpath, encoding='utf-8') as f: full_json = json.load(f) json_records = full_json.get('records', []) for i in range(len(records)): exp = {} if i < len(json_records): json_rec = json_records[i] for fd_name in file_sec: eo = json_rec.get('expected_output', {}) if fd_name in eo: exp.update(eo[fd_name]) expected_records[i] = exp group_results, gcov_data = run_all( filepath.stem, str(prog_outdir), _temp, fields_dict, fd_fields, select_info, open_dir, term_types, records, expected_records=expected_records, source_dir=source_dir, path_infos=path_infos, multi_write_fds=multi_write_fds, skip_records=skip_records, skip_term_types=skip_term_types, ) passed = sum(1 for r in group_results if r.passed) total = len(group_results) logger.info(f"\n 执行验证: {passed}/{total} 组通过") if passed < total: for r in group_results: if not r.passed and r.details: fails = [d for d in r.details if not d.match][:3] for d in fails: logger.warning(f" [{r.name}] {d.field}: " f"期望={d.expected!r}, 实际={d.actual!r}") if do_run and proc_div and _HAVE_RUNNER: run_and_compare( filepath.stem, str(prog_outdir), fields_dict, fd_fields, select_info, open_dir, term_types, records, ) logger.info(f"\n输出：{outpath}（{len(records)} 条记录）") logger.debug(f"\n记录明细：") for i, rec in enumerate(records, 1): vals = [] for f in fields_dict: r = roles.get(f['name'], '?') marker = f"[{r[0].upper()}]" if r != '?' and r != 'unused' else '' vals.append(f"{marker}{f['name']}={rec.get(f['name'], '?')}") logger.debug(f" 记录 {i}: {' | '.join(vals)}") (prog_outdir / 'coverage').mkdir(parents=True, exist_ok=True) cov_prefix = str(prog_outdir / 'coverage' / filepath.stem) # DEBUG: check DP#3 constraints dp3_t_count = 0 dp3_f_count = 0 dp3_t_paths = 0 dp3_f_paths = 0 dp3_sample = set() for cons, _ in branch_paths_with_assigns: has_t = False has_f = False for c in cons: if len(c) == 4: c0 = str(c[0]).strip() c1 = str(c[1]).strip() c2 = str(c[2]).strip() c3 = c[3] if c0 == 'WRK-R02KEY' and c1 == '>=' and c2 == 'WRK-R01KEY': if c3: dp3_t_count += 1 has_t = True else: dp3_f_count += 1 has_f = True elif c0 == 'WRK-R02KEY': dp3_sample.add(f"({c0},{c1},{c2},{c3})") if has_t: dp3_t_paths += 1 if has_f: dp3_f_paths += 1 logger.info(f"DEBUG DP#3: T={dp3_t_count}/{dp3_t_paths}paths, F={dp3_f_count}/{dp3_f_paths}paths (total={len(branch_paths_with_assigns)})") if dp3_sample: logger.info(f"DEBUG DP#3 other constraints: {sorted(dp3_sample)[:5]}") cov_result = run_coverage(branch_tree, branch_paths_with_assigns, fields_dict, orig_source, cov_prefix, index_relpath='index.html', gcov_data=gcov_data) programs.append(cov_result) programs[-1]['detail_relpath'] = f'{filepath.stem}/coverage/{filepath.stem}_coverage.html' if programs: generate_coverage_index(programs, outdir / 'coverage') logger.info(f"\n覆盖率总览：{outdir / 'coverage' / 'index.html'}") # ════════════════════════════════════════════ # Phase 1: 可编程 API（供 orchestrator.py 调用） # ════════════════════════════════════════════ def extract_structure(cobol_source: str, copybook_dirs: list = None) -> dict: """分析 COBOL 源码的结构，返回结构摘要。不生成测试数据，只做静态分析。 Args: cobol_source: COBOL source text. copybook_dirs: Optional list of COPYBOOK search paths. Returns: dict with: paragraphs, decision_points, branch_tree, file_count, open_directions, has_search_all, has_evaluate, has_call, has_break, total_branches, total_paragraphs """ preprocessed = preprocess(cobol_source, extra_search_paths=copybook_dirs) data_div = extract_data_division(preprocessed) data_fields = parse_data_division(data_div) if data_div else [] fields_dict = [] for idx, f in enumerate(data_fields): entry = { 'name': f.name if f.name != 'FILLER' else f'FILLER_{idx + 1}', 'level': f.level, 'pic': f.pic, 'pic_info': { 'type': f.pic_info.type if f.pic_info else 'unknown', 'digits': f.pic_info.digits if f.pic_info else 0, 'decimal': f.pic_info.decimal if f.pic_info else 0, 'length': f.pic_info.length if f.pic_info else 0, 'signed': f.pic_info.signed if f.pic_info else False, }, 'section': f.section, 'occurs': f.occurs_count, 'occurs_depending': f.occurs_depending, 'redefines': f.redefines, 'usage': f.usage, } if f.is_88: entry['is_88'] = True entry['parent'] = f.parent entry['value'] = f.value entry['values'] = f.values fields_dict.append(entry) fields_dict = expand_occurs(fields_dict) proc_div = extract_procedure_division(preprocessed) branch_tree = None assignments = {} if proc_div: branch_tree, assignments = build_branch_tree_fallback(proc_div, fields_dict) file_sec = parse_file_section(preprocessed) open_dir = scan_all_file_directions(proc_div) if proc_div else {} from .models import BrIf, BrEval, BrSeq, BrPerform, BrSearch, Assign, CondAnd, CondOr decision_points = [] total_branches = 0 def _walk(node, counter): nonlocal total_branches if isinstance(node, BrIf): counter[0] += 1 branches = 2 decision_points.append({ "id": counter[0], "kind": "IF", "label": str(node.condition)[:80], "branches": branches, }) total_branches += branches _walk(node.true_seq, counter) _walk(node.false_seq, counter) elif isinstance(node, BrEval): counter[0] += 1 seen_br = set() uni_count = 0 for v, _ in node.when_list: brn = f"WHEN {v}" if brn not in seen_br: uni_count += 1 seen_br.add(brn) n = uni_count + (1 if node.has_other and "OTHER" not in seen_br else 0) decision_points.append({"id": counter[0], "kind": "EVALUATE", "label": str(node.subject)[:80], "branches": n}) total_branches += n for _, seq in node.when_list: _walk(seq, counter) _walk(node.other_seq, counter) elif isinstance(node, BrSeq): for child in node.children: _walk(child, counter) elif isinstance(node, BrPerform): if node.condition and node.perf_type in ('until', 'para_until', 'varying', 'para_varying'): counter[0] += 1 decision_points.append({ "id": counter[0], "kind": "PERFORM", "label": str(node.condition)[:80], "branches": 2, }) total_branches += 2 _walk(node.body_seq, counter) elif isinstance(node, BrSearch): _walk(node.at_end_seq, counter) for _, seq in node.when_list: _walk(seq, counter) if branch_tree: _walk(branch_tree, [0]) lines = proc_div.split('\n') if proc_div else [] paragraphs = set() for line in lines: m = re.match(r'^\s*([A-Z0-9][A-Z0-9-]*)\.\s*$', line.strip()) if m: paragraphs.add(m.group(1)) select_files = parse_file_control(preprocessed) open_directions_detail = open_dir has_divide = bool(re.search(r'\bDIVIDE\b', cobol_source.upper())) has_inspect = bool(re.search(r'\bINSPECT\b', cobol_source.upper())) has_string = bool(re.search(r'\bSTRING\b', cobol_source.upper())) divide_constants = [] if has_divide and proc_div: for dm in re.finditer(r'\bDIVIDE\s+([\d.]+)\b', proc_div, re.IGNORECASE): val = dm.group(1) try: divide_constants.append(float(val)) except ValueError: pass perform_patterns = [] def _walk_performs(node): if isinstance(node, BrPerform): entry = { "type": node.perf_type, "target": node.target, "condition": node.condition, "times": node.times, "varying_var": node.varying_var, } perform_patterns.append(entry) _walk_performs(node.body_seq) elif isinstance(node, BrIf): _walk_performs(node.true_seq) _walk_performs(node.false_seq) elif isinstance(node, BrEval): for _, seq in node.when_list: _walk_performs(seq) _walk_performs(node.other_seq) elif isinstance(node, BrSeq): for c in node.children: _walk_performs(c) if branch_tree: _walk_performs(branch_tree) main_loop = None def _find_main_loop(node, depth=0): nonlocal main_loop if main_loop is not None: return if isinstance(node, BrPerform): if _perform_has_read(node): main_loop = { "type": node.perf_type, "read_file": _perform_read_file(node), "has_at_end": False, } return _find_main_loop(node.body_seq, depth + 1) elif isinstance(node, BrIf): _find_main_loop(node.true_seq, depth + 1) _find_main_loop(node.false_seq, depth + 1) elif isinstance(node, BrEval): for _, seq in node.when_list: _find_main_loop(seq, depth + 1) _find_main_loop(node.other_seq, depth + 1) elif isinstance(node, BrSeq): for c in node.children: _find_main_loop(c, depth + 1) def _perform_has_read(perf_node): def _walk_seq(seq): if isinstance(seq, Assign): if seq.source_info.get('type') == 'read_into': return True elif isinstance(seq, BrSeq): for ch in seq.children: if _walk_seq(ch): return True return False return _walk_seq(perf_node.body_seq) def _perform_read_file(perf_node): def _walk_seq(seq): if isinstance(seq, Assign): if seq.source_info.get('type') == 'read_into': return seq.source_info.get('file', '') elif isinstance(seq, BrSeq): for ch in seq.children: result = _walk_seq(ch) if result: return result return None return _walk_seq(perf_node.body_seq) if branch_tree: _find_main_loop(branch_tree) if_types = {"total": 0, "comparison": 0, "equality": 0, "compound": 0, "nested_depth": 0} def _walk_if_types(node, depth=0): if isinstance(node, BrIf): if_types["total"] += 1 if_types["nested_depth"] = max(if_types["nested_depth"], depth) ct = node.cond_tree if ct: leaves = collect_leaves(ct) if isinstance(ct, (CondAnd, CondOr)): if_types["compound"] += 1 for leaf in leaves: if leaf.op in ('>', '<', '>=', '<='): if_types["comparison"] += 1 elif leaf.op in ('=', '<>'): if_types["equality"] += 1 _walk_if_types(node.true_seq, depth + 1) _walk_if_types(node.false_seq, depth + 1) elif isinstance(node, BrEval): for _, seq in node.when_list: _walk_if_types(seq, depth + 1) _walk_if_types(node.other_seq, depth + 1) elif isinstance(node, BrPerform): _walk_if_types(node.body_seq, depth + 1) elif isinstance(node, BrSeq): for c in node.children: _walk_if_types(c, depth + 1) if branch_tree: _walk_if_types(branch_tree) variable_patterns = { "has_prev_key": False, "has_accumulator": False, "has_error_flag": False, "has_switch": False, "has_index": False, "has_save_area": False, "has_counter": False, "has_work": False, } for f in fields_dict: name = f.get('name', '') if re.search(r'\bWS-PREV[-_]', name, re.IGNORECASE): variable_patterns["has_prev_key"] = True if re.search(r'[-_]CNT\b', name, re.IGNORECASE) or re.search(r'[-_]ACCUM\b', name, re.IGNORECASE): variable_patterns["has_accumulator"] = True if re.search(r'[-_]ERR\b', name, re.IGNORECASE) or re.search(r'[-_]ERROR[-_]', name, re.IGNORECASE): variable_patterns["has_error_flag"] = True if re.search(r'[-_]SW\b', name, re.IGNORECASE) or re.search(r'[-_]FLAG\b', name, re.IGNORECASE): variable_patterns["has_switch"] = True if re.search(r'[-_]IDX\b', name, re.IGNORECASE) or re.search(r'[-_]INDX\b', name, re.IGNORECASE) or re.search(r'[-_]SUB\b', name, re.IGNORECASE): variable_patterns["has_index"] = True if re.search(r'[-_]SAVE[-_]', name, re.IGNORECASE) or re.search(r'[-_]HOLD[-_]', name, re.IGNORECASE): variable_patterns["has_save_area"] = True if re.search(r'[-_]CNT\b', name, re.IGNORECASE) or re.search(r'[-_]COUNT\b', name, re.IGNORECASE): variable_patterns["has_counter"] = True if name.startswith('WS-') and not re.search(r'(?:CNT|ERR|SW|IDX|INDX|SUB|SAVE|HOLD|PREV|ACCUM)', name, re.IGNORECASE): if re.search(r'[-_]W\b|[-_]WORK\b|[-_]WK\b|^WS-W[0O]\w', name, re.IGNORECASE): variable_patterns["has_work"] = True open_pattern = "sequential" if proc_div: proc_upper = proc_div.upper() open_positions = [m.start() for m in re.finditer(r'\bOPEN\b', proc_upper)] close_positions = [m.start() for m in re.finditer(r'\bCLOSE\b', proc_upper)] if open_positions and close_positions: for i, opos in enumerate(open_positions): for cpos in close_positions: if cpos > opos: for opos2 in open_positions: if opos2 > cpos: open_pattern = "open-close-open" break if open_pattern == "open-close-open": break if open_pattern == "open-close-open": break return { "paragraphs": sorted(paragraphs) if paragraphs else [], "decision_points": decision_points, "branch_tree": branch_tree, "file_count": len(file_sec) if file_sec else 0, "open_directions": open_dir, "has_search_all": any('SEARCH' in str(dp.get('label', '')) for dp in decision_points), "has_evaluate": any(dp['kind'] == 'EVALUATE' for dp in decision_points), "has_call": 'CALL' in cobol_source.upper(), "has_break": any('KEY' in str(dp.get('label', '')).upper() for dp in decision_points), "total_branches": total_branches, "total_paragraphs": len(paragraphs), "branch_tree_obj": branch_tree, "select_files": select_files, "open_directions_detail": open_directions_detail, "has_divide": has_divide, "divide_constants": divide_constants, "has_inspect": has_inspect, "has_string": has_string, "perform_patterns": perform_patterns, "main_loop": main_loop, "if_types": if_types, "variable_patterns": variable_patterns, "open_pattern": open_pattern, "data_fields": fields_dict, } def generate_data(cobol_source: str, structure: dict = None, copybook_dirs: list = None) -> list[dict]: """根据 COBOL 源码生成覆盖所有路径的测试数据。 Args: cobol_source: COBOL 程序原始源码文本（未预处理）。 内部会调 preprocess + resolve_copybooks + resolve_sql_includes。 如果已预处理过，传进来会因字段列表不全导致数据不完整。 COPYBOOK 路径通过 copybook_dirs 参数传入。 structure: 可选，如果已调用 extract_structure() 可传入避免重复解析 copybook_dirs: 可选，COPYBOOK 搜索路径列表。指定后可自动展开 COPY 和 EXEC SQL INCLUDE。 Returns: list[dict]: 测试数据记录列表，每条包含所有字段的值 """ if structure is None: structure = extract_structure(cobol_source) branch_tree = structure.get("branch_tree_obj") if branch_tree is None: return [] if copybook_dirs: src_resolved = resolve_copybooks(cobol_source, '.', extra_search_paths=copybook_dirs) src_resolved = resolve_sql_includes(src_resolved, '.') preprocessed = preprocess(src_resolved) else: # Also try SQL include resolution without copybook src_sql = resolve_sql_includes(cobol_source, '.') preprocessed = preprocess(src_sql) data_div = extract_data_division(preprocessed) data_fields = parse_data_division(data_div) if data_div else [] fields_dict = [] for f in data_fields: entry = { 'name': f.name, 'level': f.level, 'pic': f.pic, 'pic_info': { 'type': f.pic_info.type if f.pic_info else 'unknown', 'digits': f.pic_info.digits if f.pic_info else 0, 'decimal': f.pic_info.decimal if f.pic_info else 0, 'length': f.pic_info.length if f.pic_info else 0, 'signed': f.pic_info.signed if f.pic_info else False, }, 'section': f.section, 'occurs': f.occurs_count, 'occurs_depending': f.occurs_depending, 'value': f.value, 'values': f.values, 'redefines': f.redefines, 'usage': f.usage, } if f.is_88: entry['is_88'] = True entry['parent'] = f.parent fields_dict.append(entry) fields_dict = expand_occurs(fields_dict) proc_div = extract_procedure_division(preprocessed) _, assignments = build_branch_tree_fallback(proc_div, fields_dict) # EXEC SQL ブロックは preprocess で除去されるため、 # 原ソースから直接抽出して assignments にマージする from .core import extract_sql_assignments sql_assigns = extract_sql_assignments(cobol_source) for tgt, asgn_list in sql_assigns.items(): for asgn in asgn_list: assignments.setdefault(tgt, []).append(asgn) file_sec = parse_file_section(preprocessed) branch_paths_unfiltered = mcdc_enum_paths(branch_tree, fields_dict) path_infos = [] for c, a in branch_paths_unfiltered: for cc in c: if len(cc) >= 4 and str(cc[0]) in ('WS-STATUS', 'WS-APPL-ID'): print(f" PATH-DEBUG: {cc}", flush=True) break filtered_c, term = get_term_type(c) path_infos.append((filtered_c, a, term)) _fdict_names = {f['name'] for f in fields_dict} def _resolve_field(fn: str) -> str: if fn.startswith("_"): return fn ufn = fn.upper() if ' OF ' in ufn: fn = fn.split(' OF ')[0].strip() if fn in _fdict_names: return fn # Check subscript: WS-PLAN-CODE(WS-PLAN-IDX) -> WS-PLAN-CODE m = re.match(r'^(\w[\w-]*)\s*\(', fn) if m: base = m.group(1) if base in _fdict_names: return base # Check if any field in fdict starts with base + "(" if any(f.startswith(base + "(") for f in _fdict_names): return base return fn def _is_arith_expr(fn): return any(op in fn for op in [' + ', ' - ', ' * ', ' / ']) filtered_paths = [] for cons_list, asgn, term in path_infos: clean = [] for c in cons_list: if len(c) >= 4: fn = _resolve_field(str(c[0])) if fn in _fdict_names or fn.startswith("_") or _is_arith_expr(str(c[0])) or \ any(f.startswith(fn + "(") for f in _fdict_names): c = list(c); c[0] = fn clean.append(tuple(c)) else: clean.append(c) filtered_paths.append((clean, asgn, term)) path_infos = filtered_paths records, kept_paths, term_types = generate_records(path_infos, fields_dict, assignments, file_sec=file_sec) # ── Coverage marking: which decision branches are actually covered ── if branch_tree and fields_dict: try: dp_list, leaf_stats = collect_decision_points(branch_tree, fields_dict) cov_paths = [(pi[0], pi[1]) for pi in path_infos if isinstance(pi, (list, tuple)) and len(pi) >= 2] mark_coverage(dp_list, leaf_stats, cov_paths, fields_dict) if structure is not None: structure['coverage'] = { 'decision_points': [{ 'id': dp.id, 'kind': dp.kind, 'label': getattr(dp, 'label', '')[:60], 'branches': len(dp.branch_names), 'covered': len(dp.active_branches), } for dp in dp_list], 'total': sum(len(dp.branch_names) for dp in dp_list), 'covered': sum(len(dp.active_branches) for dp in dp_list), 'pct': sum(len(dp.active_branches) for dp in dp_list) / max(sum(len(dp.branch_names) for dp in dp_list), 1) * 100, } except Exception as e: if structure is not None: structure['coverage'] = {'error': str(e)[:80]} if records: import re as _re proc_upper = (proc_div or "").upper() for m in _re.finditer(r'IF\s+(\w[\w-]*)\s*[=<>]\s*(\w[\w-]*)', proc_upper): lhs, rhs = m.group(1), m.group(2) lhs_in = any(lhs == f['name'] for f in fields_dict) rhs_in = any(rhs == f['name'] for f in fields_dict) if lhs_in and rhs_in and any(lhs in r for r in records) and any(rhs in r for r in records): half = max(1, len(records) // 2) for i, rec in enumerate(records): if lhs in rec and rhs in rec and i < half: rec[rhs] = rec[lhs] # P6: 跨文件キー協調（DB/非DB 両パイプ共通）— 部分記錄のキーを # 主表（R02/R03 等）と一致させ、照合 STAGE/DB SELECT 経路を到達可能にする。 # 出力 FD は open_dir で除外し、入力 FD 同士のみ協調する。 if records and file_sec: try: open_dir = scan_all_file_directions(proc_div or '') _coordinate_multi_file_keys( records, kept_paths, fields_dict, assignments, file_sec, term_types=term_types, open_dir=open_dir, ) # P6.5: 金额-区间对齐（N:1 集約→GRADE 匹配路径可达，DB/非DB 共通） _coordinate_range_matching( records, fields_dict, file_sec, open_dir, term_types, ) except Exception as e: logger.warning(f" cross-file key coordination skipped: {e}") return records def incremental_supplement(branch_tree, decision_gaps: list[int], fields_dict: list = None) -> list[dict]: """针对未覆盖的决策点，增量生成补充测试数据。 Args: branch_tree: extract_structure() 返回的 branch_tree 字段 decision_gaps: 未覆盖的决策点 ID 列表，如 [1, 3, 5] fields_dict: 字段定义列表（DATA DIVISION 展开后），提供后生成含字段值的记录 Returns: list[dict]: 增量测试数据，格式与 generate_data() 兼容 """ from .models import BrIf, BrEval, BrSeq target_decisions = set(decision_gaps) found = [] def _find_decisions(node, counter): if isinstance(node, BrIf): counter[0] += 1 if counter[0] in target_decisions: found.append(("IF", node.condition)) _find_decisions(node.true_seq, counter) _find_decisions(node.false_seq, counter) elif isinstance(node, BrEval): counter[0] += 1 if counter[0] in target_decisions: found.append(("EVALUATE", node.subject)) for _, seq in node.when_list: _find_decisions(seq, counter) _find_decisions(node.other_seq, counter) elif isinstance(node, BrSeq): for child in node.children: _find_decisions(child, counter) _find_decisions(branch_tree, [0]) supplements = [] for i, (kind, label) in enumerate(found): rec = {} if fields_dict: rec = make_base_record(i + 1, fields_dict) rec["_dec_id"] = f"incr_{i}" rec["_kind"] = kind rec["_label"] = str(label)[:60] supplements.append(rec) return supplements