# cobol_testgen 核心引擎模块 - 详细设计文档 > 模块路径: `cobol_testgen/` > 版本: V3 (2026技术大赛) > 依赖: Python 3.13+, `lark>=1.1.0` --- ## 1. 模块概述 ### 1.1 职责 `cobol_testgen` 是 COBOL 迁移验证平台 V3 的核心引擎, 负责: 1. **解析** COBOL 源码 (DATA DIVISION + PROCEDURE DIVISION) 2. **构建** 分支控制流树 (Branch Tree) 3. **枚举** 所有可达路径 (路径级约束) 4. **生成** 满足路径约束的测试数据记录 5. **输出** JSON 格式测试数据 + HTML 覆盖率报告 ### 1.2 边界 | 在范围内 | 不在范围内 | |---------|-----------| | COBOL 源码静态分析 | COBOL 程序动态执行 (由 runner.py 调用 GnuCOBOL) | | 分支树构建 + 路径枚举 | LLM 推理 (由外部 deepseek API 调用) | | 测试数据值生成 | SQL 数据库操作 (to_sql.py 辅助生成 DB 输入行) | | JSON/HTML 输出 | IDE 集成、CI/CD 管道 | ### 1.3 依赖关系 ``` cobol_testgen/ __init__.py <- 公开 API 入口 (main) models.py <- 共享数据模型 (零外部依赖) read.py <- INPUT 层: 预处理 + DATA DIVISION 解析 core.py <- CORE 层: PROCEDURE DIVISION 分支树构建 procedure_parser.py <- CORE 层: 行级状态机解析器 (新) pipeline_bridge.py <- CORE 层: 新旧解析器桥接 cond.py <- CONDITION 层: 条件解析 + MC/DC design.py <- DESIGN 层: 路径枚举 + 值生成 design_mcdc.py <- DESIGN 层: O(N) 非爆炸路径枚举 coverage.py <- COVERAGE 层: 分支覆盖标记 + HTML 报告 output.py <- OUTPUT 层: JSON 输出 to_sql.py <- SQL 辅助: WHERE 约束解析 + DB 输入行生成 flatfile.py <- I/O 辅助: 固定长度平面文件读写 file_io.py <- I/O 辅助: DISPLAY/COMP/COMP-3 二进制编解码 runner.py <- 执行层: 编译-执行-验证 (调用 GnuCOBOL) gcov.py <- 覆盖率辅助: gcov 数据解析 data_merger.py <- 数据整合: 白盒+机能+策略数据合并 grammar.lark <- Lark 语法: DATA DIVISION 解析 procedure_grammar.lark <- Lark 语法: PROCEDURE DIVISION 解析 __main__.py <- 入口: python -m cobol_testgen ``` **外部依赖:** | 依赖 | 用途 | |------|------| | lark>=1.1.0 | Lark 解析器框架 (Earley parser, dynamic lexer) | | deepseek API | LLM 路径生成 (可选, 回退到规则引擎) | | GnuCOBOL (cobc) | COBOL 编译-执行 (runner.py 调用) | | gcov | 代码覆盖率采集 (gcov.py 调用) | --- ## 2. 文件清单 | 文件 | 行数 | 职责 | |------|------|------| | `__init__.py` | 2189 | 公开 API 入口、OCCURS 展开、PREV 连锁、跨文件键值协调、MERGE/SORT/LINAGE 注入、子程序输入供给 | | `__main__.py` | 3 | python -m cobol_testgen 入口 | | `models.py` | 118 | 共享数据模型: PicInfo, FieldDef, BrSeq, BrIf, BrEval, BrPerform, BrSearch, CondLeaf, Assign, CallNode, ExitNode, GoTo, ParseError, ProcParseResult | | `read.py` | 644 | COBOL 源码预处理 (fixed/free format)、COPYBOOK 展开、DATA DIVISION 解析 (Lark grammar.lark)、FILE-CONTROL 解析、OPEN 语句扫描 | | `core.py` | 2137 | PROCEDURE DIVISION 分支树构建 (_BrParser 类)、段落扫描、赋值追踪、SQL 虚拟字段注册、算术表达式解析、EVALUATE/PERFORM/SEARCH/READ/WRITE 解析 | | `procedure_parser.py` | 527 | 新版行级状态机解析器 (Tier 1: 状态机提取嵌套结构, Tier 2: 规则条件解析) | | `pipeline_bridge.py` | 225 | 新旧解析器桥接: 新解析器优先, 旧解析器 3s 超时回退 | | `cond.py` | 497 | COBOL 条件表达式解析 (AND/OR/NOT/括号)、MC/DC 约束集生成 (mcdc_sets)、满足值计算 (satisfying_value) | | `design.py` | 2014 | 路径枚举 (enum_paths)、基础记录生成 (make_base_record)、约束应用 (apply_constraint)、赋值传播 (propagate_assignments)、链追溯 (trace_to_root) | | `design_mcdc.py` | 409 | O(N) 非爆炸路径枚举: 每个决策点生成 T/F 两条路径, 保证全覆盖无指数爆炸 | | `coverage.py` | 1451 | 决策点收集 (collect_decision_points)、分支覆盖标记 (mark_coverage)、中文 HTML 报告生成 | | `output.py` | 184 | JSON 输出: 按 FD 分组 input/expected_output/working_storage | | `to_sql.py` | 948 | SQL 元数据提取 (collect_sql_meta)、WHERE 宿主变量 MOVE 链解析、DB 输入行生成 (build_db_input) | | `flatfile.py` | 309 | 平面文件 I/O: FD 布局分析 (analyze_fd_layout)、固定长度记录读写 | | `file_io.py` | 234 | 二进制编解码: DISPLAY/COMP/COMP-3 pack/unpack、RECORDING MODE V 处理 | | `runner.py` | 516 | 编译-执行-验证: GnuCOBOL cobc 调用、字段对比、分组执行 | | `gcov.py` | 165 | gcov 覆盖率数据解析 (.cbl.gcov -> {行号: 执行次数}) | | `data_merger.py` | 130 | 数据整合: 白盒 (MC/DC) + 机能 (LLM) + 策略 (HINA 分类) 数据合并去重 | | `grammar.lark` | 40 | Lark 语法: DATA DIVISION 解析 | | `procedure_grammar.lark` | 203 | Lark 语法: PROCEDURE DIVISION 解析 | --- --- ## 3. 核心数据结构 所有模型定义在 `models.py` (118 行), 无外部依赖。 ### 3.1 字段定义 ```python @dataclass class PicInfo: type: str = 'unknown' # "numeric" | "alphanumeric" | "alphabetic" digits: int = 0 # 整数位数 (如 9(7) -> 7) decimal: int = 0 # 小数位数 (如 V99 -> 2) length: int = 0 # 总长度 (alphanumeric 用) signed: bool = False # 是否有符号 @dataclass class FieldDef: name: str # 字段名 (大写) level: int # 层号 (01, 05, 77, 88 等) pic: str | None = None # PIC 子句原始文本 pic_info: PicInfo | None = None is_filler: bool = False occurs_count: int = 0 occurs_depending: str | None = None redefines: str | None = None usage: str | None = None # "COMP" | "COMP-3" | "BINARY" | "DISPLAY" value: str | None = None values: list[str] | None = None # 88 级多值 is_88: bool = False parent: str | None = None section: str | None = None ``` ### 3.2 分支树节点 ```python class BrSeq: # 顺序语句序列 (容器节点) children = [] # BrIf | BrEval | BrPerform | BrSearch | Assign | ... class BrIf: # IF 条件分支 condition # 条件原文 cond_tree = None # 条件树 (core.py 解析时赋值) true_seq # THEN 分支 false_seq # ELSE 分支 class BrEval: # EVALUATE 多分支 subject # 主体 subjects = [] # ALSO 多主体 when_list = [] # [(condition_text, BrSeq)] other_seq # WHEN OTHER has_other = False class BrPerform: # PERFORM 循环/调用 perf_type # "until" | "varying" | "times" | "para" | "sort" condition # UNTIL 条件 body_seq # 循环体 class BrSearch: # SEARCH 表查找 table_name, is_all, at_end_seq, when_list, has_at_end ``` ### 3.3 条件树 ```python class CondLeaf: field, op, value # 叶条件 class CondNot: child # NOT 取反 class CondAnd: left, right # AND class CondOr: left, right # OR ``` ### 3.4 其他节点 ```python class Assign: target, source_info # 赋值 (MOVE/COMPUTE/READ INTO/WRITE FROM) class CallNode: program_name, using_params # CALL 子程序 class GoTo: target, body_seq # GO TO 跳转 class ExitNode: exit_type # EXIT 退出 ``` ### 3.5 约束与路径 ```python Constraint = tuple # (field, op, value, want_true) Path = list[Constraint] class ParseError: line, message, severity class ProcParseResult: tree, assignments, errors, fallback_to_ai ``` --- ## 4. 解析流程 ### 4.1 源码预处理 (read.py) 入口: `preprocess(source, extra_search_paths)` 1. COPYBOOK 展开 -> EXEC SQL/CICS 移除 -> VALUE 逗号清理 2. & 连接行合并 -> PIC 小数点转换 -> 格式检测 (fixed/free) 3. 注释移除 -> 续行合并 ### 4.2 DATA DIVISION 解析 (read.py) 使用 `grammar.lark` (Lark Earley parser)。关键: Earley parser 处理歧义语法, 命名终端 USAGE_VAL 避免 Lark 过滤 tree children。 ### 4.3 PROCEDURE DIVISION 分支树 (core.py) 入口: `build_branch_tree(proc_text, fields, full_source)` 段落扫描 (`scan_paragraphs`) -> 分支解析 (`_BrParser` 递归下降) -> 赋值追踪 (`assignments` 字典)。 ### 4.4 新版解析器 (procedure_parser.py + pipeline_bridge.py) Tier 1: 行级状态机提取嵌套结构 -> Tier 2: 规则条件解析。桥接: 新解析器优先, 旧解析器 3s 超时回退。 --- ## 5. 分支树构建细节 (core.py) ### 5.1 流程 ``` 输入: proc_text, fields 1. raw_lines = proc_text.split('\n') 2. blocked_names = 收集数据名, 阻止误匹配段落名 3. paragraphs = scan_paragraphs(raw_lines, blocked_names) 4. parser = _BrParser(filtered, paragraphs, raw_lines, assignments, fields) 5. tree = parser.parse_seq(terminators={'GOBACK', 'STOP RUN', 'EXIT PROGRAM'}) ``` ### 5.2 _BrParser 支持的语句 IF -> EVALUATE -> PERFORM -> SEARCH -> READ/WRITE -> CALL -> GO TO -> MOVE/COMPUTE/ADD/SUBTRACT/MULTIPLY/DIVIDE -> SORT/MERGE -> EXEC SQL -> INITIALIZE -> STRING/UNSTRING ### 5.3 IF 解析 提取条件 -> parse_compound_condition -> 递归 parse_seq (THEN) -> 递归 parse_seq (ELSE) -> 消费 END-IF -> 返回 BrIf + cond_tree ### 5.4 EVALUATE 解析 提取主体 -> 检测 ALSO -> 循环解析 WHEN (条件 + 递归体) -> WHEN OTHER -> 返回 BrEval --- ## 6. 条件解析 (cond.py) ### 6.1 单条件 (parse_single_condition) | 模式 | 示例 | 返回 | |------|------|------| | 标准比较 | `AMOUNT > 1000` | `('AMOUNT', '>', '1000')` | | 88 级 | `STATUS-APPROVED` | `(parent, '=', value)` | | NOT | `X NOT = 5` | `('X', '<>', '5')` | | 裸字段 | `WS-EOF` | `('WS-EOF', '=', 'Y')` | | SQLCODE | `SQLCODE = 100` | `('SQLCODE', '=', '100')` | | Class | `WS-KEY IS NUMERIC` | `('WS-KEY', 'IS', 'NUMERIC', True)` | | FUNCTION | `FUNCTION MOD(X,2) NOT = 0` | `('_FUNC_MOD', '<>', '0')` | ### 6.2 复合条件 (parse_compound_condition) 构建 CondAnd/CondOr/CondNot/CondLeaf 树。优先级: AND > OR。 ### 6.3 MC/DC (mcdc_sets) 为复合条件生成 Modified Condition/Decision Coverage 约束集。每个条件独立影响决策。 ### 6.4 满足值 (satisfying_value) 数值: 边界值 (val-1, val, val+1)。字母数字: 字符级增减。Class: 匹配/不匹配类型值。 --- ## 7. 路径枚举 (design.py) ### 7.1 核心算法 (enum_paths) 入口: `enum_paths(node, fields)` -> `list[(constraints, assignments)]` | 节点类型 | 路径生成 | |---------|---------| | Assign | 单路径, 空约束 + 赋值 | | BrSeq | 子路径笛卡尔积 + 去重 | | BrIf | True 分支 + False 分支 两组路径 | | BrEval | 每个 WHEN 一条路径 + OTHER (EVALUATE TRUE 用 MC/DC) | | BrPerform | Enter + Skip 两条路径 | | BrSearch | 每个 WHEN + AT END | | CallNode | 黑盒: 单路径 | **BrIf 简单条件:** ``` true_sub = enum_paths(true_seq) false_sub = enum_paths(false_seq) result = [(field, op, val, True) + sp for sp in true_sub] + [(field, op, val, False) + sp for sp in false_sub] ``` **BrIf 复合条件:** ``` sets = mcdc_sets(cond_tree, fields) for constraints, decision in sets: body = enum_paths(true_seq if decision else false_seq) result.append(constraints + body) ``` ### 7.2 路径截断 (_cap_paths) 最大 50,000 条路径 (`_MAX_PATHS`)。公平截断: - Phase 1: 每个前置路径至少保留一条子路径 - Phase 2: 用剩余配额填充未覆盖分支 - 哨兵路径 (STOP/ABEND) 始终保留 ### 7.3 EVALUATE TRUE 特殊处理 (eval_true_branch_constraints) - 每个 WHEN 解析为复合条件 - MC/DC 集合为每个 WHEN 生成 - `prior_false` 累积所有前序 WHEN 的 false 集 (笛卡尔积) - 每个 True 路径与所有可能的 prior-false 组合配对 --- ## 8. 值生成 (design.py) ### 8.1 基础记录生成 (make_base_record) 入口: `make_base_record(seq_num, fields)` -> `dict` 1. 遍历所有字段: - VALUE 子句 -> `_apply_value` 初始值 - 数值字段 -> `_make_numeric_value` 序列值 - 字母字段 -> `_make_alpha_value` 序列值 - 日期字段 -> `seq_date` 2. REDEFINES: 父字段值复制到重定义字段 3. 组 REDEFINES: 按位置递归复制子字段 4. 跨 FD 字段对齐: 同名不同前缀的数值字段共享 index ### 8.2 约束应用 (apply_constraint) 入口: `apply_constraint(rec, field, op, value, want_true, fields, ...)` 处理管线: 1. 变量下标解析: `WS-FIXED-VALUE(WS-IDX)` -> 具体下标 2. 下标传播: 裸字段名应用到所有下标变体 3. REDEFINES 重定向: 约束转到父字段 (共享存储) 4. 组字段展开: 组比较分解为子字段约束 5. Class 条件: `IS NUMERIC/ALPHABETIC` 通过 satisfying_value 处理 6. 字母比较: 字符级边界值 7. 数值比较: 整数边界值 8. 算术表达式: 启发式 steering 9. 零保护: 防止零值字段在 False 分支取最小值 10. 字段间协调: `WS-A >= WS-B` 同时设置两个字段 ### 8.3 赋值传播 (propagate_assignments) 模拟程序数据流到每个决策点: | Pass | 操作 | |------|------| | 1 | MOVE 传播 (源复制到目标) | | 2 | COMPUTE 求值 (算术表达式) | | 3 | ADD/SUBTRACT/MULTIPLY/DIVIDE | | 3.5 | READ INTO (文件读到工作存储) | | 4 | UNSTRING 分割 | | 5 | INITIALIZE (填充零/空格) | | 6 | STRING 拼接 | | 7 | SET TO FALSE (88 级条件名) | ### 8.4 链追溯 (trace_to_root) 沿 MOVE/COMPUTE 链回溯到源字段。返回 `(root_field, chain_of_assignments)`。用于检测不可能路径 (字面量 MOVE 与约束矛盾)。 --- ## 9. 覆盖率分析 (coverage.py) ### 9.1 决策点收集 (collect_decision_points) | 节点 | 决策点 | |------|-------| | BrIf | kind="IF", branches=["T", "F"], label=条件 | | BrEval | kind="EVALUATE", branches=["WHEN ...", "OTHER"] | | BrPerform | kind="PERFORM", branches=["Enter", "Skip"] | | BrSearch | kind="SEARCH", branches=["WHEN ...", "AT END"] | 每个决策点包含: - `id`: 自增编号 - `cond_tree`: 复合条件树 (用于 evaluate_tree) - `cond_leaves`: 叶条件列表 (用于 _match_leaf) - `active_branches`: 已覆盖的分支集合 - `implied_branches`: 推断的覆盖分支 ### 9.2 分支覆盖标记 (mark_coverage) 对每条路径的约束列表, 逐个决策点匹配: **IF 标记** (`_mark_if`): - 简单条件: 匹配 (field, op, value) 确定 T/F 分支 - 复合条件: 构建 leaf->bool assignment, 用 evaluate_tree 求值 - 合成函数 (_FUNC_*): 全部标记覆盖 **EVALUATE 标记** (`_mark_eval`): - 简单: 匹配 subject 值确定 WHEN 分支 - EVALUATE TRUE: prior_false 累积 + 逐 WHEN 匹配 **PERFORM 标记** (`_mark_perform`): - 条件为真 -> Enter, 条件为假 -> Skip **叶条件标记** (`_match_leaf`): - 去除下标后匹配 field + op + value - covered_true / covered_false 独立追踪 ### 9.3 HTML 报告生成 `generate_coverage_index()`: 中文 HTML 报告, 包含: - 覆盖率统计 (总分支数 / 已覆盖 / 未覆盖) - 每个决策点的覆盖状态 (badge: 覆盖/未覆盖) - 叶条件级别的 MC/DC 覆盖详情 --- ## 10. 输出生成 (output.py) ### 10.1 JSON 输出 (output_json) 入口: `output_json(records, outpath, roles, fd_fields, field_to_fd, open_dir, term_types, db_input, data_fields)` **输出格式:** ```json { "program": "程序名", "records": [ { "input": { "R01EMP-ID": "A0000001", ... }, "expected_output": { "W01RESULT": "PASS", ... }, "working_storage": { "WS-COUNT": "003", ... }, "termination": "normal" } ], "db_input": { ... } } ``` **字段分组逻辑:** - `input`: 方向为 INPUT/I-O 的 FD, 角色为 input/inout 的字段 - `expected_output`: 方向为 OUTPUT/I-O 的 FD, 角色为 output/inout 的字段 - `working_storage`: 不属于任何 FD 的字段 - 未分配字段 (`_assigned_fields` 集合) 的归属判定 ### 10.2 平面文件输出 (output_input_files) 将 JSON 记录写入 COBOL 输入文件 (固定长度/行顺序)。支持: - DISPLAY 格式 (文本) - COMP/COMP-3 格式 (二进制打包) - RECORDING MODE V (变长记录, 4 字节 RDW 前缀) --- ## 11. 接口定义 ### 11.1 公开 API (__init__.py) ```python def extract_structure(source: str, ...) -> dict: """解析 COBOL 控制流 -> dict (字段定义、分支树、赋值)""" def generate_data(source: str, ...) -> list[dict]: """生成测试数据 -> list[dict] (每条路径一条记录)""" def incremental_supplement(...) -> list[dict]: """差分补充数据 -> list[dict]""" ``` ### 11.2 核心模块 API **read.py:** ```python def preprocess(source: str, extra_search_paths: list[str] = None) -> str def extract_data_division(source: str) -> str def extract_procedure_division(source: str) -> str def parse_data_division(dd_text: str) -> list[FieldDef] def parse_file_section(source: str) -> dict def parse_file_control(source: str) -> dict def scan_open_statements(source: str) -> dict def scan_all_file_directions(source: str) -> dict def resolve_copybooks(source: str, base_dir: str, ...) -> str def resolve_sql_includes(source: str) -> str ``` **core.py:** ```python def build_branch_tree(proc_text: str, fields: list, full_source: str = None) -> (BrSeq, dict) def scan_paragraphs(raw_lines: list, blocked_names: set = None) -> dict def sql_register_virtual_fields(fields_dict: list[dict]) -> list[dict] def classify_field_roles(fields, assignments, file_sec, ...) -> dict ``` **cond.py:** ```python def parse_single_condition(text: str, fields: list = None) -> tuple | None def parse_compound_condition(text: str, fields: list = None) -> CondAnd | CondOr | CondLeaf | CondNot | None def collect_leaves(tree) -> list[CondLeaf] def evaluate_tree(tree, assignment: dict) -> bool def is_field(name: str, fields: list) -> bool def mcdc_sets(tree, fields: list = None) -> list | None def satisfying_value(pic_info: dict, op: str, value: str, want_true: bool) -> str def merge_field_constraints(cons_list: list) -> list ``` **design.py:** ```python def enum_paths(node, fields: list) -> list[(list, dict)] def make_base_record(seq_num: int, fields: list) -> dict def apply_constraint(rec: dict, field_name: str, op: str, value: str, want_true: bool, fields: list, ...) -> None def propagate_assignments(rec: dict, assignments: dict, fields: list, ...) -> None def trace_to_root(field: str, assignments: dict, fields: list, path_assign: dict) -> (str, list) def generate_records(path_infos: list, data_fields: list, ...) -> (list, list, list) def get_term_type(cons: list) -> (list, str) def extend_abend_programs(names: list[str]) -> None ``` **coverage.py:** ```python def collect_decision_points(node, fields: list) -> (list[DecisionPoint], list[LeafStat]) def mark_coverage(decision_points: list, leaf_stats: list, branch_paths: list, fields: list) -> None def run_coverage(source: str, records: list, ...) -> dict def generate_coverage_index(...) -> str # HTML string ``` **output.py:** ```python def output_json(records: list, outpath: Path, ...) -> None def output_input_files(records: list, outpath: Path, ...) -> None ``` **to_sql.py:** ```python def collect_sql_meta(source: str, fields: list) -> list[dict] def build_db_input(sql_meta: list, records: list, ...) -> dict ``` **flatfile.py:** ```python def analyze_fd_layout(source_text: str, ...) -> dict[str, dict] def write_flat_file(records: list, layout: dict, outpath: Path) -> None ``` --- ## 12. 错误处理 ### 12.1 解析错误 (ParseError) `models.py` 定义 `ParseError` 数据类: - `line: int` -- 错误所在行号 - `message: str` -- 错误描述 - `severity: str` -- 'warning' 或 'error' `ProcParseResult.errors` 收集解析过程中产生的所有错误。 ### 12.2 处理策略 | 错误类型 | 处理方式 | |---------|---------| | COPYBOOK 找不到 | 跳过该 COPY, 继续解析, 记录 warning | | Lark 语法不匹配 | 回退到规则引擎, 标记 `fallback_to_ai=True` | | 未知 COBOL 语句 | 跳过该行, 不中断解析 | | 字段未找到 | 使用默认值, 记录 debug 日志 | | 条件解析失败 | 返回 None, 由上层决定回退策略 | | 路径枚举超过上限 | `_cap_paths` 公平截断到 50,000 条 | | 新解析器超时/失败 | `pipeline_bridge` 回退到旧解析器 (3s 超时) | | 旧解析器超时/失败 | 返回空 BrSeq + 空 assignments | | gcov 执行失败 | 返回空 dict, 跳过覆盖率标记 | | 外部调用异常 | try/except 捕获, logger.warning 记录, 不中断主流程 | ### 12.3 日志策略 所有模块使用 Python `logging`: - `logger.info()` -- 关键流程节点 (解析开始/结束, 路径数, 记录数) - `logger.debug()` -- 详细调试信息 (约束应用, 赋值传播) - `logger.warning()` -- 非致命错误 (解析失败, 回退) - `logger.error()` -- 致命错误 (不应发生) ### 12.4 已知限制 1. **MC/DC 复合 IF bug**: `merge_field_constraints` 合并同字段约束会破坏 `_match_leaf` 匹配 2. **OCCURS DEPENDING ON**: 捕获但未用于记录生成 3. **88 级 VALUE**: 目标程序无 88 级 VALUE 子句, 解析器不依赖此特性 4. **路径多样性丢失**: 规则引擎 100 路径限制 (LLM 模式 50000 路径不受影响) 5. **合成函数字段**: `_FUNC_MOD` 等合成名 is_field 返回 False, 叶条件不可匹配