hangshuo652/cobol-java-v3

Fork 0

Files

T

hangshuo652 d1f3d69445 docs: add V3 detailed design documents, AI usage log, and development paradigm

2026-08-22 18:05:24 +08:00

21 KiB

Raw Blame History

cobol_testgen 核心引擎模块 - 详细设计文档

模块路径: cobol_testgen/ 版本: V3 (2026技术大赛) 依赖: Python 3.13+, lark>=1.1.0

1. 模块概述

1.1 职责

cobol_testgen 是 COBOL 迁移验证平台 V3 的核心引擎, 负责:

解析 COBOL 源码 (DATA DIVISION + PROCEDURE DIVISION)
构建分支控制流树 (Branch Tree)
枚举所有可达路径 (路径级约束)
生成满足路径约束的测试数据记录
输出 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 字段定义

@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 分支树节点

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 条件树

class CondLeaf:   field, op, value     # 叶条件
class CondNot:    child                 # NOT 取反
class CondAnd:    left, right           # AND
class CondOr:     left, right           # OR

3.4 其他节点

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 约束与路径

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)

COPYBOOK 展开 -> EXEC SQL/CICS 移除 -> VALUE 逗号清理
& 连接行合并 -> PIC 小数点转换 -> 格式检测 (fixed/free)
注释移除 -> 续行合并

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

遍历所有字段:
- VALUE 子句 -> _apply_value 初始值
- 数值字段 -> _make_numeric_value 序列值
- 字母字段 -> _make_alpha_value 序列值
- 日期字段 -> seq_date
REDEFINES: 父字段值复制到重定义字段
组 REDEFINES: 按位置递归复制子字段
跨 FD 字段对齐: 同名不同前缀的数值字段共享 index

8.2 约束应用 (apply_constraint)

入口: apply_constraint(rec, field, op, value, want_true, fields, ...)

处理管线:

变量下标解析: WS-FIXED-VALUE(WS-IDX) -> 具体下标
下标传播: 裸字段名应用到所有下标变体
REDEFINES 重定向: 约束转到父字段 (共享存储)
组字段展开: 组比较分解为子字段约束
Class 条件: IS NUMERIC/ALPHABETIC 通过 satisfying_value 处理
字母比较: 字符级边界值
数值比较: 整数边界值
算术表达式: 启发式 steering
零保护: 防止零值字段在 False 分支取最小值
字段间协调: 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)

输出格式:

{
  "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)

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:

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:

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:

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:

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:

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:

def output_json(records: list, outpath: Path, ...) -> None
def output_input_files(records: list, outpath: Path, ...) -> None

to_sql.py:

def collect_sql_meta(source: str, fields: list) -> list[dict]
def build_db_input(sql_meta: list, records: list, ...) -> dict

flatfile.py:

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 已知限制

MC/DC 复合 IF bug: merge_field_constraints 合并同字段约束会破坏 _match_leaf 匹配
OCCURS DEPENDING ON: 捕获但未用于记录生成
88 级 VALUE: 目标程序无 88 级 VALUE 子句, 解析器不依赖此特性
路径多样性丢失: 规则引擎 100 路径限制 (LLM 模式 50000 路径不受影响)
合成函数字段: _FUNC_MOD 等合成名 is_field 返回 False, 叶条件不可匹配

21 KiB Raw Blame History