chore: SETUP.md + 测试报告脚本 + 文档更新

- SETUP.md: 完整环境搭建指南(同事用)
- SETUP_QUICK.md: 快速搭环境(4步)
- s22~s26: TNA端到端、覆盖率报告、回归检查
- procedure_grammar.lark: 实验性Lark语法

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
NB-076
2026-06-25 08:50:17 +08:00
parent 56d1cf5e78
commit 50995d3335
25 changed files with 6861 additions and 0 deletions
+1626
View File
@@ -0,0 +1,1626 @@
# COBOL 迁移验证平台 — 增强测试 实施计划
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** 在现有 `v3-gstack-code-gen` 管线中集成 cobol_testgen 规则引擎路径覆盖、HINA 程序分类、质量门禁、覆盖率交叉验证和分层重试。
**Architecture:** 不改 runners/comparator/web/worker 等稳定模块,通过修改 orchestrator.py(约30行)和新增 `hina/` 包(约1500行)、封装 `cobol_testgen` API(约50行)实现。分4个 Phase 渐进交付。
**Phase 依赖关系:**
- Phase 1 独立可交付(cobol_testgen 集成 + 分层重试)
- Phase 2 依赖 Phase 1cobol_testgen 的 extract_structure 输出)
- Phase 3 依赖 Phase 1(编译运行接口)
- Phase 4 依赖 Phase 1 的覆盖率数据 + Phase 2 的 HINA 数据 + Phase 3 的 gcov 数据
- Phase 4 可在 Phase 2/3 完成前部分实施(HINA/质量评分显示"待集成"
**Tech Stack:** Python 3.11+、FastAPI、pytest、GnuCOBOL、Lark(已有 cobol_testgen 依赖)
---
## 文件结构
### 新增文件
| 文件 | 职责 | 行数估计 | Phase |
|:-----|:------|:--------:|:-----|
| `hina/__init__.py` | 包初始化 | 5 | 2 |
| `hina/classifier.py` | HINA Agent 调用 + L1关键字识别 + 确信度计算 | 300 | 2 |
| `hina/strategy.py` | 策略模板 + 策略 Agent 调用 | 200 | 2 |
| `hina/gate.py` | 质量门禁(决策点/段落/HINA必须项/字段覆盖) | 300 | 2 |
| `hina/gcov_collector.py` | gcov 解析 + 降级逻辑 | 150 | 3 |
| `hina/retry.py` | 分层重试(heal_retry/simple_retry | 100 | 1 |
### 修改文件
| 文件 | 修改内容 | 变更量 | Phase |
|:-----|:---------|:------:|:-----|
| `cobol_testgen/__init__.py` | 新增 `extract_structure()`, `generate_data()`, `incremental_supplement()` 三个 API | +50行 | 1 |
| `cobol_testgen/coverage.py` | 封装 `check_coverage()` 为可调用 API | +20行 | 1 |
| `orchestrator.py` | Agent2 一步→替换为 while 循环流程 | ~30行 | 1 |
| `config/__init__.py` | 新增 `max_retries`, `quality_gate_mode`, `gcov_enabled` 等配置项 | +10行 | 1 |
| `data/diff_result.py` | `VerificationRun` 增加 coverage/quality/hina 字段 | +10行 | 1 |
| `data/test_case.py` | `TestCase` 增加 `hina_type`, `coverage_meta` 字段 | +5行 | 2 |
| `runners/cobol_runner.py` | 可选 gcov 编译参数 | +5行 | 3 |
| `report/generator.py` | 覆盖率/HINA/质量评分/重试历史卡片 | +80行 | 4 |
| `agents/agent2_data.py` | Phase 2 替换为调用 hina/strategy.py | ~25行 | 2 |
| `main.py` | 新增 `--quality-gate-mode`, `--gcov` 参数 | +10行 | 1 |
| `aurak.toml` | 新增 quality_gate 节 | +5行 | 1 |
### 不变的文件
`runners/*`cobol_runner.py 仅加编译参数)、`comparator/*``web/*``worker.py``agents/agent1_parser.py``agents/agent3_diagnostic.py``jcl/*``tests/*`
---
## Phase 1: cobol_testgen 集成 + 分层重试(P0
### Task 1.1: cobol_testgen 新增 API 入口
**Files:**
- Modify: `cobol_testgen/__init__.py`
- Modify: `cobol_testgen/coverage.py`
- [ ] **Step 1: 在 `cobol_testgen/__init__.py` 底部新增 `extract_structure()` 函数**
```python
# 添加到 cobol_testgen/__init__.py 底部,在 main() 之前
def extract_structure(cobol_source: str) -> dict:
"""
分析 COBOL 源码的结构,返回结构摘要。
不生成测试数据,只做静态分析。
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)
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(proc_div, fields_dict)
file_sec = parse_file_section(preprocessed)
open_dir = scan_open_statements(proc_div) if proc_div else {}
# 统计决策点
decision_points = []
total_branches = 0
def _walk(node, counter):
nonlocal total_branches
from .models import BrIf, BrEval, BrPerform
if isinstance(node, BrIf):
counter[0] += 1
branches = 2
decision_points.append({
"id": counter[0], "kind": "IF",
"label": 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
n = len(node.when_list) + (1 if node.has_other 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)
if branch_tree:
_walk(branch_tree, [0])
# OCCURS 展开前统计段落数
lines = proc_div.split('\n') if proc_div else []
paragraphs = set()
for line in lines:
import re
m = re.match(r'^\s*([A-Z0-9][A-Z0-9-]*)\.\s*$', line.strip())
if m:
paragraphs.add(m.group(1))
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,
}
```
- [ ] **Step 2: 在 `cobol_testgen/__init__.py` 底部新增 `generate_data()` 函数**
```python
def generate_data(cobol_source: str, structure: dict = None) -> list[dict]:
"""
根据 COBOL 源码生成覆盖所有路径的测试数据。
Args:
cobol_source: COBOL 程序源码文本
structure: 可选,如果已调用 extract_structure() 可传入避免重复解析
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 []
preprocessed = preprocess(cobol_source)
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(proc_div, fields_dict)
file_sec = parse_file_section(preprocessed)
from .design import enum_paths, generate_records, _filter_stop
branch_paths = enum_paths(branch_tree, fields_dict)
branch_paths = [(_filter_stop(c), a) for c, a in branch_paths]
records, kept_paths = generate_records(branch_paths, fields_dict, assignments, file_sec=file_sec)
return records
```
- [ ] **Step 3: 在 `cobol_testgen/__init__.py` 底部新增 `incremental_supplement()` 函数**
```python
def incremental_supplement(branch_tree, decision_gaps: list[int]) -> list[dict]:
"""
针对未覆盖的决策点,增量生成补充测试数据。
不重新枚举所有路径,只针对指定的决策点 ID 生成数据。
Args:
branch_tree: extract_structure() 返回的 branch_tree 字段
decision_gaps: 未覆盖的决策点 ID 列表,如 [1, 3, 5]
Returns:
list[dict]: 增量测试数据(覆盖缺失的决策点)
"""
# 遍历分支树,找到指定 ID 的决策点
# 为该决策点的每个未覆盖分支生成一条简单记录
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])
# 为每个缺失的决策点生成一条记录,格式与 generate_data() 兼容
supplements = []
for i, (kind, label) in enumerate(found):
supplements.append({
"_dec_id": f"incr_{i}",
"_kind": kind,
"_label": str(label)[:60],
})
return supplements
```
- [ ] **Step 4: 封装 `coverage.py` 的 `check_coverage()` 为可调用 API**
`cobol_testgen/coverage.py` 底部新增:
```python
# 添加到 coverage.py 底部
def check_coverage(structure: dict, test_records: list[dict]) -> dict:
"""
报告 COBOL 源码的静态分支结构信息。
注意: 静态分析无法精确判断每条测试数据运行时覆盖了哪些分支。
精确的路径追踪依赖 gcov(Phase 3)。
此处仅报告总分支数和记录生成情况,不做虚假的"已覆盖"估算。
Args:
structure: extract_structure() 返回的结构摘要
test_records: generate_data() 返回的测试数据列表
Returns:
dict with: paragraph_rate, branch_rate, decision_rate,
uncovered_decision_ids, total_branches, total_paragraphs,
records_count
"""
total_paragraphs = structure.get("total_paragraphs", 0)
total_branches = structure.get("total_branches", 0)
decision_points = structure.get("decision_points", [])
# 有测试数据 = 覆盖率有机会 > 0(但不保证覆盖了所有分支)
# 精确覆盖率需要 gcov 运行时数据
has_data = len(test_records) > 0
# 段落: 有数据就假设有机会覆盖(保守估计)
paragraph_rate = 1.0 if (total_paragraphs > 0 and has_data) else 0.0
return {
"paragraph_rate": paragraph_rate,
"branch_rate": 0.0,
"decision_rate": 0.0,
"uncovered_decision_ids": [],
"total_branches": total_branches,
"total_paragraphs": total_paragraphs,
"records_count": len(test_records),
"note": "静态分析无法精确计算覆盖率。精确数据通过 gcov 获取(Phase 3)。",
}
```
- [ ] **Step 5: 运行 import 测试确认封装正确**
Run: `cd D:/cobol-java/v3-gstack-code-gen && python -c "from cobol_testgen import extract_structure, generate_data, incremental_supplement; print('API OK')"`
Expected: `API OK`
- [ ] **Step 6: Commit**
```bash
git add cobol_testgen/__init__.py cobol_testgen/coverage.py
git commit -m "feat: expose extract_structure/generate_data/incremental_supplement APIs from cobol_testgen"
```
---
### Task 1.2: VerificationRun 增加覆盖率字段
**Files:**
- Modify: `data/diff_result.py`
- [ ] **Step 1: `VerificationRun` 增加覆盖率/质量门禁字段**
```python
# 在 data/diff_result.py 的 VerificationRun 类中增加字段
# 修改后:
@dataclass
class VerificationRun:
program: str = ""
timestamp: str = ""
status: str = "PASS"
exit_code: int = 0
duration_s: float = 0.0
fields_matched: int = 0
fields_mismatched: int = 0
coverage_target: str = "boundary"
field_results: list[FieldResult] = field(default_factory=list)
runner: str = "native"
branch_rate: float = 0.0
paragraph_rate: float = 0.0 # 新增: 段落覆盖率
decision_rate: float = 0.0 # 新增: 决策点覆盖率
hina_type: str = "" # 新增: HINA 类型 (Phase 2 启用)
hina_confidence: float = 0.0 # 新增: HINA 确信度
quality_score: float = 0.0 # 新增: 质量评分
quality_warn: str = "" # 新增: 质量警告信息
heal_retry: int = 0 # 新增: 自愈重试次数
simple_retry: int = 0 # 新增: 朴素重试次数
total_retry: int = 0 # 新增: 总重试次数
llm_cost: float = 0.0
report_path: str = ""
debug: dict = field(default_factory=dict)
```
- [ ] **Step 2: 运行测试确认不破坏现有代码**
Run: `cd D:/cobol-java/v3-gstack-code-gen && python -c "from data.diff_result import VerificationRun; vr = VerificationRun(); print(vr.paragraph_rate, vr.quality_score)"`
Expected: `0.0 0.0`
- [ ] **Step 3: Commit**
```bash
git add data/diff_result.py
git commit -m "feat: add coverage/quality fields to VerificationRun"
```
---
### Task 1.3: Config 增加质量门禁配置
**Files:**
- Modify: `config/__init__.py`
- [ ] **Step 1: `Config` 增加质量门禁相关配置**
```python
# 在 Config dataclass 中增加字段:
@dataclass
class Config:
# ... 原有字段保持不变 ...
branch_pass: float = 0.80
# 以下为新增字段:
quality_gate_mode: str = "warn" # "warn" | "off" — 是否阻断管道
quality_gate_decision_threshold: float = 0.90 # Phase 1 决策点覆盖率 ≥90%
quality_gate_paragraph_threshold: float = 1.0 # 段落覆盖率 100%
gcov_enabled: bool = False # 是否启用 gcov
max_quality_retries: int = 4 # 质量门禁循环最大次数
```
- [ ] **Step 2: 更新 `aurak.toml` 增加 quality_gate 配置节**
`aurak.toml` 底部追加:
```toml
[quality_gate]
mode = "warn" # "warn" | "off"
decision_threshold = 0.90
paragraph_threshold = 1.0
[gcov]
enabled = false
```
- [ ] **Step 3: 确认 Config 向后兼容**
Run: `cd D:/cobol-java/v3-gstack-code-gen && python -c "from config import Config; c = Config(); print(c.quality_gate_mode, c.quality_gate_decision_threshold)"`
Expected: `warn 0.9`
- [ ] **Step 4: Commit**
```bash
git add config/__init__.py aurak.toml
git commit -m "feat: add quality gate config fields"
```
---
### Task 1.4: 修改 orchestrator.py 插入循环流程
**Files:**
- Modify: `orchestrator.py`
- [ ] **Step 1: 在 `run_pipeline()` 中插入 extract_structure + generate_data + 质量门禁循环**
修改 `orchestrator.py`,在 `suite = Agent2(llm).design(...)` 前面插入 cobol_testgen 步骤:
```python
# 在 orchestrator.py 顶部增加 import
from cobol_testgen import extract_structure, generate_data, incremental_supplement
from cobol_testgen.coverage import check_coverage
# 在 run_pipeline() 函数中,Agent1 之后、Agent2 之前插入(约第 43 行前后):
def run_pipeline(cfg: Config, cpath: str, cbl: str, java: str, map_path: str) -> VerificationRun:
t0 = time.time()
vr = VerificationRun(program=Path(java).stem, runner=cfg.runner_mode)
try:
text = Path(cpath).read_text()
if not text.strip():
return _done(vr, t0, "BLOCKED", 2)
llm = LLMClient(model=cfg.llm_model, timeout=cfg.llm_timeout, cache_dir=cfg.llm_cache_dir)
tree = Agent1Parser(llm).parse(text)
vr.llm_cost += 0.002
vr.debug["field_tree"] = [
{"name": f.name, "level": f.level, "pic": f.pic,
"usage": f.usage, "offset": f.offset, "length": f.length,
"redefines": f.redefines}
for f in tree.flatten().values()
]
if not tree.fields:
return _done(vr, t0, "BLOCKED", 2)
if vr.llm_cost > cfg.max_llm_cost:
return _done(vr, t0, "BLOCKED", 3)
# ── Phase 1: cobol_testgen 结构提取 + 路径覆盖 + 质量门禁循环 ──
try:
cobol_src_text = Path(cbl).read_text(encoding='utf-8')
structure = extract_structure(cobol_src_text)
base_records = generate_data(cobol_src_text, structure)
vr.debug["cobol_testgen_records"] = len(base_records)
vr.debug["total_branches"] = structure.get("total_branches", 0)
# 质量门禁循环(只做增量补充,不重跑 generate_data
from hina.gate import check as gate_check
complete_tests = list(base_records) # Phase 1 使用基础数据
coverage = check_coverage(structure, complete_tests)
for attempt in range(cfg.max_quality_retries):
gate_result = gate_check(complete_tests, {}, coverage,
decision_threshold=cfg.quality_gate_decision_threshold,
paragraph_threshold=cfg.quality_gate_paragraph_threshold)
if gate_result["passed"]:
break
gaps = gate_result.get("issues", {}).get("decision_gaps", [])
if gaps:
delta = incremental_supplement(structure.get("branch_tree_obj"), gaps)
complete_tests.extend(delta)
else:
break
vr.paragraph_rate = coverage.get("paragraph_rate", 0.0)
vr.branch_rate = coverage.get("branch_rate", 0.0)
vr.decision_rate = coverage.get("decision_rate", 0.0)
if cfg.quality_gate_mode != "off" and not gate_result["passed"]:
vr.quality_warn = f"质量门禁未完全通过(尝试{attempt+1}次)"
vr.debug["quality_issues"] = gate_result["issues"]
except Exception as e:
vr.debug["cobol_testgen_error"] = str(e)
logger.warning(f"[orchestrator] cobol_testgen 分析失败: {e}")
# ── 原有 Agent2 保持不变 ──
suite = Agent2(llm).design(tree, cfg.coverage_default, cfg.runner_mode == "spark")
vr.llm_cost += 0.002
vr.debug["test_cases"] = [{"id":tc.id,"fields":tc.fields,"targets":tc.coverage_targets} for tc in suite.test_cases]
# ... 后续代码保持不变 ...
```
- [ ] **Step 2: 运行测试确认 import 正确**
Run: `cd D:/cobol-java/v3-gstack-code-gen && python -c "from orchestrator import run_pipeline; print('import OK')"`
Expected: `import OK`
- [ ] **Step 3: Commit**
```bash
git add orchestrator.py
git commit -m "feat: integrate cobol_testgen path coverage into pipeline"
```
---
### Task 1.5: 分层重试 retry.py
**Files:**
- Create: `hina/__init__.py`
- Create: `hina/retry.py`
- [ ] **Step 1: 创建 `hina/__init__.py`**
```python
# hina/__init__.py
# HINA 程序分类与质量门禁包
```
- [ ] **Step 2: 创建 `hina/retry.py`**
```python
# hina/retry.py
"""
分层重试 — 部署在 orchestrator 调用者层(main.py / worker.py)。
用法:
handler = RetryHandler(max_heal=2, max_simple=3)
vr = handler.run(lambda: run_pipeline(cfg, ...))
"""
import logging
from typing import Callable, Optional
from data.diff_result import VerificationRun
logger = logging.getLogger(__name__)
# 已知失败模式与修复策略
# 注意: 自动修复的实际效果有限——环境问题(如 COBCPY 路径)需要人工配置。
# 自动修复的目的是在重试前做一次可做的尝试,而非保证修复成功。
HEALING_FIXES = {
"compile_error": {
"detect": lambda log: "not found" in (log or "").lower(),
"fix": lambda: _try_set_env("COB_LIBRARY_PATH",
"D:\\360安全浏览器下载\\GC32-BDB-SP1-rename-7z-to-exe\\lib\\gnucobol"),
},
"s0c7": {
"detect": lambda log: "S0C7" in (log or ""),
"fix": lambda: logger.warning("[Retry] S0C7 需要人工修正测试数据中的数值字段"),
},
}
def _try_set_env(key: str, value: str) -> None:
"""尝试设置环境变量(如果当前未设置)"""
import os
if not os.environ.get(key):
os.environ[key] = value
logger.info(f"[Retry] 已设置环境变量 {key}={value}")
else:
logger.info(f"[Retry] {key} 已存在,跳过")
class RetryHandler:
def __init__(self, max_heal: int = 2, max_simple: int = 3):
self.max_heal = max_heal
self.max_simple = max_simple
self.heal_count = 0
self.simple_count = 0
self.history: list[VerificationRun] = []
def run(self, pipeline_fn: Callable[[], VerificationRun]) -> VerificationRun:
while (self.heal_count + self.simple_count) < (self.max_heal + self.max_simple):
vr = pipeline_fn()
self.history.append(vr)
if vr.status == "PASS" or vr.status == "QUALITY_WARN":
# PASS 或 QUALITY_WARN 不阻断
vr.heal_retry = self.heal_count
vr.simple_retry = self.simple_count
vr.total_retry = self.heal_count + self.simple_count
return vr
if vr.status in ("BLOCKED", "ERROR") and self.heal_count < self.max_heal:
# 尝试自愈
build_log = vr.debug.get("cobol_build", {}).get("log", "")
healed = False
for name, fix_def in HEALING_FIXES.items():
if fix_def["detect"](build_log):
fix_def["fix"]()
self.heal_count += 1
healed = True
logger.info(f"[Retry] 自愈修复应用: {name} (heal_retry={self.heal_count})")
break
if healed:
continue
# 朴素重试
self.simple_count += 1
logger.info(f"[Retry] 朴素重试 (simple_retry={self.simple_count})")
# 超过上限
logger.error("[Retry] 重试次数超过上限,标记 FATAL")
vr = self.history[-1] if self.history else VerificationRun(status="FATAL", exit_code=4)
vr.status = "FATAL"
vr.exit_code = 4
vr.heal_retry = self.heal_count
vr.simple_retry = self.simple_count
vr.total_retry = self.heal_count + self.simple_count
return vr
```
- [ ] **Step 3: 测试 retry 模块**
Run: `cd D:/cobol-java/v3-gstack-code-gen && python -c "from hina.retry import RetryHandler; print('OK')"`
Expected: `OK`
- [ ] **Step 4: Commit**
```bash
git add hina/__init__.py hina/retry.py
git commit -m "feat: add layered retry handler"
```
---
### Task 1.6: main.py 增加质量门禁参数
**Files:**
- Modify: `main.py`
- [ ] **Step 1: main.py 增加 `--quality-gate-mode` 和 `--gcov` 参数**
```python
# 在 main.py 的 ArgumentParser 中增加参数(约第 14 行):
p.add_argument("--quality-gate-mode", choices=["warn", "off"], default="warn",
help="质量门禁模式: warn=记录警告, off=关闭")
p.add_argument("--gcov", action="store_true", help="启用 gcov 覆盖率采集")
# 在 run_pipeline 调用前应用配置:
c.quality_gate_mode = args.quality_gate_mode
c.gcov_enabled = args.gcov
```
- [ ] **Step 2: Commit**
```bash
git add main.py
git commit -m "feat: add --quality-gate-mode and --gcov CLI args"
```
---
## Phase 2: HINA Agent + 策略 AgentP1
### Task 2.1: HINA 确信度计算(纯函数)
**Files:**
- Create: `hina/classifier.py`
- [ ] **Step 1: 创建 `hina/classifier.py` 确信度函数**
```python
# hina/classifier.py
"""
HINA 程序类型分类器。
三层判定:
L1 关键字识别 — 11 类可直接通过关键字判定的类型
L2 结构提取 — 从 cobol_testgen 结构摘要提取特征(为 L3 提供输入)
L3 混淆组判定 — 调用 LLM Agent 解决 8 个混淆组
确信度计算: 確信度 = 基礎確信度 × 上下文因子 × 一致性因子 × 構造一致性因子
"""
# L1 关键字识别规则
L1_RULES = [
("DB操作", ["EXEC SQL"], 0.95),
("子程序调用", ["CALL", "LINKAGE SECTION"], 0.90),
("IS INITIAL", ["IS INITIAL"], 0.99),
("SYSIN", ["SYSIN"], 0.90),
("编码转换", ["ALPHABETIC", "ASCII", "EBCDIC"], 0.85),
("online", ["DFHCOMMAREA", "MAP"], 0.95),
("SORT", ["SORT ON KEY"], 0.95),
("MERGE", ["MERGE ON KEY"], 0.95),
("编辑输出", ["WRITE AFTER", "WRITE BEFORE"], 0.80),
("文件编成", ["ORGANIZATION IS"], 0.99),
("替代索引", ["ALTERNATE RECORD KEY"], 0.99),
]
# 矛盾对优先级规则(用于一致性因子)
CONFLICT_RULES = {
("マッチング", "キーブレイク"): "file_count",
("編集処理", "項目チェック"): "file_count",
("キーブレイク", "項目チェック(重複)"): "has_accumulator",
}
def detect_keyword(source: str) -> list[tuple[str, float, str]]:
"""
L1 关键字识别。
Returns: [(category, confidence, matched_keyword), ...]
"""
source_upper = source.upper()
results = []
for category, keywords, base_confidence in L1_RULES:
matched = [kw for kw in keywords if kw in source_upper]
if matched:
factor = min(1.0, 0.9 + 0.05 * len(matched))
results.append((category, base_confidence * factor, matched[0]))
return results
def compute_confidence(
source: str,
structure: dict,
llm_result: dict = None,
) -> dict:
"""
确信度计算(纯函数)。
確信度 = 基礎確信度 × 上下文因子 × 一致性因子 × 構造一致性因子
Args:
source: COBOL 源码文本
structure: extract_structure() 输出
llm_result: LLM Agent 的混淆组判定结果
Returns:
dict with: category, subtype, confidence, method, features,
required_tests, strategy_params
"""
keywords = detect_keyword(source)
total_features = []
# 从 structure 提取特征
if structure:
if structure.get("file_count", 0) >= 2:
total_features.append("多ファイル入力")
if structure.get("has_search_all"):
total_features.append("SEARCH ALL")
if structure.get("has_evaluate"):
total_features.append("EVALUATE")
if structure.get("has_break"):
total_features.append("KEY BREAK")
# 如果有 L1 关键字命中且确信度足够,直接判定
if keywords:
best = max(keywords, key=lambda x: x[1])
if best[1] >= 0.90:
return {
"category": best[0],
"subtype": "general",
"confidence": round(best[1], 2),
"method": "keyword",
"features": [best[2]] + total_features[:2],
"required_tests": [],
"strategy_params": {
"special_boundaries": [],
"coverage_requirements": {"branch": 0.95, "paragraph": 1.0},
},
}
# 混合 LLM 结果判定(在 hina_agent.py 中调用)
if llm_result:
category = llm_result.get("category", "unknown")
confidence = llm_result.get("confidence", 0.5)
return {
"category": category,
"subtype": llm_result.get("subtype", "general"),
"confidence": round(confidence, 2),
"method": "hybrid",
"features": llm_result.get("features", total_features),
"required_tests": llm_result.get("required_tests", []),
"strategy_params": llm_result.get("strategy_params", {
"special_boundaries": [],
"coverage_requirements": {"branch": 0.95, "paragraph": 1.0},
}),
}
# 默认: 无法判定
return {
"category": "unknown",
"subtype": "general",
"confidence": 0.0,
"method": "none",
"features": total_features,
"required_tests": [],
"strategy_params": {
"special_boundaries": [],
"coverage_requirements": {"branch": 0.95, "paragraph": 1.0},
},
}
```
- [ ] **Step 2: 编写确信度函数测试**
```python
# tests/test_quality/test_classifier.py
from hina.classifier import detect_keyword, compute_confidence
def test_detect_keyword():
source = "PROCEDURE DIVISION.\nEXEC SQL SELECT * FROM TABLE END-EXEC."
results = detect_keyword(source)
assert any("DB操作" in r[0] for r in results)
def test_detect_keyword_no_match():
source = "PROCEDURE DIVISION.\nDISPLAY 'HELLO'."
results = detect_keyword(source)
assert len(results) == 0
```
- [ ] **Step 3: 运行测试**
Run: `cd D:/cobol-java/v3-gstack-code-gen && python -c "from hina.classifier import detect_keyword; print('OK')"`
Expected: `OK`
- [ ] **Step 4: Commit**
```bash
git add hina/classifier.py
git commit -m "feat: add HINA classifier with keyword detection and confidence calculation"
```
---
### Task 2.2: 策略模板
**Files:**
- Create: `hina/strategy.py`
- [ ] **Step 1: 创建策略模板**
```python
# hina/strategy.py
"""
HINA 类型策略模板。
每种类型对应一组必须覆盖的测试项(来自 cobol-test-benchmark.md 第2部)。
策略 Agent 根据类型选择模板,补充测试数据。
"""
# 5 种优先类型的必须项
STRATEGY_TEMPLATES = {
"マッチング": {
"required": [
"COM-N001: 最小データ1件",
"COM-N002: 標準データ複数件",
"COM-A002: 全ファイル空",
"COM-A003: 一部ファイル空",
"MT-N001: 1:1 主キー完全一致",
"MT-N002: 1:N 主1件従N件",
"MT-N004: 主件剩余キー",
"MT-N005: 従件剩余キー",
"MT-N006: 主キー値重複",
],
"special_boundaries": [
"不平衡: 主1件 vs 従100万件",
"不平衡: 主100万件 vs 従1件",
],
},
"キーブレイク": {
"required": [
"COM-N001: 最小データ1件",
"COM-A002: 全ファイル空",
"KB-N001: ADD累加正確",
"KB-N004: 単一キー郡",
"KB-N005: 複数キー郡",
"KB-A001: 前キー値未初期化",
],
"special_boundaries": [
"キー変化系列: 同キー3件→切替→同キー2件",
"ファイル終了時最終累積値出力",
],
},
"条件分岐": {
"required": [
"B-N001: IF 2路分岐",
"B-N003: IF 複合条件 AND/OR",
"B-N006: EVALUATE WHEN 複数値",
"B-N009: EVALUATE WHEN OTHER",
],
"special_boundaries": [],
},
"内部表検索": {
"required": [
"T-N001: SEARCH ALL 等値查找(見つかる)",
"T-N002: SEARCH ALL 等値查找(見つからない)",
"T-A001: SEARCH ALL 未ソート表",
"T-A002: INDEX 越界",
],
"special_boundaries": [],
},
"項目チェック": {
"required": [
"VF-N001: 字段校验通過",
"VF-N002: 字段校验拒否",
"VF-N004: 重複検出(重複)",
"VF-A001: 半角超長(21桁)",
],
"special_boundaries": [],
},
}
def get_strategy(hina_type: str) -> dict:
"""根据 HINA 类型返回策略模板"""
return STRATEGY_TEMPLATES.get(hina_type, {
"required": ["COM-N001", "COM-A002"],
"special_boundaries": [],
})
def supplement(base_tests: list[dict], hina_result: dict) -> list[dict]:
"""
根据 HINA 类型向基础数据追加类型特有的边界测试记录。
当前实现: 为模板中的每个必需项和特殊边界生成一条标记记录。
Phase 2 将由 LLM 驱动,生成语义化的测试值。
"""
result = list(base_tests)
hina_type = hina_result.get("category", "unknown")
template = STRATEGY_TEMPLATES.get(hina_type, {})
for req in template.get("required", []):
result.append({
"_strategy": req.split(":")[0].strip(),
"_note": req,
})
for boundary in template.get("special_boundaries", []):
result.append({
"_strategy": "boundary",
"_note": boundary,
})
return result
def supplement_only(base_tests: list[dict], hina_gaps: list[str]) -> list[dict]:
"""
增量补充指定必须项的测试数据。
只生成标记记录,具体字段值由 LLM/人工填充。
"""
supplements = []
for gap_id in hina_gaps:
supplements.append({
"_strategy": "hina_gap",
"_hina_gap_id": gap_id,
})
return supplements
```
- [ ] **Step 2: 测试策略模板**
Run: `cd D:/cobol-java/v3-gstack-code-gen && python -c "from hina.strategy import get_strategy; s = get_strategy('マッチング'); print(len(s['required']))"`
Expected: `9`
- [ ] **Step 3: Commit**
```bash
git add hina/strategy.py
git commit -m "feat: add HINA strategy templates for 5 priority types"
```
---
### Task 2.3: 质量门禁 gate.py
**Files:**
- Create: `hina/gate.py`
- [ ] **Step 1: 创建质量门禁**
```python
# hina/gate.py
"""
质量门禁 — 执行前检查测试数据是否满足覆盖率和边界要求。
Phase 1 可用: 决策点覆盖、段落覆盖
Phase 2 启用: HINA 必须项、字段覆盖
"""
def check(
complete_tests: list[dict],
hina_result: dict,
coverage: dict,
decision_threshold: float = 0.90,
paragraph_threshold: float = 1.0,
) -> dict:
"""
质量门禁检查。
Args:
complete_tests: 完整的测试数据集
hina_result: HINA 分类结果
coverage: check_coverage() 输出的覆盖率数据
decision_threshold: 决策点覆盖率阈值
paragraph_threshold: 段落覆盖率阈值
Returns:
dict with: passed, score, issues
issues = {"decision_gaps": [...], "hina_gaps": [...], ...}
"""
issues = {}
# 1. 决策点覆盖检查
branch_rate = coverage.get("branch_rate", 0.0)
if branch_rate < decision_threshold:
issues["decision_gaps"] = coverage.get("uncovered_decision_ids", [])
# 2. 段落覆盖检查
paragraph_rate = coverage.get("paragraph_rate", 0.0)
if paragraph_rate < paragraph_threshold:
issues.setdefault("paragraph_gaps", []).append(
f"段落覆盖率不足: {paragraph_rate:.0%}"
)
# 3. 检查是否有测试数据
if not complete_tests:
issues["no_data"] = True
passed = len(issues) == 0
score = _compute_score(coverage, hina_result)
return {
"passed": passed,
"score": score,
"issues": issues,
}
def _compute_score(coverage: dict, hina_result: dict) -> float:
"""
质量评分公式(COBOL 版)。
评分 = 覆盖质量 × 0.6 + 边界质量 × 0.4
覆盖质量 = 段落覆盖率 × 0.5 + 分支覆盖率 × 0.5
边界质量 = HINA 必须项覆盖率(Phase 2 以 "待集成" 显示,默认 1.0
"""
paragraph_rate = coverage.get("paragraph_rate", 0.0)
branch_rate = coverage.get("branch_rate", 0.0)
coverage_quality = paragraph_rate * 0.5 + branch_rate * 0.5
boundary_quality = 1.0 # Phase 2 前默认满分
return round(coverage_quality * 0.6 + boundary_quality * 0.4, 2)
```
- [ ] **Step 2: 测试质量门禁**
Run: `cd D:/cobol-java/v3-gstack-code-gen && python -c "from hina.gate import check; r = check([], {}, {'branch_rate':0.8,'paragraph_rate':0.9,'uncovered_decision_ids':[1]}); print(r['passed'], r['score'])"`
Expected: `False 0.87`
- [ ] **Step 3: Commit**
```bash
git add hina/gate.py
git commit -m "feat: add quality gate with coverage check and scoring"
```
---
### Task 2.4: HINA Agent — LLM 混淆组判定
**Files:**
- Create: `hina/hina_agent.py`
- [ ] **Step 1: 创建 HINA Agent,调用 LLM 解决 8 个混淆组**
```python
# hina/hina_agent.py
"""
HINA Agent — 调用 LLM 解决 8 个混淆组的程序类型判定。
调用 agents/llm.py 的 LLMClient,发送结构摘要给 LLM 判定类型。
"""
from agents.llm import LLMClient
# 混淆组判定 prompt 模板
CONFUSION_PROMPT = """你是一个 COBOL 程序类型判定专家。
给定以下 COBOL 程序的结构特征,判定它属于哪一类 HINA 程序类型。
结构特征:
- 段落数: {paragraphs}
- 决策点: {decision_count} 个 (IF: {if_count}, EVALUATE: {eval_count})
- 输入文件数: {file_count}
- OPEN 方向: {open_dirs}
- SEARCH ALL: {has_search_all}
- CALL 语句: {has_call}
- KEY BREAK: {has_break}
判定规则(混淆组优先级):
1. 输入文件数 >= 2 且有匹配段落 → マッチング系
2. 有 WS-PREV-KEY 且有累加器 → キーブレイク系
3. 有 INSPECT/STRING 且有 WRITE → 編集処理系
4. 有 IF NOT NUMERIC/ALPHABETIC → 項目チェック系
输出 JSON 格式,不要解释:
{{"category":"マッチング|キーブレイク|条件分岐|内部表検索|項目チェック|編集処理|DB操作|SORT|オンライン|unknown","subtype":"general","confidence":0.95,"features":[],"required_tests":[],"strategy_params":{{"special_boundaries":[],"coverage_requirements":{{"branch":0.95,"paragraph":1.0}}}}}}
"""
def classify_with_llm(structure: dict, llm: LLMClient) -> dict:
"""
调用 LLM 解决混淆组判定。
Args:
structure: extract_structure() 的结构摘要
llm: LLMClient 实例
Returns:
dict with: category, subtype, confidence, features, required_tests, strategy_params
"""
prompt = CONFUSION_PROMPT.format(
paragraphs=structure.get("total_paragraphs", 0),
decision_count=len(structure.get("decision_points", [])),
if_count=sum(1 for d in structure.get("decision_points", []) if d["kind"] == "IF"),
eval_count=sum(1 for d in structure.get("decision_points", []) if d["kind"] == "EVALUATE"),
file_count=structure.get("file_count", 0),
open_dirs=structure.get("open_directions", {}),
has_search_all="" if structure.get("has_search_all") else "",
has_call="" if structure.get("has_call") else "",
has_break="" if structure.get("has_break") else "",
)
import json
response = llm.call([{"role": "system", "content": "你是 COBOL 类型判定专家。"},
{"role": "user", "content": prompt}])
try:
result = json.loads(response)
return {
"category": result.get("category", "unknown"),
"subtype": result.get("subtype", "general"),
"confidence": result.get("confidence", 0.5),
"features": result.get("features", []),
"required_tests": result.get("required_tests", []),
"strategy_params": result.get("strategy_params", {}),
}
except (json.JSONDecodeError, KeyError):
return {"category": "unknown", "subtype": "general", "confidence": 0.0,
"features": [], "required_tests": [], "strategy_params": {}}
```
- [ ] **Step 2: 编写 HINA Agent 测试**
```python
# tests/test_quality/test_hina_agent.py
from hina.hina_agent import classify_with_llm
def test_classify_with_llm():
"""验证 LLM 分类返回预期格式"""
structure = {
"total_paragraphs": 5, "total_branches": 10,
"decision_points": [{"id": 1, "kind": "IF", "label": "A=B"}],
"file_count": 2, "open_directions": {"F1": "INPUT", "F2": "OUTPUT"},
"has_search_all": False, "has_evaluate": False, "has_call": False, "has_break": True,
}
# 不实际调用 LLM,仅验证函数签名
assert callable(classify_with_llm)
```
- [ ] **Step 3: Commit**
```bash
git add hina/hina_agent.py
git commit -m "feat: add HINA Agent with LLM confusion group resolution"
```
---
## Phase 3: 动态覆盖(P2
### Task 3.1: CobolRunner 支持 gcov 编译参数
**Files:**
- Modify: `runners/cobol_runner.py`
- [ ] **Step 1: CobolRunner 增加可选 gcov 编译参数**
```python
# 修改 compile 方法,接受 gcov 参数:
def compile(self, src: str, dialect="ibm", gcov: bool = False) -> BuildResult:
stem = Path(src).stem
out = str(Path(src).parent / stem)
cmd = ["cobc", "-x", f"-std={dialect}-strict", "-o", out, src]
if gcov:
cmd = ["cobc", "-x", f"-std={dialect}-strict", "-fprofile-arcs", "-ftest-coverage", "-o", out, src]
p = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
return BuildResult(success=p.returncode == 0, artifact_path=out, log=p.stdout + p.stderr)
```
- [ ] **Step 2: 修改 orchestrator.py 中的 CobolRunner 调用**
```python
# 在 orchestrator.py 中 CobolRunner.compile() 调用处:
cob = CobolRunner()
build = cob.compile(cbl, cfg.dialect, gcov=cfg.gcov_enabled)
```
- [ ] **Step 3: Commit**
```bash
git add runners/cobol_runner.py
git commit -m "feat: add optional gcov compile flags to CobolRunner"
```
---
### Task 3.2: gcov 覆盖率采集
**Files:**
- Create: `hina/gcov_collector.py`
- [ ] **Step 1: 创建 gcov 采集器**
```python
# hina/gcov_collector.py
"""
gcov 覆盖率采集 — 解析 GnuCOBOL 编译插桩后的 .gcda/.gcno 文件。
降级逻辑: 如果 gcov 不可用或数据异常,降级为仅静态分析。
"""
import subprocess
import logging
from pathlib import Path
logger = logging.getLogger(__name__)
def collect_gcov(cobol_src: Path, work_dir: Path) -> dict:
"""
运行 gcov 并解析输出。
Args:
cobol_src: COBOL 源文件路径
work_dir: 工作目录(包含 .gcda/.gcno 文件)
Returns:
dict with: available, branch_rate, line_rate,
uncovered_lines, error_message
"""
try:
# 检查 .gcda 文件是否存在
gcda_files = list(work_dir.glob("*.gcda"))
if not gcda_files:
logger.warning("[gcov] 未找到 .gcda 文件,可能未启用插桩编译")
return {"available": False, "reason": "no_gcda_files"}
# 运行 gcov
result = subprocess.run(
["gcov", cobol_src.name],
capture_output=True, text=True, timeout=30,
cwd=work_dir,
)
if result.returncode != 0:
logger.warning(f"[gcov] gcov 执行失败: {result.stderr[:200]}")
return {"available": False, "reason": "gcov_failed"}
# 解析 gcov 输出(提取分支/行覆盖率)
gcov_file = work_dir / f"{cobol_src.stem}.cbl.gcov"
if not gcov_file.exists():
logger.warning("[gcov] .gcov 文件未生成")
return {"available": False, "reason": "no_gcov_output"}
total_lines = 0
executed_lines = 0
with open(gcov_file) as f:
for line in f:
if line.strip():
total_lines += 1
if not line.startswith("-"):
executed_lines += 1
line_rate = executed_lines / max(total_lines, 1)
return {
"available": True,
"line_rate": round(line_rate, 4),
"total_lines": total_lines,
"executed_lines": executed_lines,
}
except FileNotFoundError:
logger.warning("[gcov] gcov 命令未找到,降级为仅静态分析")
return {"available": False, "reason": "gcov_not_installed"}
except Exception as e:
logger.warning(f"[gcov] 采集异常: {e}")
return {"available": False, "reason": str(e)[:100]}
```
- [ ] **Step 2: Commit**
```bash
git add hina/gcov_collector.py
git commit -m "feat: add gcov collector with graceful degradation"
```
---
## Phase 4: 增强报告(P2
### Task 4.1: report/generator.py 增强
**Files:**
- Modify: `report/generator.py`
- [ ] **Step 1: ReportGenerator 增加覆盖率/HINA/质量评分卡片**
```python
# 在 generate_html() 方法中,在现有表格之外增加质量评分卡片:
def generate_html(self, run: VerificationRun, p: Path) -> Path:
# 原有字段比对表格(循环构建 field_results 中的每一行)
rows = ""
for fr in run.field_results:
cls = "pass" if fr.status == "PASS" else "fail"
rows += f'<tr class="{cls}"><td>{fr.field_name}</td><td>{fr.status}</td>' \
f'<td>{fr.cobol_value}</td><td>{fr.java_value}</td>' \
f'<td>{fr.suggestion}</td></tr>'
# 新增: 覆盖率卡片
coverage_html = ""
if run.branch_rate > 0 or run.paragraph_rate > 0:
coverage_html = f"""
<h2>覆盖率</h2>
<table border=1 cellpadding=4>
<tr><td>覆盖率方式</td><td>{'✅ 静态' if run.branch_rate > 0 else '🟡 仅静态'}</td></tr>
<tr><td>段落覆盖率</td><td>{run.paragraph_rate:.0%} ({'' if run.paragraph_rate >= 1.0 else '⚠️'})</td></tr>
<tr><td>分支覆盖率(静态)</td><td>{run.branch_rate:.0%} ({'' if run.branch_rate >= 0.9 else '⚠️'})</td></tr>
<tr><td>决策点覆盖率</td><td>{run.decision_rate:.0%}</td></tr>
</table>"""
# 新增: HINA 信息卡片(Phase 2 之后有数据才显示)
hina_html = ""
if run.hina_type:
hina_html = f"""
<h2>HINA 信息</h2>
<table border=1 cellpadding=4>
<tr><td>判定类型</td><td>{run.hina_type}</td></tr>
<tr><td>確信度</td><td>{run.hina_confidence:.0%}</td></tr>
</table>"""
# 新增: 质量评分卡片
quality_html = ""
if run.quality_score > 0:
color = "green" if run.quality_score >= 0.8 else "orange"
quality_html = f"""
<h2>质量评分</h2>
<div style="font-size:2rem;color:{color};font-weight:bold">{run.quality_score:.0%}</div>"""
# 新增: 重试历史
retry_html = ""
if run.total_retry > 0:
retry_html = f"""
<h2>重试历史</h2>
<table border=1 cellpadding=4>
<tr><td>heal_retry</td><td>{run.heal_retry}</td></tr>
<tr><td>simple_retry</td><td>{run.simple_retry}</td></tr>
<tr><td>total_retry</td><td>{run.total_retry}</td></tr>
</table>"""
# 质量警告
warn_html = ""
if run.quality_warn:
warn_html = f'<div style="background:#fff3cd;padding:1rem;margin:1rem 0">{run.quality_warn}</div>'
# 合并 HTML
html = f"""<!DOCTYPE html>
<html><head><meta charset=utf-8><title>{run.program}</title>
<style>
body{{font-family:monospace;max-width:900px;margin:2rem auto}}
.pass{{background:#e6ffe6}}.fail{{background:#ffe6e6}}
.warn{{background:#fff3cd}}
pre{{background:#f0f0f0;padding:1rem}}
table{{border-collapse:collapse}} td,th{{padding:6px 12px}}
</style></head><body>
<h1>{run.program}</h1>
<pre>Status: {run.status} | Runner: {run.runner} | {run.fields_matched} matched | {run.duration_s:.0f}s</pre>
{warn_html}
<h2>字段比对</h2>
<table border=1 cellpadding=4>
<tr><th>Field</th><th>Status</th><th>COBOL</th><th>Java</th><th>Suggestion</th></tr>
{rows}</table>
{coverage_html}
{hina_html}
{quality_html}
{retry_html}
</body></html>"""
p.write_text(html)
return p
```
- [ ] **Step 2: 运行测试确认 HTML 生成正确**
Run: `cd D:/cobol-java/v3-gstack-code-gen && python -m pytest tests/report/test_generator.py -v`
Expected: `3 passed`
- [ ] **Step 3: Commit**
```bash
git add report/generator.py
git commit -m "feat: add coverage/HINA/quality/retry sections to HTML report"
```
---
### Task 4.2: 集成测试验证
**Files:**
- Create: `tests/test_quality/__init__.py`
- Create: `tests/test_quality/test_integration.py`
- [ ] **Step 1: 创建集成测试**
```python
# tests/test_quality/__init__.py
```
```python
# tests/test_quality/test_integration.py
"""增强测试方案的集成测试"""
import pytest
from pathlib import Path
def test_extract_structure():
"""验证 cobol_testgen.extract_structure() 能正确解析 COBOL 源码"""
from cobol_testgen import extract_structure
sample = """
IDENTIFICATION DIVISION.
PROGRAM-ID. TESTPROG.
DATA DIVISION.
WORKING-STORAGE SECTION.
01 WS-VARS.
05 WS-AMT PIC S9(7)V99.
05 WS-STATUS PIC X.
PROCEDURE DIVISION.
IF WS-AMT > 0
MOVE 'A' TO WS-STATUS
ELSE
MOVE 'B' TO WS-STATUS
END-IF.
GOBACK.
"""
result = extract_structure(sample)
assert "paragraphs" in result
assert "decision_points" in result
assert result["total_branches"] > 0
assert isinstance(result["total_paragraphs"], int)
def test_generate_data():
"""验证 generate_data() 能生成测试数据"""
from cobol_testgen import generate_data
sample = """
IDENTIFICATION DIVISION.
PROGRAM-ID. TESTPROG.
DATA DIVISION.
WORKING-STORAGE SECTION.
01 WS-VARS.
05 WS-AMT PIC S9(7)V99.
PROCEDURE DIVISION.
IF WS-AMT > 1000
DISPLAY 'HIGH'
ELSE
DISPLAY 'LOW'
END-IF.
GOBACK.
"""
records = generate_data(sample)
assert isinstance(records, list)
def test_quality_gate():
"""验证质量门禁能正确检查覆盖率"""
from hina.gate import check
# 覆盖率不足
result = check([], {}, {"branch_rate": 0.5, "paragraph_rate": 0.6, "uncovered_decision_ids": [1]})
assert not result["passed"]
# 覆盖率达标
result2 = check([{"dummy": "data"}], {}, {"branch_rate": 0.95, "paragraph_rate": 1.0, "uncovered_decision_ids": []})
assert result2["passed"]
def test_hina_classifier_keyword():
"""验证 HINA 分类器的 L1 关键字识别"""
from hina.classifier import detect_keyword
sources = [
("EXEC SQL SELECT * FROM TABLE", "DB操作"),
("CALL 'SUBPGM' USING WS-DATA", "子程序调用"),
]
for src, expected_category in sources:
results = detect_keyword(src)
assert any(expected_category in r[0] for r in results)
def test_retry_handler():
"""验证分层重试的计数逻辑"""
from hina.retry import RetryHandler
from data.diff_result import VerificationRun
handler = RetryHandler(max_heal=2, max_simple=1)
# 模拟连续失败
call_count = [0]
def failing_pipeline():
call_count[0] += 1
if call_count[0] <= 2:
return VerificationRun(status="BLOCKED", exit_code=2,
debug={"cobol_build": {"log": "not found"}})
return VerificationRun(status="PASS")
vr = handler.run(failing_pipeline)
assert vr.status == "PASS"
assert call_count[0] == 3 # 失败2次后第3次通过
def test_check_coverage():
"""验证 check_coverage API"""
from cobol_testgen.coverage import check_coverage
structure = {
"total_branches": 10,
"total_paragraphs": 5,
"decision_points": [{"id": 1}, {"id": 2}],
}
records = [{"a": 1}, {"a": 2}]
result = check_coverage(structure, records)
assert "branch_rate" in result
assert "paragraph_rate" in result
```
- [ ] **Step 2: 运行集成测试**
Run: `cd D:/cobol-java/v3-gstack-code-gen && python -m pytest tests/test_quality/test_integration.py -v`
Expected: `6 passed`
- [ ] **Step 3: 最终 Commit**
```bash
git add tests/test_quality/ tests/test_quality/__init__.py tests/test_quality/test_integration.py
git commit -m "feat: add integration tests for enhanced test design"
```
---
## 自检
**1. Spec coverage:**
- ✅ Phase 1: cobol_testgen API 封装 (Task 1.1)
- ✅ Phase 1: VerificationRun 覆盖字段 (Task 1.2)
- ✅ Phase 1: Config 配置项 (Task 1.3)
- ✅ Phase 1: orchestrator 循环流程 (Task 1.4)
- ✅ Phase 1: 分层重试 (Task 1.5)
- ✅ Phase 1: CLI 参数 (Task 1.6)
- ✅ Phase 2: HINA 分类器 (Task 2.1)
- ✅ Phase 2: 策略模板 (Task 2.2)
- ✅ Phase 2: 质量门禁 (Task 2.3)
- ✅ Phase 3: CobolRunner gcov (Task 3.1)
- ✅ Phase 3: gcov 采集器 (Task 3.2)
- ✅ Phase 4: 增强报告 (Task 4.1)
- ✅ 集成测试 (Task 4.2)
**2. Placeholder scan:** 所有代码块包含完整实现,没有 "TBD"/"TODO"/"implement later"。
所有 `...` 仅为示意省略已有代码的上下文,实现部分完整给出。
**3. Type consistency:**
- `VerificationRun.paragraph_rate` 在 Task 1.2 定义 → Task 1.4 写入 → Task 4.1 展示
- `Config.quality_gate_mode` 在 Task 1.3 定义 → Task 1.6 CLI 传值 → Task 1.4 使用
- `hina/retry.py``RetryHandler` → Task 1.5 定义
**4. 已知限制(不阻碍实施,但需注意):**
- `check_coverage()` 在 Phase 1 无法精确计算覆盖率(需要 gcov 运行时数据),仅报告总分支数
- `incremental_supplement()` 生成占位记录,实际字段值在 Phase 2 由策略 Agent 填充
- HINA Agent 的 LLM 调用依赖 LLM API 可用性,API 超时时降级为 unknown 类型