From 0203ead96b4e1f2e30c8b63667c53ebe6ba0ab2a Mon Sep 17 00:00:00 2001 From: hangshuo652 Date: Sat, 18 Jul 2026 08:42:55 +0800 Subject: [PATCH] feat: config-driven multi-scenario + gcda accumulation fix + gcov merge fix Config-driven architecture: - coverage_dates: YAML-driven date replacement for LEAVE_RECORDS - row_overrides: per-scenario field overrides (e.g. STATUS='9') - delete_all_rows: per-scenario table emptying for empty-cursor tests gcda accumulation fix (V3 bug #7): - Delete .gcda before each scenario run to prevent GnuCOBOL accumulation - Force-copy scenario gcda (skip mtime check) - Copy scenario DB to CWD data/kin.db for CONNECT TO path resolution Multi-run gcov merge fix: - Always merge multi-run gcov data regardless of generate_coverage flag - Fix v3_root UnboundLocalError in cleanup path Other fixes: - _make_key_unique: skip WHERE-constrained columns to avoid PK conflict - incremental_supplement: support fields_dict for base record generation - check_coverage: use structure coverage data if available - orchestrator.py: filter _-prefixed fields in TestCase; merge Agent2Data cases --- cobol_testgen/__init__.py | 18 ++++--- cobol_testgen/coverage.py | 24 ++++++++- cobol_testgen/to_sql.py | 4 +- config/program_schema.py | 6 +++ config/programs/KIN03EXP.yaml | 44 ++++++++++++++++ orchestrator.py | 17 +++++-- orchestrator_db.py | 94 ++++++++++++++++++++++++++++++++--- 7 files changed, 186 insertions(+), 21 deletions(-) diff --git a/cobol_testgen/__init__.py b/cobol_testgen/__init__.py index cb1f2c4..aaea8dd 100644 --- a/cobol_testgen/__init__.py +++ b/cobol_testgen/__init__.py @@ -28,7 +28,7 @@ from .core import build_branch_tree, classify_field_roles, _init_child_names, sq 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 +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 @@ -1195,6 +1195,7 @@ def extract_structure(cobol_source: str, copybook_dirs: list = None) -> dict: "if_types": if_types, "variable_patterns": variable_patterns, "open_pattern": open_pattern, + "data_fields": fields_dict, } @@ -1353,12 +1354,13 @@ def generate_data(cobol_source: str, structure: dict = None, return records -def incremental_supplement(branch_tree, decision_gaps: list[int]) -> list[dict]: +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() 兼容 @@ -1390,10 +1392,12 @@ def incremental_supplement(branch_tree, decision_gaps: list[int]) -> list[dict]: supplements = [] for i, (kind, label) in enumerate(found): - supplements.append({ - "_dec_id": f"incr_{i}", - "_kind": kind, - "_label": str(label)[:60], - }) + 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 diff --git a/cobol_testgen/coverage.py b/cobol_testgen/coverage.py index 9e92c41..6348967 100644 --- a/cobol_testgen/coverage.py +++ b/cobol_testgen/coverage.py @@ -1358,11 +1358,33 @@ def run_coverage(branch_tree, branch_paths_with_assigns, fields, def check_coverage(structure: dict, test_records: list[dict]) -> dict: total_paragraphs = structure.get("total_paragraphs", 0) total_branches = structure.get("total_branches", 0) - decision_points = structure.get("decision_points", []) has_data = len(test_records) > 0 paragraph_rate = 1.0 if (total_paragraphs > 0 and has_data) else 0.0 + cov = structure.get("coverage", {}) + if cov and "total" in cov: + total = cov["total"] + covered = cov["covered"] + branch_rate = covered / max(total, 1) if total > 0 else 0.0 + decision_rate = branch_rate + + uncovered = [] + for dp in cov.get("decision_points", []): + if dp["covered"] < dp["branches"]: + uncovered.append(dp["id"]) + + return { + "paragraph_rate": paragraph_rate, + "branch_rate": branch_rate, + "decision_rate": decision_rate, + "uncovered_decision_ids": uncovered, + "total_branches": total, + "total_paragraphs": total_paragraphs, + "records_count": len(test_records), + "note": "静态分析覆盖率（来自 mark_coverage）。精确数据通过 gcov 获取。", + } + return { "paragraph_rate": paragraph_rate, "branch_rate": 0.0, diff --git a/cobol_testgen/to_sql.py b/cobol_testgen/to_sql.py index 82b95de..e1d21ff 100644 --- a/cobol_testgen/to_sql.py +++ b/cobol_testgen/to_sql.py @@ -434,12 +434,14 @@ def build_db_input( if sc not in [c['name'] for c in col_infos]: col_infos.append({'name': sc, 'db_type': 'CHAR', 'size': 20}) + where_cols = set() for col_info in col_infos: col_name = col_info['name'] val = None for wc in where_cons: if wc['type'] == 'literal' and wc.get('col', '').upper() == col_name: val = wc.get('literal', '') + where_cols.add(col_name) break if wc['type'] == 'host_var': hv = wc.get('host_var', '').upper() @@ -473,7 +475,7 @@ def build_db_input( if col_infos: first_col = col_infos[0]['name'] - if first_col in row: + if first_col in row and (not where_cols or first_col not in where_cols): row[first_col] = _make_key_unique(row[first_col], path_idx, seen_keys[table]) db_input[table].append(row) diff --git a/config/program_schema.py b/config/program_schema.py index 030940b..96d7a63 100644 --- a/config/program_schema.py +++ b/config/program_schema.py @@ -40,6 +40,8 @@ class ScenarioDef: id: str sysin: SysinDef = field(default_factory=SysinDef) inject_duplicate_pk: bool = False + row_overrides: dict[str, dict[str, str]] = field(default_factory=dict) + delete_all_rows: bool = False @dataclass @@ -50,6 +52,7 @@ class ProgramSchema: db_type: str = "SQLite" db_name: str = "OVERTIME.DB" runs: list[ScenarioDef] = field(default_factory=list) + coverage_dates: dict[str, list[dict[str, str]]] = field(default_factory=dict) @classmethod def from_yaml(cls, path: str | Path) -> ProgramSchema: @@ -71,6 +74,8 @@ class ProgramSchema: id=r["id"], sysin=SysinDef(**sysin_raw), inject_duplicate_pk=r.get("inject_duplicate_pk", False), + row_overrides=r.get("row_overrides", {}), + delete_all_rows=r.get("delete_all_rows", False), )) return cls( program_id=raw["program_id"], @@ -79,6 +84,7 @@ class ProgramSchema: db_type=raw.get("db_type", "SQLite"), db_name=raw.get("db_name", "OVERTIME.DB"), runs=runs, + coverage_dates=raw.get("coverage_dates", {}), ) diff --git a/config/programs/KIN03EXP.yaml b/config/programs/KIN03EXP.yaml index a4021cc..9ef5e4d 100644 --- a/config/programs/KIN03EXP.yaml +++ b/config/programs/KIN03EXP.yaml @@ -35,3 +35,47 @@ subprograms: - SUB01DAT - SUB02MSG - SUB03END + +coverage_dates: + LEAVE_RECORDS: + - start: "20260701" + end: "20260701" + emp: "00000001" + - start: "20260702" + end: "20260702" + emp: "00000001" + - start: "20260704" + end: "20260704" + emp: "00000001" + - start: "20260705" + end: "20260705" + emp: "00000001" + - start: "20260131" + end: "20260203" + emp: "00000002" + - start: "20260430" + end: "20260502" + emp: "00000002" + - start: "20240228" + end: "20240302" + emp: "00000003" + - start: "20230228" + end: "20230302" + emp: "00000003" + - start: "20240229" + end: "20240302" + emp: "00000004" + - start: "20230227" + end: "20230302" + emp: "00000004" + - start: "20261230" + end: "20270101" + emp: "00000099" + +runs: + - id: normal + - id: empty_cursor + delete_all_rows: true + row_overrides: + LEAVE_RECORDS: + STATUS: "9" diff --git a/orchestrator.py b/orchestrator.py index 6b24d56..cde532e 100644 --- a/orchestrator.py +++ b/orchestrator.py @@ -49,7 +49,8 @@ def run_pipeline(cfg: Config, cpath: str, cbl: str, java: str, map_path: str) -> # 转换为 TestCase 列表（增强管线的基础数据集） complete_tests = [] for i, rec in enumerate(base_records): - complete_tests.append(TestCase(id=f"CTG-{i+1:04d}", fields=dict(rec))) + fields = {k: v for k, v in rec.items() if not k.startswith('_')} + complete_tests.append(TestCase(id=f"CTG-{i+1:04d}", fields=fields)) # HINA 完整类型判定管道（Keyword / 规则引擎 / LLM 辅助三路径） classification: dict = {} @@ -84,13 +85,15 @@ def run_pipeline(cfg: Config, cpath: str, cbl: str, java: str, map_path: str) -> break gaps = gate_result.get("issues", {}).get("decision_gaps", []) if gaps and structure.get("branch_tree_obj"): - delta = incremental_supplement(structure["branch_tree_obj"], gaps) + data_fields = structure.get("data_fields", []) + delta = incremental_supplement(structure["branch_tree_obj"], gaps, data_fields) base_records.extend(delta) # 同步更新 complete_tests for i, d in enumerate(delta): + d_fields = {k: v for k, v in d.items() if not k.startswith('_')} complete_tests.append(TestCase( id=f"CTG-S{attempt+1}-{i+1:04d}", - fields=dict(d), + fields=d_fields, )) cov = check_coverage(structure, base_records) else: @@ -109,7 +112,13 @@ def run_pipeline(cfg: Config, cpath: str, cbl: str, java: str, map_path: str) -> suite = Agent2Data(llm).design(tree, cfg.coverage_default, cfg.runner_mode == "spark") vr.llm_cost += 0.002 - suite.test_cases = complete_tests # 替换为增强管线数据（P1/P2 修复） + # 合并: cobol_testgen 基线 + 策略标记 + Agent2Data LLM 场景数据 + merged = list(complete_tests) + existing_ids = {tc.id for tc in merged} + for tc in suite.test_cases: + if tc.id not in existing_ids: + merged.append(tc) + suite.test_cases = merged vr.debug["test_cases"] = [{"id":tc.id,"fields":tc.fields,"targets":tc.coverage_targets} for tc in suite.test_cases] vr.debug["spark_config"] = {"records":suite.spark_config.num_records} if suite.has_spark else None diff --git a/orchestrator_db.py b/orchestrator_db.py index 8c68129..7708ed2 100644 --- a/orchestrator_db.py +++ b/orchestrator_db.py @@ -234,7 +234,7 @@ class GixsqlOrchestrator: self._init_database(db_path) # DB 初期行投入（DELETE/UPDATE が作用する行、SELECT が返す行） - self._populate_database(db_path, src_text, recs) + self._populate_database(db_path, src_text, recs, scenario=scenario) # P5: inject duplicate-PK rows (scenario で制御) if scenario is None or scenario.inject_duplicate_pk: self._inject_sql_error_rows(db_path, recs) @@ -551,8 +551,23 @@ class GixsqlOrchestrator: self.db_path.unlink() shutil.copy2(str(db_path), str(self.db_path)) db_path = self.db_path + # CONNECT TO 'data/kin.db' のパス解釈に備え CWD にもコピー + if scenario is not None: + cwd_data = cwd / "data" + cwd_data.mkdir(parents=True, exist_ok=True) + cwd_db = cwd_data / "kin.db" + if cwd_db.exists(): + cwd_db.unlink() + shutil.copy2(str(db_path), str(cwd_db)) # .gcda は CWD（= run_dir）に書き出されるので、実行後に gcov/run_{id}/ に移動する + # 各シナリオ実行前に前回の .gcda を削除（GnuCOBOL は累積書込みを行うため） + exe_dir_for_gcda = self.work_dir / "bin" + for f in exe_dir_for_gcda.glob("*.gcda"): + try: + f.unlink() + except PermissionError: + pass # Subprogram DLLs cobol_bin = Path(self.cobol_src_dir).parent / "bin" @@ -585,7 +600,7 @@ class GixsqlOrchestrator: for f in sd.glob("*.gcda"): if f.is_file() and f.stat().st_size > 0: dst = gcda_dst_dir / f.name - if not dst.exists() or f.stat().st_mtime > dst.stat().st_mtime: + if scenario or not dst.exists() or f.stat().st_mtime > dst.stat().st_mtime: try: shutil.copy2(str(f), str(dst)) except PermissionError: @@ -772,7 +787,8 @@ class GixsqlOrchestrator: generate_coverage_index([cov_result], str(output_dir.parent)) # Clean up .gcno/.gcda from v3_root + CWD (avoid accumulation) - for clean_dir in (v3_root, Path.cwd()): + _v3_root = Path(__file__).parent + for clean_dir in (_v3_root, Path.cwd()): if clean_dir == gcov_dir: continue for ext in (".gcno", ".gcda"): @@ -967,12 +983,13 @@ class GixsqlOrchestrator: elif step_num == 6: vr = self.step6_verify() + # Always merge multi-run gcov data (needed by external coverage report) + if is_multi: + merged = self._merge_multi_run_gcov() + self._multi_run_gcov_data = merged # Optional coverage report (non-blocking) cv_flags = getattr(self.config, 'gixsql_compile_flags', '') if '--coverage' in cv_flags and generate_coverage: - if is_multi: - merged = self._merge_multi_run_gcov() - self._multi_run_gcov_data = merged self.generate_coverage_report() vr = VerificationRun( @@ -1061,7 +1078,8 @@ class GixsqlOrchestrator: conn.close() logger.info(f" DB initialized: {db_path}") - def _populate_database(self, db_path: Path, src_text: str, records: list[dict]): + def _populate_database(self, db_path: Path, src_text: str, records: list[dict], + scenario: ScenarioDef | None = None): """テストデータから DB 初期行を生成し挿入する。""" from cobol_testgen.pipeline_bridge import build_branch_tree_fallback from cobol_testgen.read import extract_procedure_division @@ -1118,6 +1136,43 @@ class GixsqlOrchestrator: logger.info(" No DB input rows generated") return + # -- Coverage-driven data enrichment -- + # build_db_input generates counter-value dates; replace with valid + # YYYYMMDD dates targeting specific uncovered decision branches. + # Configurations are from YAML coverage_dates (program-specific). + if 'LEAVE_RECORDS' in db_input: + lr_rows = db_input['LEAVE_RECORDS'] + date_cfgs_raw = (self.schema.coverage_dates or {}).get('LEAVE_RECORDS', []) + date_cfgs = [ + (d['start'], d['end'], d['emp']) + for d in date_cfgs_raw + ] + for i, row in enumerate(lr_rows): + if i < len(date_cfgs): + sd, ed, eid = date_cfgs[i] + else: + sd, ed, eid = ('20260701', '20260703', f'{i+10:08d}') + row['START_DATE'] = sd + row['END_DATE'] = ed + row['EMP_ID'] = eid + row['APPLICATION_ID'] = str(i + 1) + + if 'HOLIDAY_CALENDAR' in db_input: + hc_rows = db_input['HOLIDAY_CALENDAR'] + holiday_overrides = ['20260701', '20260715', '20260801', + '20260101', '20260501', '20261001'] + for i, row in enumerate(hc_rows): + if i < len(holiday_overrides): + row['HOLIDAY_DATE'] = holiday_overrides[i] + + # -- Per-scenario row overrides (from YAML runs[].row_overrides) -- + if scenario and scenario.row_overrides: + for table_name, overrides in scenario.row_overrides.items(): + if table_name in db_input: + for row in db_input[table_name]: + for col, val in overrides.items(): + row[col.upper()] = val + conn = sqlite3.connect(str(db_path)) for table_name, rows in db_input.items(): if not rows: @@ -1127,12 +1182,14 @@ class GixsqlOrchestrator: # Debug logger.info(f" Table {table_name}: {len(rows)} rows, cols={list(rows[0].keys()) if rows else []}") - # Filter columns: only keep those that actually exist in the table + # Query DB column types for type-aware value conversion + col_types = {} try: pragma_cols = conn.execute( f"PRAGMA table_info([{table_name}])" ).fetchall() valid_cols = {r[1].upper() for r in pragma_cols} + col_types = {r[1].upper(): r[2].upper() for r in pragma_cols} except Exception: valid_cols = set() @@ -1149,12 +1206,33 @@ class GixsqlOrchestrator: logger.info(f" Table {table_name}: all rows filtered out, skipping") continue + # Convert values to match DB column types + for row in rows: + for k in list(row.keys()): + ct = col_types.get(k.upper(), '') + v = row[k] + if ct.startswith('INTEGER') or ct in ('INT', 'SMALLINT', 'BIGINT', 'TINYINT'): + try: + row[k] = str(int(v)) if v and v.strip() else '0' + except (ValueError, TypeError): + row[k] = '0' + elif ct.startswith('DECIMAL') or ct.startswith('NUMERIC') or ct.startswith('FLOAT') or ct.startswith('REAL'): + try: + row[k] = str(float(v)) if v and v.strip() else '0' + except (ValueError, TypeError): + row[k] = '0' + col_names = list(rows[0].keys()) placeholders = ", ".join("?" for _ in col_names) quoted_cols = ", ".join(f"[{c}]" for c in col_names) sql = f"INSERT OR IGNORE INTO [{table_name}] ({quoted_cols}) VALUES ({placeholders})" conn.executemany(sql, [tuple(r.get(c, "") for c in col_names) for r in rows]) logger.info(f" Table {table_name}: {len(rows)} initial rows inserted") + # -- Per-scenario row deletion (e.g. empty cursor scenario) -- + if scenario and scenario.delete_all_rows: + for table_name in db_input.keys(): + conn.execute(f"DELETE FROM [{table_name}]") + logger.info(f" Table {table_name}: all rows deleted (scenario={scenario.id})") conn.commit() conn.close() logger.info(f" DB populated: {db_path}")