"""GixsqlOrchestrator — DB COBOL プログラムの全6Step実行""" from __future__ import annotations import json import logging import os import re import subprocess import sqlite3 from dataclasses import dataclass, field from pathlib import Path from typing import Optional from config import Config from config.program_schema import ProgramSchema, load_schema, ScenarioDef from cobol_testgen import extract_structure, generate_data from cobol_testgen.flatfile import write_all_files, write_sysin_file from cobol_testgen.file_io import read_output_file from cobol_testgen.read import preprocess, resolve_copybooks, resolve_sql_includes, parse_file_control, parse_file_section, parse_data_division, extract_data_division, extract_procedure_division, scan_open_statements from cobol_testgen.read import strip_exec_sql_from_data_div from cobol_testgen.gcov import run_gcov from cobol_testgen.coverage import run_coverage, generate_coverage_index from cobol_testgen.design_mcdc import enum_paths as mcdc_enum_paths from cobol_testgen.to_sql import collect_sql_meta, build_db_input from cobol_testgen.core import extract_sql_assignments, classify_field_roles, _init_child_names from cobol_testgen import expand_occurs from cobol_testgen.design import get_term_type, generate_records from cobol_testgen.output import output_json from cobol_testgen.pipeline_bridge import build_branch_tree_fallback import shutil from data.diff_result import VerificationRun, FieldResult from runners.gixsql_runner import GixsqlCobolRunner, GixsqlTableData from tools.registry import get_registry logger = logging.getLogger(__name__) # DB 管道 SQLite 存放目录。默认保持短 ASCII 路径： # COBOL CONNECT TO 字符串受固定格式第 72 列限制，且 gixpp 无法处理中文路径。 # 可用环境变量 GIXSQL_DB_DIR 覆盖（见 .env.example）。 GIXSQL_DB_DIR = Path(os.environ.get("GIXSQL_DB_DIR", "C:/Temp/gix")) def _calc_birth_date(age: int, as_of: str = '20260802') -> str: """年龄 → 出生日期 YYYYMMDD（通用）。as_of 为运行时运用日。""" from datetime import datetime, timedelta base = datetime.strptime(as_of, '%Y%m%d') d = base - timedelta(days=max(age, 0) * 365) return d.strftime('%Y%m%d') def _merge_run_dirs_gcov(gcov_dir: str | Path, program: str, gcov_func=run_gcov) -> dict[int, int]: """Merge gcov line counts for ONE program across multi-run scenario dirs. Returns {line: max_count}. Subprogram gcov MUST be collected separately (per subprogram name), never merged into the main program's dict: both use plain integer line numbers, so SUB*.cbl line 167 would collide with and overwrite the main program's line 167 (e.g. SUB04CHK 167=0 wiping the main loop's 167=25). See _sub_gcov_data. """ merged: dict[int, int] = {} for sd in sorted(Path(gcov_dir).glob("run_*")): data = gcov_func(program, str(sd)) if data: for line, cnt in data.items(): merged[line] = max(merged.get(line, 0), cnt) return merged @dataclass class DbPipelineResult: """DB 管线単体実行結果""" program_id: str step: int | float # pipeline step number success: bool message: str = "" data: dict = field(default_factory=dict) class GixsqlOrchestrator: """6Step DB 管线オーケストレーター""" def __init__(self, config: Config, program_id: str, cobol_src_dir: str | Path, copybook_dirs: list[str | Path] | None = None, work_dir: str | Path | None = None, skip_jvm: bool = True): self.config = config self.program_id = program_id self.cobol_src_dir = Path(cobol_src_dir) self.copybook_dirs = copybook_dirs or [] self.skip_jvm = skip_jvm # DB-Java 文件式运行 + DB 表比对（run_java=True 时在每场景 Step3 后触发） self.run_java: bool = False self._java_cfg_cache = None self._db_java_results: list[dict] = [] v3_root = Path(__file__).parent # cobol-java-v3/ # Build artifacts in temp (ASCII-only, gixpp can't handle Chinese paths) if work_dir is None: import tempfile temp = Path(tempfile.gettempdir()) work_dir = temp / "gixsql_build" / program_id self.work_dir = Path(work_dir) # Runtime data under V3 (DB, flat files, CWD) # 新目录结构：output//cobol/ 存放COBOL输出 self.runtime_dir = v3_root / "output" / program_id / "cobol" self.schema: ProgramSchema = load_schema(program_id) # 使用工具注册表获取GixsqlCobolRunner registry = get_registry() GixsqlCobolRunnerClass = registry.get("gixsql_runner") self.runner = GixsqlCobolRunnerClass( gixpp_path=config.gixsql_path, lib_path=config.gixsql_lib_path, compile_flags=config.gixsql_compile_flags, ) # Derive DB path: \.db (matches COBOL CONNECT TO, short enough for col 72) self.db_path = GIXSQL_DB_DIR / f"{self.program_id}.db" # Pipeline state self.src_path: Optional[Path] = None self.pp_path: Optional[Path] = None self.exe_path: Optional[Path] = None self.java_input_path: Optional[Path] = None self._current_db_path: Optional[Path] = None # scenario-specific DB path self._java_initial_db_snapshots: dict[str, Path] = {} # scenario label -> initial DB snapshot (before COBOL) self._multi_run_gcov_data: dict[int, int] | None = None # merged multi-run gcov data self._sub_gcov_data: dict[str, dict[int, int]] = {} # per-subprogram gcov (kept separate from main) self.java_output_path: Optional[Path] = None self.generated_records: list[dict] = [] self.generated_structure: dict | None = None # ── Step 1: 環境整備（gixpp + compile） ── def _copy_sources_to_workdir(self) -> tuple[Path, list[str]]: """Copy source + copybooks to ASCII-only workdir (gixpp can't handle Chinese paths).""" src_dir = self.work_dir / "src" src_dir.mkdir(parents=True, exist_ok=True) # Copy main source orig = self.cobol_src_dir / f"{self.program_id}.cbl" ascii_src = src_dir / f"{self.program_id}.cbl" if not ascii_src.exists(): ascii_src.write_bytes(orig.read_bytes()) self.src_path = ascii_src # Copy copybooks flat_cpy = [] for d in self.copybook_dirs: pd = Path(d) if pd.exists(): for f in pd.glob("*.cpy"): dst = src_dir / f.name if not dst.exists(): dst.write_bytes(f.read_bytes()) flat_cpy.append(str(dst)) # Copy SUB programs v3_root = Path(__file__).parent sub_dirs = [ self.cobol_src_dir, self.cobol_src_dir.parent / "sub", v3_root.parent / "cobol-tna-system" / "sub", v3_root.parent / "production" / "sub", ] for sub in self.schema.subprograms: found = False for sd in sub_dirs: sp = sd / f"{sub}.cbl" if sp.exists(): dst = src_dir / f"{sub}.cbl" if not dst.exists(): dst.write_bytes(sp.read_bytes()) found = True break if not found: logger.warning(f" SUB {sub}.cbl not found in {sub_dirs}") return src_dir, flat_cpy def step1_setup_environment(self) -> DbPipelineResult: """gixpp 前処理 → cobc コンパイル""" try: ascii_dir, flat_cpy = self._copy_sources_to_workdir() src = ascii_dir / f"{self.program_id}.cbl" pp = self.runner.preprocess(src, self.work_dir / "preprocessed", copybook_dirs=[ascii_dir]) self.pp_path = Path(pp) # Patch gixpp's broken CONNECT string # gixpp converts CONNECT TO 'data/kin.db' -> 'sqlite://localhost/kin' # Fix: use absolute path that gixsql runtime can resolve if self.pp_path and self.pp_path.exists(): pp_text = self.pp_path.read_text(encoding='utf-8') old_conn = 'sqlite://localhost/kin' new_conn = f'sqlite:///{self.db_path}' if old_conn in pp_text: pp_text = pp_text.replace(old_conn, new_conn) self.pp_path.write_text(pp_text, encoding='utf-8') logger.info(f" Patched CONNECT: {old_conn} -> {new_conn}") else: logger.info(f" CONNECT string not found (already patched?)") exe = self.work_dir / "bin" / f"{self.program_id}.exe" extra_srcs = [] for sub in self.schema.subprograms: sp = ascii_dir / f"{sub}.cbl" if sp.exists(): extra_srcs.append(sp) result = self.runner.compile( pp, exe, copybook_dirs=[ascii_dir], extra_srcs=extra_srcs, ) log_dir = self.runtime_dir / "logs" / "compile" log_dir.mkdir(parents=True, exist_ok=True) log_dir.joinpath(f"{self.program_id}.log").write_text( result.log, encoding='utf-8') if result.success: self.exe_path = Path(result.exe_path) return DbPipelineResult( self.program_id, 1, result.success, message=result.log[:200], data={"exe_path": str(exe), "log": result.log[:500]}, ) except Exception as e: return DbPipelineResult(self.program_id, 1, False, str(e)) # ── Step 2: 入力データ生成 ── def step2_generate_inputs(self, scenario: ScenarioDef | None = None) -> DbPipelineResult: """テストデータ生成 + フラットファイル出力 + DB初期化 Args: scenario: 多輪実行時のシナリオ定義。None=単輪（従来動作）。 """ try: src_text = self.src_path.read_text(encoding="utf-8-sig") # Use the pre-gixpp source for Lark parsing (gixpp output contains SQLCA etc.) parse_text = self.pp_path.read_text(encoding="utf-8") if self.pp_path and self.pp_path.exists() else src_text # COBOL 解析 + テストデータ生成（白盒 + 機能 + 策略 統合） cbd = [str(d) for d in self.copybook_dirs] st = extract_structure(src_text, copybook_dirs=cbd) self.generated_structure = st from cobol_testgen.data_merger import generate_all_data v3_root = Path(__file__).parent design_doc_dir = v3_root / "detailed-design-docs" if not design_doc_dir.exists(): design_doc_dir = None # LLMClient は API key が必要な場合のみ初期化（未設定時は None → スキップ） llm = None if hasattr(self.config, 'llm_model') and self.config.llm_model: from agents.llm import LLMClient try: llm = LLMClient(model=self.config.llm_model, timeout=self.config.llm_timeout) except Exception: pass recs = generate_all_data( program_id=self.program_id, src_text=src_text, st=st, copybook_dirs=cbd, design_doc_dir=str(design_doc_dir) if design_doc_dir else None, llm_client=llm, config=self.config, ) # Post-process: link R02 cancel APPL-IDs to matching R01 insert APPL-IDs for rec in recs: if 'R02APPL-ID' in rec and 'R01APPL-ID' in rec: rec['R02APPL-ID'] = rec['R01APPL-ID'] # シナリオに応じた DB パス if scenario: db_path = GIXSQL_DB_DIR / f"{self.program_id}_{scenario.id}.db" self._current_db_path = db_path else: db_path = self.db_path self._current_db_path = None # DB 初期データ構築: clean stale DB first if db_path.exists(): db_path.unlink() db_path.parent.mkdir(parents=True, exist_ok=True) self._init_database(db_path) # DB 初期行投入（DELETE/UPDATE が作用する行、SELECT が返す行） self._populate_database(db_path, src_text, recs, scenario=scenario) # seed_extra_rows: 大结果集注入（SELECT 型プログラムの表头重出等の分支） if scenario: self._inject_extra_seed_rows(db_path, scenario) # P5: inject duplicate-PK rows (scenario で制御) if scenario is None or scenario.inject_duplicate_pk: self._inject_sql_error_rows(db_path, recs) # First record: empty EMP-ID to trigger R01EMP-ID = SPACE path (DP#12). # Independent of R01LINE (which may not exist for this program's FD layout). if len(recs) > 0: recs[0]['R01EMP-ID'] = ' ' * 8 # 全ゼロ EMP-ID レコードを SPACE にクレンジング（汎用）。 # プログラムの空社員チェック（R01EMP-ID = SPACE / LOW-VALUES）は # '00000000' を捕捉しないため、そのまま INSERT され DAILY_RECORDS の # (EMP_ID, TARGET_DATE) 主キー衝突 → 早期 ABEND を引き起こす。 for rec in recs: _eid = str(rec.get('R01EMP-ID', '')).strip() if _eid == '00000000': rec['R01EMP-ID'] = ' ' * 8 # Patch R01LINE records with EMP-IDs matching the record's own EMP-ID for i, rec in enumerate(recs): line = rec.get('R01LINE', '') if not line: continue parts = line.split(',', 1) if len(parts) != 2: continue # Get EMP-ID from the record's own field (e.g., R01EMP-ID) emp_id = rec.get('R01EMP-ID', '') if not emp_id or emp_id == '00000000': emp_id = rec.get('HV-EMP-ID', '') if not emp_id or emp_id == '00000000': emp_id = f"EMP{str(i).zfill(5)}" if i == 0: rec['R01LINE'] = f"{' '*8},{parts[1]}" else: rec['R01LINE'] = f"{emp_id.ljust(8)},{parts[1]}" rec['R01EMP-ID'] = emp_id # Inject duplicate EMP-IDs for last 3 records to trigger AGG UPDATE # path (DP#19-#20). Use the LARGEST EMP-ID from the last 8 of the # sorted unique list, so the dup ID falls in the last TARGET card # batch (which sets TARGET-COUNT). Set R01DATE to keep YEAR_MONTH # the same but different day avoids PK conflict in DAILY_RECORDS. # Works for both FIXED (KIN07REC) and LINE SEQUENTIAL formats. if len(recs) > 3: # Collect unique EMP-IDs (non-blank, non-zero) all_ids = set() for r in recs: eid = r.get('R01EMP-ID', '') if eid and eid.strip() and eid != '00000000': all_ids.add(eid) sorted_ids = sorted(all_ids) # Pick from the last chunk (matches last TARGET card batch) # T chunk = 8 per card, so last chunk index = len % 8 or 8 n = len(sorted_ids) last_chunk_start = n - (n % 8 or 8) dup_eid = sorted_ids[last_chunk_start] if sorted_ids else '' # Find a record with this EMP-ID for its DATE src_date = '' for r in recs: if r.get('R01EMP-ID', '') == dup_eid: src_date = r.get('R01DATE', '') break dup_date = src_date # Track used days to avoid PK conflict with src record's date src_day = dup_date[6:8] if dup_date and len(dup_date) >= 8 else '' used_days = set() if src_day: used_days.add(src_day) for j in range(max(1, len(recs)-3), len(recs)): rec = recs[j] if not dup_eid: continue rec['R01EMP-ID'] = dup_eid # FIXED format: keep same YEAR_MONTH but uniquely different day # to avoid PK conflict in DAILY_RECORDS INSERT (EMP_ID + TARGET_DATE). if dup_date and len(dup_date) >= 6: dup_ym = dup_date[:6] orig_date = rec.get('R01DATE', '') orig_day = orig_date[6:8] if orig_date and len(orig_date) >= 8 else '' day = orig_day if (orig_day and orig_day not in used_days) else f"{len(used_days)+1:02d}" used_days.add(day) rec['R01DATE'] = dup_ym + day # LINE SEQUENTIAL format: patch R01LINE line = rec.get('R01LINE', '') if line: parts = line.split(',', 1) if len(parts) == 2: rec['R01LINE'] = f"{dup_eid.ljust(8)},{parts[1]}" # 聚合边界数据（overflow / agg table full），通用注入，作用于共享 records self._inject_aggregation_boundaries(recs) # ── Coverage-driven data modifications (per-scenario) ── # Normal scenario or legacy single-run: no modifications needed. # Collision scenario: INSERT duplicate, OVT-MONTHLY match, COMMIT threshold. # Abnormal scenario: orphan cancel ABEND (last, to avoid polluting other branches). apply_collision = scenario is not None and scenario.id == "collision" apply_abnormal = scenario is not None and scenario.id == "abnormal" if apply_collision: # #10 T: Ensure >= 50 R01 records for COMMIT threshold (CNS-COMMIT-CNT=50) r01_recs = [r for r in recs if 'R01APPL-ID' in r] r01_count = len(r01_recs) if r01_count < 50: template = r01_recs[-1].copy() if r01_recs else {} needed = 50 - r01_count for i in range(needed): nr = {} for key, val in template.items(): if not key.startswith('R02'): nr[key] = val nr['R01APPL-ID'] = f"X50{str(i).zfill(5)}" if 'R01EMP-ID' in nr: nr['R01EMP-ID'] = str(int(str(nr.get('R01EMP-ID', '0') or '0')) + i + 10000).zfill(8) recs.append(nr) r01_recs = [r for r in recs if 'R01APPL-ID' in r] logger.info(f" Coverage #10T: added {needed} R01-only records -> {len(r01_recs)} total") # #8 T: Two R01 records with same APPL-ID -> 2nd INSERT collides -> UPDATE if len(r01_recs) >= 4: dup_appl_id = 'COLISN01' for idx in (2, 3): r01_recs[idx]['R01APPL-ID'] = dup_appl_id if 'R02APPL-ID' in r01_recs[idx]: r01_recs[idx]['R02APPL-ID'] = dup_appl_id logger.info(f" Coverage #8T: set APPL-ID={dup_appl_id} on records [2]&[3] for INSERT duplicate") # #11 T: Two R01 records with same (EMP-ID, APPL-DATE, OVT-TYPE) if len(r01_recs) >= 2: match_emp = r01_recs[1].get('R01EMP-ID', '00000000').strip() or '00000000' match_date = r01_recs[1].get('R01APPL-DATE', '00000000').strip() or '00000000' match_type = r01_recs[1].get('R01OVT-TYPE', '1').strip() or '1' r01_recs[0]['R01EMP-ID'] = match_emp r01_recs[0]['R01APPL-DATE'] = match_date r01_recs[0]['R01OVT-TYPE'] = match_type r01_recs[1]['R01EMP-ID'] = match_emp r01_recs[1]['R01APPL-DATE'] = match_date r01_recs[1]['R01OVT-TYPE'] = match_type logger.info( f" Coverage #11T: unified (EMP={match_emp} DATE={match_date}" f" TYPE={match_type}) for R01 records [0]&[1]" ) if apply_abnormal: # #14 T: Last R02 record has non-existent APPL-ID -> orphan cancel ABEND # NOTE: ABEND prevents 3000STPSOR (#19 T/F); covered by normal scenario. r02_recs = [r for r in recs if 'R02APPL-ID' in r] if r02_recs: r02_recs[-1]['R02APPL-ID'] = 'ZZZZZZZZ' logger.info(f" Coverage #14T: set last R02 APPL-ID='ZZZZZZZZ' for orphan cancel") # 全レコードの(EMP_ID, DATE)重複チェック（PK衝突→ABEND防止） self._deduplicate_r01_pk(recs) # 出力先ディレクトリ（シナリオ毎に分離） run_label = f"run_{scenario.id}" if scenario else "" output_root = self.work_dir / run_label if scenario else self.work_dir input_dir = output_root / "main" / "input" input_dir.mkdir(parents=True, exist_ok=True) # フラットファイル書き出し（全シナリオ同一） flats = write_all_files(recs, src_text, input_dir, copybook_dirs=[str(d) for d in self.copybook_dirs]) # SYSIN 設定ファイル生成（シナリオ毎に run_cfg を渡す） run_cfg = None if scenario: run_cfg = { "period": scenario.sysin.period, "include_invalid_period": scenario.sysin.include_invalid_period, "modes": scenario.sysin.modes, "final_mode": scenario.sysin.final_mode, } sysin_path = write_sysin_file(recs, src_text, input_dir, copybook_dirs=[str(d) for d in self.copybook_dirs], run_cfg=run_cfg) if sysin_path: logger.info(f" SYSIN file written: {sysin_path}") flats.append(("SYSIN", sysin_path, 0)) # Pre-populate MONTHLY_ABSENCE with matching EMP-ID/YEAR-MONTH for # SELECT COUNT(*) → HV-CNT > 0 → UPDATE path (DP#27). Must run AFTER # write_all_files (so the R01 flat file exists). db_for_seed = self._current_db_path or self.db_path self._seed_matching_monthly_rows(db_for_seed, recs, max_seed=20, r01_dir=input_dir) # ── JSON 出力（Java 検証用） ── try: pp = preprocess(src_text, extra_search_paths=cbd) data_div = extract_data_division(pp) data_fields = parse_data_division(data_div) if data_div else [] fdict = [] 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 fdict.append(_entry) fdict = expand_occurs(fdict) proc_div = extract_procedure_division(pp) branch_tree, assignments = build_branch_tree_fallback(proc_div, fdict) sql_assigns = extract_sql_assignments(src_text) for tgt, asgn_list in sql_assigns.items(): for asgn in asgn_list: assignments.setdefault(tgt, []).append(asgn) # fd_fields / field_to_fd file_sec = parse_file_section(pp) or {} fd_fields = {} field_to_fd = {} 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, fdict): 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 open_dir = scan_open_statements(proc_div) if proc_div else {} # Roles + path info + termination types roles = classify_field_roles(branch_tree, assignments, fdict, source=src_text, proc_text=proc_div) branch_paths = mcdc_enum_paths(branch_tree, fdict) path_infos = [(c, a, get_term_type(c)[1]) for c, a in branch_paths] json_records, _, term_types = generate_records( path_infos, fdict, assignments, file_sec=file_sec) # DB input for JSON data_div2, declared_columns = strip_exec_sql_from_data_div(data_div) declared_columns = self._merge_schema_columns(declared_columns) sql_meta = collect_sql_meta(assignments, declared_columns) db_input = None if sql_meta: db_input = build_db_input( branch_paths, fdict, assignments, sql_meta, declared_columns, records=recs, insert_pk=self._insert_pk_map()) # Write main JSON（シナリオ毎に分離） json_outdir = output_root / "main" / "json" json_outdir.mkdir(parents=True, exist_ok=True) json_path = json_outdir / f"{self.program_id}.json" output_json(json_records, json_path, roles, fd_fields=fd_fields, field_to_fd=field_to_fd, open_dir=open_dir, term_types=term_types, db_input=db_input, data_fields=fdict) logger.info(f" JSON output: {json_path}") flats.append(("JSON", json_path, 0)) except Exception as ej: logger.warning(f" JSON output skipped: {ej}") self.generated_records = recs db_for_result = str(self._current_db_path or self.db_path) return DbPipelineResult( self.program_id, 2, True, data={"records": len(recs), "flat_files": len(flats), "db_path": db_for_result}, ) except Exception as e: return DbPipelineResult(self.program_id, 2, False, str(e)) # ── Step 3: COBOL 実行 ── def step3_run_cobol(self, scenario: ScenarioDef | None = None) -> DbPipelineResult: """COBOL DB プログラム実行（環境変数で入出力先を振り分け） Args: scenario: 多輪実行時のシナリオ。None=単輪。 """ if not self.exe_path or not self.exe_path.exists(): return DbPipelineResult(self.program_id, 3, False, "exe not found (run step1 first)") # シナリオ毎の出力先 run_label = f"run_{scenario.id}" if scenario else "main" run_dir = self.runtime_dir / run_label input_dir = run_dir / "input" output_dir = run_dir / "output" gcov_dir = self.runtime_dir.parent / "gcov" input_dir.mkdir(parents=True, exist_ok=True) output_dir.mkdir(parents=True, exist_ok=True) gcov_dir.mkdir(parents=True, exist_ok=True) # シナリオ毎の CWD = run_{id}/、単輪時は runtime_dir 直下 cwd = run_dir # 入力ファイル（work_dir/run_{id}/main/input/ → runtime/run_{id}/main/input/） gen_input_dir = self.work_dir / f"run_{scenario.id}" / "main" / "input" if scenario else self.work_dir / "main" / "input" if gen_input_dir.exists(): for f in gen_input_dir.iterdir(): if f.is_file(): (input_dir / f.name).write_bytes(f.read_bytes()) # JSON 出力（work_dir/run_{id}/main/json/ → runtime/run_{id}/main/json/） gen_json_dir = self.work_dir / f"run_{scenario.id}" / "main" / "json" if scenario else self.work_dir / "main" / "json" if gen_json_dir.exists(): json_dir = run_dir / "main" / "json" json_dir.mkdir(parents=True, exist_ok=True) for f in gen_json_dir.iterdir(): if f.is_file() and f.suffix.lower() == '.json': (json_dir / f.name).write_bytes(f.read_bytes()) # Scan ASSIGN TO + OPEN direction → build env overrides assign_map = self._scan_assign_to() env_overrides = {} for fname, direction in assign_map.items(): if direction == "INPUT": env_overrides[fname] = os.path.join("input", fname) else: env_overrides[fname] = os.path.join("output", fname) # シナリオ毎の DB パス db_path = self._current_db_path or self.db_path # GIXSQL_DB_PATH が効かないため、デフォルト DB にシナリオ DB をコピーする if scenario is not None and db_path != self.db_path: if db_path.exists(): if self.db_path.exists(): self.db_path.unlink() shutil.copy2(str(db_path), str(self.db_path)) db_path = self.db_path # CONNECT TO 'data/kin.db' のパス解釈に備え CWD にもコピー（単輪/多輪共通） 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)) # gixsql regex requires sqlite://host/path (single segment, no dots). # Copy to CWD/kin (no extension) for sqlite://localhost/kin. # 同时复制到 kin.db，确保 gixsql 打开的是有 PRIMARY KEY 的版本 cwd_kin = cwd / "kin" if cwd_kin.exists(): cwd_kin.unlink() shutil.copy2(str(db_path), str(cwd_kin)) cwd_kin_db = cwd / "kin.db" if cwd_kin_db.exists(): cwd_kin_db.unlink() shutil.copy2(str(db_path), str(cwd_kin_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 # Create parent directories for all ASSIGN TO files (COBOL needs them to exist) for fname, direction in assign_map.items(): if os.sep in fname or '/' in fname: parent = cwd / os.path.dirname(fname) parent.mkdir(parents=True, exist_ok=True) # Subprogram DLLs cobol_bin = Path(self.cobol_src_dir).parent / "bin" # command_line: scenario-level (if set) overrides program-level default cmd_line = self.schema.command_line if scenario and scenario.command_line is not None: cmd_line = scenario.command_line command_args = cmd_line.split() if cmd_line else None result = self.runner.run( self.exe_path, cwd, db_path, input_dir=None, cobol_lib_path=str(cobol_bin) if cobol_bin.exists() else None, env_overrides=env_overrides, command_args=command_args, ) log_dir = self.runtime_dir.parent / "logs" log_dir.mkdir(parents=True, exist_ok=True) log_dir.joinpath(f"{run_label or self.program_id}.log").write_text( result.log, encoding='utf-8') # .gcda を gcov/ にコピー（シナリオ毎に gcov/run_{id}/） # GnuCOBOL は .gcno が生成された CWD (= compile CWD = exe_dir) に .gcda を書き出す。 # 複数シナリオで .gcno は共有されるため COPY で行う（MOVE 不可）。 gcda_src_dirs = [cwd] # ランタイム CWD exe_dir_for_gcda = self.work_dir / "bin" if exe_dir_for_gcda.exists() and exe_dir_for_gcda not in gcda_src_dirs: gcda_src_dirs.append(exe_dir_for_gcda) if scenario is None: gcda_src_dirs.append(self.runtime_dir) # 従来互換 gcda_dst_dir = gcov_dir / run_label gcda_dst_dir.mkdir(parents=True, exist_ok=True) for sd in gcda_src_dirs: for f in sd.glob("*.gcda"): if f.is_file() and f.stat().st_size > 0: dst = gcda_dst_dir / f.name if scenario or not dst.exists() or f.stat().st_mtime > dst.stat().st_mtime: try: shutil.copy2(str(f), str(dst)) except PermissionError: pass for f in sd.glob("*.gcno"): if f.is_file() and f.stat().st_size > 0: dst = gcda_dst_dir / f.name try: shutil.copy2(str(f), str(dst)) except PermissionError: pass # gixsql 通过 GIXSQL_DB_PATH 把 COBOL 结果写回 db_path（运行时唯一真实落盘文件）。 # CWD 下的 kin/kin.db 是运行前从种子库复制的快照；这里运行后回拷， # 保证 _java_compare_scenario 读取的是 COBOL 的真实输出而不是过期的种子快照。 if db_path and db_path.exists(): for _dst in (cwd / 'kin', cwd / 'kin.db', cwd / 'data' / 'kin.db'): try: _dst.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(str(db_path), str(_dst)) except OSError as _e: logger.warning(f" 运行后 DB 回拷失败 {_dst}: {_e}") return DbPipelineResult( self.program_id, 3, result.success, data={"returncode": result.returncode, "log": result.log[:500], "input_dir": str(input_dir), "output_dir": str(output_dir), "gcov_dir": str(gcov_dir)}, ) # ── マルチラン gcov マージ ── def _merge_multi_run_gcov(self) -> dict[int, int] | None: """Run gcov per scenario, parse results, merge {line: count} dicts. Returns merged gcov_data or None if no multi-run data available. """ from cobol_testgen.gcov import run_gcov, parse_cbl_gcov gcov_dir = self.runtime_dir / "gcov" run_dirs = sorted(gcov_dir.glob("run_*")) if len(run_dirs) <= 1: return None # Ensure .gcno is in each run dir (copy from compile CWD if needed) bin_gcno = self.work_dir / "bin" if bin_gcno.exists(): for sd in run_dirs: for f in bin_gcno.glob("*.gcno"): dst = sd / f.name if not dst.exists(): shutil.copy2(str(f), str(dst)) merged_data = _merge_run_dirs_gcov(gcov_dir, f"{self.program_id}_pp") logger.info(f" Merged gcov from {len(run_dirs)} runs ({len(merged_data)} lines)") return merged_data # ── カバレッジレポート（パイプライン外、オプション） ── def generate_coverage_report(self, output_dir: str | Path | None = None) -> DbPipelineResult: """COBOL 実行後：gcov データ収集 + 静的パスとマージし HTML レポート""" try: if not self.exe_path or not self.exe_path.exists(): # Fallback: look for exe in standard build location fallback = self.work_dir / "bin" / f"{self.program_id}.exe" if fallback.exists(): self.exe_path = fallback else: return DbPipelineResult(self.program_id, 0, False, f"exe not found at {self.exe_path} or {fallback}") if output_dir is None: v3_root = Path(__file__).parent output_dir = v3_root / "output" / self.program_id / "coverage" output_dir = Path(output_dir) # 1. Use pre-merged multi-run gcov data if available (skip gcov re-run) if self._multi_run_gcov_data is not None: gcov_data = self._multi_run_gcov_data # Subprogram gcov is kept separate: SUB*.cbl line numbers are # plain integers that collide with the main program's (e.g. # SUB04CHK line 167=0 would overwrite main line 167=25 and # wipe real coverage). Stored per-subprogram for reference. gcov_dir = self.runtime_dir.parent / "gcov" self._sub_gcov_data = {} for sub in self.schema.subprograms: sub_merged = _merge_run_dirs_gcov(gcov_dir, sub) if sub_merged: self._sub_gcov_data[sub] = sub_merged else: # Single-run: collect .gcno/.gcda and run gcov gcov_dir = self.runtime_dir / "gcov" gcov_dir.mkdir(parents=True, exist_ok=True) v3_root = Path(__file__).parent extra_search = list(gcov_dir.glob("run_*")) + [self.work_dir / "bin"] for search_dir in (v3_root, self.work_dir, self.runtime_dir, gcov_dir, Path.home(), *extra_search): for f in search_dir.glob("*.gcda"): if f.stat().st_size > 0: dst = gcov_dir / f.name if not dst.exists() or f.stat().st_mtime > dst.stat().st_mtime: try: shutil.copy2(str(f), str(dst)) except PermissionError: pass for f in search_dir.glob("*.gcno"): if f.stat().st_size > 0: dst = gcov_dir / f.name try: shutil.copy2(str(f), str(dst)) except PermissionError: pass # Count what we have gcno_gcda_count = 0 for ext in (".gcno", ".gcda"): for f in gcov_dir.glob(f"*{ext}"): if f.stat().st_size > 0 and (f.name.startswith(self.program_id) or f.name.startswith("SUB")): gcno_gcda_count += 1 if gcno_gcda_count == 0: for sd in (v3_root, self.work_dir, self.runtime_dir, gcov_dir): for ext2 in (".gcno", ".gcda"): files = list(sd.glob(f"*{ext2}")) logger.error(f"gcov-check: {sd}\\*{ext2} -> {len(files)} files: {[f.name for f in files[:5]]}") return DbPipelineResult(self.program_id, 0, False, "no .gcno/.gcda found (--coverage missing?)") # 3. Parse gcov data gcov_data = run_gcov(f"{self.program_id}_pp", str(gcov_dir)) if not gcov_data: gcov_data = run_gcov(self.program_id, str(gcov_dir)) # Subprogram gcov kept separate (line numbers collide with main). self._sub_gcov_data = {} for sub in self.schema.subprograms: sd = run_gcov(sub, str(gcov_dir)) if sd: self._sub_gcov_data[sub] = sd # 4. Static branch tree from step2 st = self.generated_structure branch_tree = st.get("branch_tree_obj") if st else None if not branch_tree: return DbPipelineResult(self.program_id, 0, True, data={"gcov_lines": len(gcov_data), "note": "no branch tree — gcov data only"}) # 5. Re-parse fields (same as generate_data) src_text = self.src_path.read_text(encoding="utf-8-sig") cbd = [str(d) for d in self.copybook_dirs] pp = preprocess(src_text, extra_search_paths=cbd) data_div = extract_data_division(pp) data_fields = parse_data_division(data_div) if data_div else [] fdict = [] for idx, f in enumerate(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 fdict.append(entry) fdict = expand_occurs(fdict) # 6. Enumerate paths branch_paths = mcdc_enum_paths(branch_tree, fdict) # 7. Read preprocessed source for gcov line number matching gcov_source = None if self.pp_path and self.pp_path.exists(): gcov_source = self.pp_path.read_text(encoding="utf-8") # 8. Generate merged HTML (use gcov_source for line numbers) output_dir.mkdir(parents=True, exist_ok=True) prefix = str(output_dir / self.program_id) cov_result = run_coverage( branch_tree, branch_paths, fdict, src_text, prefix, index_relpath="index.html", gcov_data=gcov_data or None, gcov_source=gcov_source, ) generate_coverage_index([cov_result], str(output_dir.parent)) # Clean up .gcno/.gcda from v3_root + CWD (avoid accumulation) _v3_root = Path(__file__).parent for clean_dir in (_v3_root, Path.cwd()): if clean_dir == gcov_dir: continue for ext in (".gcno", ".gcda"): for f in clean_dir.glob(f"*{ext}"): try: f.unlink() except PermissionError: pass total = cov_result.get("total_branches", 0) covered = cov_result.get("covered_branches", 0) pct = covered / total * 100 if total else 0 self._last_coverage_dict = cov_result return DbPipelineResult( self.program_id, 0, True, data={ "gcov_lines": len(gcov_data), "coverage": f"{covered}/{total} ({pct:.1f}%)", "reports": str(output_dir), "_cov_dict": cov_result, }, ) except Exception as e: logger.exception("generate_coverage_report failed") return DbPipelineResult(self.program_id, 0, False, str(e)) # ── DB-Java 通用适配（文件式 Java 运行 + DB 表比对） ── # 依赖 JavaSrc 工程（target/classes + target/*.jar + lib/*.jar，显式主类）， # 无需 mvn。Java 从“场景初始 DB 快照 + 同入力”出发，最终 DB/W01 与 COBOL 结果比对。 def _java_cfg(self): if self._java_cfg_cache is None: self._java_cfg_cache = None try: from cobol_testgen.runner import _discover_java root = Path(self.cobol_src_dir).parent / 'JavaSrc' cfg = _discover_java(str(root), self.program_id) self._java_cfg_cache = cfg if cfg: logger.info(f" DB-Java: 发现主类 {cfg['main_class']}") except Exception as e: # noqa: BLE001 logger.warning(f" DB-Java 发现失败: {e}") return self._java_cfg_cache def _java_db_rel_path(self) -> Optional[str]: """扫描 Java 主类源码中 jdbc:sqlite:<相对路径>，返回 DB 相对路径。""" cfg = self._java_cfg() if not cfg: return None try: main = Path(cfg['root']) / 'src' / f"{cfg['main_class']}.java" txt = main.read_text(encoding='utf-8', errors='replace') m = re.search(r'jdbc:sqlite:([^\s"\'\)]+)', txt) if m: return m.group(1).replace('\\', os.sep) except Exception: # noqa: BLE001 pass return None def _java_scenario_label(self, scenario: ScenarioDef | None) -> str: return f"run_{scenario.id}" if scenario else "main" def _java_prepare_scenario(self, scenario: ScenarioDef | None) -> Optional[dict]: """每场景准备 Java 运行目录：初始 DB 快照 + 入力 txt。""" cfg = self._java_cfg() if not cfg: return None label = self._java_scenario_label(scenario) java_dir = self.runtime_dir.parent / 'java' / label data_dir = java_dir / 'data' if java_dir.exists(): shutil.rmtree(str(java_dir)) data_dir.mkdir(parents=True, exist_ok=True) # 1) 初始 DB 快照（Step2 初始化后、COBOL 运行前的场景 DB） # db_rel 为相对 Java CWD(=java_dir) 的路径（如 data/OVERTIME.DB），直接落到 java_dir 下 db_rel = self._java_db_rel_path() # 优先使用 COBOL 运行前保存的初始快照，避免 Java 在 COBOL 修改后的 DB 上重复处理 label = self._java_scenario_label(scenario) db_src = self._java_initial_db_snapshots.get(label) or self._current_db_path or self.db_path if db_rel and db_src and Path(db_src).exists(): dst = java_dir / db_rel dst.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(str(db_src), str(dst)) logger.info(f" DB-Java[{label}]: 初始 DB 快照 -> {dst}") # 2) 入力 txt（80B 定长 → 行） gen_input_dir = (self.work_dir / f"run_{scenario.id}" / "main" / "input") if scenario \ else (self.work_dir / "main" / "input") written: list[str] = [] layouts: dict = {} try: from cobol_testgen.flatfile import analyze_fd_layout if self.src_path and self.src_path.exists(): layouts = analyze_fd_layout( self.src_path.read_text(encoding='utf-8-sig'), [str(d) for d in self.copybook_dirs]) except Exception as e: # noqa: BLE001 logger.warning(f" DB-Java 布局解析失败: {e}") for assign, lay in layouts.items(): direction = str(lay.get('direction', 'INPUT')).upper() if direction not in ('INPUT', 'I-O'): continue recs = lay.get('records') or [] rlen = recs[0].get('record_length', 0) if recs else 0 src = gen_input_dir / assign if not src.exists() or rlen <= 0: continue data = src.read_bytes() lines = [] i = 0 while i + rlen <= len(data): lines.append(data[i:i + rlen].decode('utf-8', 'replace')) i += rlen txt = '\n'.join(lines) + ('\n' if lines else '') (data_dir / f'{assign}.txt').write_text(txt, encoding='utf-8') written.append(assign) logger.info(f" DB-Java[{label}]: 入力 {assign}.txt {len(lines)} 行") if not written and not db_rel: return None return {'label': label, 'java_dir': str(java_dir), 'data_dir': str(data_dir), 'db_rel': db_rel, 'inputs': written, 'cfg': cfg} def _java_run_scenario(self, scenario: ScenarioDef | None, info: dict) -> dict: """运行 Java 主类（文件式），返回 rc/日志。异常与 rc!=0 不抛出，仅记录。""" cfg = info['cfg'] from cobol_testgen.runner import _find_java_exe java_exe = _find_java_exe() cmd = [java_exe, '-cp', cfg['classpath'], cfg['main_class']] # command_line: scenario-level (if set) overrides program-level default cmd_line = self.schema.command_line if scenario and scenario.command_line is not None: cmd_line = scenario.command_line if cmd_line: cmd.extend(cmd_line.split()) log_path = self.runtime_dir.parent / 'logs' / f"{self.program_id}_java_{info['label']}.log" log_path.parent.mkdir(parents=True, exist_ok=True) orig = os.getcwd() p = None try: os.chdir(info['java_dir']) p = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=180) except subprocess.TimeoutExpired: log_path.write_text(f"COMMAND: {' '.join(cmd)}\nTIMEOUT\n", encoding='utf-8') return {'rc': -1, 'log': 'TIMEOUT'} except Exception as e: # noqa: BLE001 log_path.write_text(f"COMMAND: {' '.join(cmd)}\nRUN ERROR: {e}\n", encoding='utf-8') return {'rc': -2, 'log': str(e)} finally: os.chdir(orig) log = f"COMMAND: {' '.join(cmd)}\nRETURNCODE: {p.returncode}\n\nSTDOUT:\n{p.stdout}\n\nSTDERR:\n{p.stderr}" log_path.write_text(log, encoding='utf-8') logger.info(f" DB-Java[{info['label']}]: rc={p.returncode}") return {'rc': p.returncode, 'log': p.stdout + p.stderr} def _collect_java_outputs(self, info: dict) -> None: """收集 Java 出力：txt 出力（排除入力）+ Java 运行后 DB 副本。""" label = info['label'] java_out_dir = self.runtime_dir.parent / 'java' / label / 'output' java_out_dir.mkdir(parents=True, exist_ok=True) inputs = set(info.get('inputs') or []) data_dir = Path(info['data_dir']) for f in sorted(data_dir.glob('*.txt')): if f.name in inputs: continue try: shutil.copy2(str(f), str(java_out_dir / f.name)) except OSError: pass if info.get('db_rel'): src = Path(info['java_dir']) / info['db_rel'] if src.exists(): try: shutil.copy2(str(src), str(java_out_dir / f"{self.program_id}_java.db")) except OSError: pass @staticmethod def _db_table_snap(db_path) -> dict[str, list[dict]]: snap: dict[str, list[dict]] = {} if not db_path or not Path(db_path).exists(): return snap try: conn = sqlite3.connect(str(db_path)) conn.row_factory = sqlite3.Row cur = conn.cursor() tables = [r[0] for r in cur.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'")] for t in tables: rows = [] try: for row in cur.execute('SELECT * FROM "%s"' % t): rows.append(dict(row)) except sqlite3.OperationalError: continue snap[t] = rows conn.close() except Exception as e: # noqa: BLE001 logger.warning(f" 读 DB 失败 {db_path}: {e}") return snap _IGNORE_DB_COLS = re.compile( r'(updated_at|created_at|inserted_at|deleted_at|^.*timestamp.*$)', re.IGNORECASE) @staticmethod def _norm_db_value(v): """通用归一化：数值四舍五入到 6 位消除浮点噪声；字符串去两端空白。""" s = str(v) t = s.strip() try: f = float(t) except ValueError: return ('s', t) return ('n', round(f, 6)) def _java_compare_scenario(self, scenario: ScenarioDef | None, info: dict, jrep: dict) -> dict: """COBOL 结果 DB（run_/kin） vs Java 结果 DB 逐表比对。""" from collections import Counter label = info['label'] run_dir = self.runtime_dir / label cobol_db = run_dir / 'kin' if not cobol_db.exists(): cobol_db = run_dir / 'data' / 'kin.db' java_db = None if info.get('db_rel'): jp = Path(info['java_dir']) / info['db_rel'] if jp.exists(): java_db = jp if not java_db: return {'label': label, 'rc': jrep.get('rc', -1), 'error': 'Java DB 未生成（无 db_rel 或运行失败）', 'summaries': []} c_tab = self._db_table_snap(cobol_db) j_tab = self._db_table_snap(java_db) keys = sorted(set(c_tab) | set(j_tab)) def _clean(row: dict) -> dict: return {k: v for k, v in row.items() if not self._IGNORE_DB_COLS.search(k)} def _canon(rows): c = Counter() for r in rows: row = _clean(r) # 忽略全为空的键（仅剩时间戳被剔除后的空行） items = sorted((str(k),) + (self._norm_db_value(v),) for k, v in row.items()) c[tuple(items)] += 1 return c summaries = [] for t in keys: cc = _canon(c_tab.get(t, [])) jc = _canon(j_tab.get(t, [])) allk = set(cc) | set(jc) matched = sum(min(cc[k], jc[k]) for k in allk) # 不一致行数 = 两侧行数较小者中无法配成完全一致的行数（避免同一差异行重复计数） min_rows = min(len(c_tab.get(t, [])), len(j_tab.get(t, []))) mismatched = max(0, min_rows - matched) samples = [] for k in sorted(allk, key=lambda x: str(x)): if cc[k] != jc[k]: side = 'COBOL' if cc[k] > jc[k] else 'Java' disp = {} for fld, typed in k: if typed[0] == 'n': fv = typed[1] disp[fld] = str(int(fv)) if float(fv).is_integer() else str(fv) else: disp[fld] = typed[1] samples.append({'side': side, 'diff': abs(cc[k] - jc[k]), 'row': {kk: str(vv)[:30] for kk, vv in disp.items()}}) if len(samples) >= 3: break summaries.append({'table': t, 'cobol_count': len(c_tab.get(t, [])), 'java_count': len(j_tab.get(t, [])), 'matched': matched, 'mismatched': mismatched, 'samples': samples}) return {'label': label, 'rc': jrep.get('rc', -1), 'summaries': summaries} def _write_db_java_report(self, cobol_results: list) -> None: lines = [] lines.append(f"# {self.program_id} COBOL vs Java 出力比对测试报告（DB 管道）") lines.append('') lines.append(f"- 程序ID: {self.program_id}") lines.append('- 比对基准: COBOL 运行后 DB 表（run_<场景>/kin） vs Java 从同初始 DB 运行后 DB 表') lines.append('') lines.append('## 1. COBOL 执行结果（DB 管道 run_all）') lines.append('') lines.append('| 组 | returncode | 判定 |') lines.append('|----|-----------|------|') for label, res in cobol_results: rc = (res.data or {}).get('returncode', '-') mark = '通过' if res.success else '差异' lines.append(f"| {label} | {rc} | {mark} |") lines.append('') lines.append('## 2. Java 执行与 DB 表比对（逐场景逐表）') lines.append('') for rep in self._db_java_results: lines.append(f'### 2.{self._db_java_results.index(rep) + 1} 组 `{rep["label"]}` (Java rc={rep.get("rc", "N/A")})') lines.append('') if rep.get('error'): lines.append(f'- {rep["error"]}') lines.append('') continue if not rep.get('summaries'): lines.append('（无可比对 DB 表）') lines.append('') continue lines.append('| 表 | COBOL行数 | Java行数 | 一致 | 不一致 | 判定 |') lines.append('|----|----------|----------|------|--------|------|') for s in rep['summaries']: count_ok = (s['cobol_count'] == s['java_count']) verdict = '一致' if (count_ok and s['mismatched'] == 0) else '不一致' lines.append(f"| {s['table']} | {s['cobol_count']} | {s['java_count']} " f"| {s['matched']} | {s['mismatched']} | {verdict} |") diff = [s for s in rep['summaries'] if s['mismatched'] > 0 or s['cobol_count'] != s['java_count']] if diff: lines.append('') lines.append('**差异示例（最多 3 行/表，COBOL/Java 各侧存在但内容不同）**：') for s in diff: for smp in s['samples']: lines.append(f"- {s['table']} [{smp['side']}侧 x{smp['diff']}]: " f"{smp['row']}") lines.append('') all_fd = [s for rep in self._db_java_results for s in rep.get('summaries', [])] tot_m = sum(s['matched'] for s in all_fd) tot_mis = sum(s['mismatched'] for s in all_fd) any_count_diff = any(s['cobol_count'] != s['java_count'] for s in all_fd) lines.append('## 3. 总结') lines.append('') if not all_fd: lines.append('- 无可比对的 DB 记录（Java 未产出或发现失败）。') elif tot_mis == 0 and not any_count_diff: lines.append(f'- COBOL 与 Java 全部 DB 记录一致（一致 {tot_m} 行，不一致 0 行）。') lines.append('- 判定: **通过**') else: lines.append(f'- COBOL 与 Java DB 记录存在差异：一致 {tot_m} 行，内容不一致 {tot_mis} 行' f"{'（另有行数不一致）' if any_count_diff else ''}。") lines.append('- 判定: **存在差异**，详见上文逐场景逐表明细。') lines.append('') lines.append('> 说明: Java 从各场景“初始 DB 快照 + 同入力”运行；Java 异常/rc!=0 已作为该场景结果记录（不阻断管道）。') lines.append('') report_path = self.runtime_dir.parent / f'{self.program_id}_测试报告.md' report_path.write_text('\n'.join(lines), encoding='utf-8') logger.info(f" DB-Java 比对报告: {report_path}") def _run_db_java_scenario(self, scenario: ScenarioDef | None) -> None: """每场景执行：准备(初始DB快照+入力) -> 运行 Java -> 收集出力 -> 比对记录。""" try: info = self._java_prepare_scenario(scenario) if not info: logger.info(" DB-Java: 未发现 Java 对应程序或入力/DB 不可用，跳过") return jrep = self._java_run_scenario(scenario, info) self._collect_java_outputs(info) cres = self._java_compare_scenario(scenario, info, jrep) self._db_java_results.append(cres) except Exception as e: # noqa: BLE001 logger.warning(f" DB-Java 场景 {scenario.id if scenario else 'main'} 失败: {e}") # ── Step 4: DB → Java 中介データ ── def step4_extract_intermediate(self) -> DbPipelineResult: """SQLite → JSON 中介データ抽出（Step 4: DB→Java中介データ）""" db_path = self._current_db_path or self.db_path if not db_path or not db_path.exists(): return DbPipelineResult(self.program_id, 4, False, "db not found (run step3 first)") try: conn = sqlite3.connect(str(db_path)) conn.row_factory = sqlite3.Row # Read from actual COBOL SQL tables (using sql_name or name) output_tables = {} for table in self.schema.db_tables: sql_name = table.sql_name or table.name try: rows = conn.execute(f"SELECT * FROM [{sql_name}]").fetchall() output_tables[table.name] = [dict(r) for r in rows] except sqlite3.OperationalError: output_tables[table.name] = [] conn.close() w01_path = self.work_dir / "intermediate" / f"{self.program_id}_W01.json" w01_path.parent.mkdir(parents=True, exist_ok=True) meta = { "program_id": self.program_id, "tables": output_tables, } w01_path.write_text(json.dumps(meta, ensure_ascii=False, indent=2)) self.java_input_path = w01_path return DbPipelineResult( self.program_id, 4, True, data={"tables": len(output_tables), "w01_path": str(w01_path)}, ) except Exception as e: return DbPipelineResult(self.program_id, 4, False, str(e)) # ── Step 5: Java 実行 ── def step5_run_java(self, java_cmd: str = "java", java_jar: str | Path | None = None, java_src_dir: str | Path | None = None) -> DbPipelineResult: """Java プログラム実行 Args: java_cmd: Java可执行文件路径 java_jar: Java JAR文件路径（如果提供） java_src_dir: Java源代码目录（如果提供，将先编译） """ if not self.java_input_path or not self.java_input_path.exists(): return DbPipelineResult(self.program_id, 5, False, "intermediate data not found (run step4 first)") run_label = "main" # 默认场景 java_out = self.runtime_dir.parent / "java" / run_label / "output" java_out.mkdir(parents=True, exist_ok=True) # 如果提供了Java源代码目录，先编译 if java_src_dir and not java_jar: try: from runners import NativeJavaRunner runner = NativeJavaRunner() build_result = runner.compile(str(java_src_dir)) if not build_result.success: return DbPipelineResult( self.program_id, 5, False, message=f"Java compilation failed: {build_result.log[:200]}", data={"log": build_result.log[:500]} ) java_jar = build_result.artifact_path except Exception as e: return DbPipelineResult( self.program_id, 5, False, message=f"Java compilation error: {str(e)}", data={"error": str(e)} ) if java_jar: # 使用JAR文件执行 cmd = [java_cmd, "-jar", str(java_jar), "-i", str(self.java_input_path), "-o", str(java_out)] else: # 没有提供Java JAR或源代码 return DbPipelineResult( self.program_id, 5, False, message="No Java JAR or source directory provided", data={"error": "No Java JAR or source"} ) try: r = subprocess.run(cmd, capture_output=True, timeout=60) log = (r.stdout.decode("utf-8", "replace") + "\n" + r.stderr.decode("utf-8", "replace")) ok = r.returncode == 0 self.java_output_path = java_out return DbPipelineResult( self.program_id, 5, ok, data={"returncode": r.returncode, "log": log[:500]}, ) except subprocess.TimeoutExpired: return DbPipelineResult(self.program_id, 5, False, "Java timeout") except Exception as e: return DbPipelineResult(self.program_id, 5, False, f"Java execution error: {str(e)}") # ── Step 6: 検証 ── def step6_verify(self) -> VerificationRun: """Java 出力と COBOL 期待値を比較""" db_path = self._current_db_path or self.db_path vr = VerificationRun( program=self.program_id, runner="gixsql", gixsql_version="0.9.1", sqlite_path=str(db_path) if db_path else "", step_reached=6, ) # 1. 读取COBOL输出（从DB表） cobol_records = [] if db_path and db_path.exists(): try: after_tables = self.runner.read_db_tables( db_path, [t.name for t in self.schema.db_tables], ) for table_data in after_tables: vr.debug[f"table_{table_data.table_name}_rows"] = len(table_data.rows) # 将表数据转换为记录格式 for row in table_data.rows: record = {} for i, col in enumerate(table_data.columns): record[col] = row[i] if i < len(row) else "" cobol_records.append(record) except Exception as e: vr.debug["cobol_read_error"] = str(e) # 2. 读取Java输出（从文件） java_records = [] if self.java_output_path and self.java_output_path.exists(): try: java_files = list(self.java_output_path.glob("*.txt")) + \ list(self.java_output_path.glob("*.json")) vr.debug["java_output_files"] = [str(f) for f in java_files] for java_file in java_files: if java_file.suffix == ".json": with open(java_file, 'r', encoding='utf-8') as f: data = json.load(f) if isinstance(data, list): java_records.extend(data) else: java_records.append(data) elif java_file.suffix == ".txt": with open(java_file, 'r', encoding='utf-8') as f: for line in f: line = line.strip() if line: java_records.append({"raw": line}) except Exception as e: vr.debug["java_read_error"] = str(e) # 3. 比较COBOL和Java输出 if not cobol_records and not java_records: vr.status = "PASS" vr.exit_code = 0 return vr if not cobol_records: vr.status = "MISMATCH" vr.exit_code = 1 vr.debug["error"] = "No COBOL records found" return vr if not java_records: vr.status = "MISMATCH" vr.exit_code = 1 vr.debug["error"] = "No Java records found" return vr # 4. 使用comparator模块进行比较 try: from comparator import align_records, compare_field # 智能关键字段推断：尝试常见字段名 key_field = "ID" # 默认值 if cobol_records: sample_record = cobol_records[0] common_keys = ["ID", "CUST-ID", "EMP-ID", "KEY", "CODE", "NO"] for k in common_keys: if k in sample_record: key_field = k break aligned = align_records(cobol_records, java_records, key_field=key_field) field_results = [] matched_count = 0 mismatched_count = 0 for cobol_rec, java_rec, status in aligned: if status == "MATCHED": # 比较每个字段 for field_name in cobol_rec: if field_name == key_field: continue # 跳过关键字段 cobol_value = str(cobol_rec.get(field_name, "")) java_value = str(java_rec.get(field_name, "")) # 确定字段类型 field_name_upper = field_name.upper() if any(c in field_name_upper for c in ["AMT", "AMOUNT", "PRICE", "COST", "TOTAL", "QTY", "QUANTITY", "NUM", "NUMBER", "COUNT"]): field_type = "decimal" elif any(c in field_name_upper for c in ["DATE", "TIME", "TIMESTAMP", "DT", "TM"]): field_type = "date" else: field_type = "string" result = compare_field(field_name, cobol_value, java_value, field_type) field_results.append(result) if result.status == "PASS" or result.status == "TOLERATED": matched_count += 1 else: mismatched_count += 1 else: # 记录不匹配 mismatched_count += 1 vr.fields_matched = matched_count vr.fields_mismatched = mismatched_count vr.field_results = field_results vr.status = "PASS" if mismatched_count == 0 else "MISMATCH" vr.exit_code = 0 if mismatched_count == 0 else 1 except Exception as e: vr.status = "ERROR" vr.exit_code = 1 vr.debug["comparison_error"] = str(e) return vr # ── 全Step一括実行 ── def run_all(self, skip_steps: set[int] | None = None, generate_coverage: bool = True) -> VerificationRun: """Step 1 → 6 を順次実行（skip_jvm=True で Step 5/6 をスキップ） 多輪実行：schema.runs が定義されていれば各シナリオを順次実行し、最後に gcov をマージ。 schema.runs が空の場合は単輪（従来動作）。 """ skip = set(skip_steps or []) if self.skip_jvm: skip.update({5, 6}) scenarios = self.schema.runs or [ScenarioDef(id="default")] is_multi = len(scenarios) > 1 or (len(scenarios) == 1 and scenarios[0].id != "default") # Step 1: compile once if 1 not in skip: logger.info(" Step 1 (compile)...") r1 = self.step1_setup_environment() if not r1.success: return VerificationRun( program=self.program_id, runner="gixsql", status="BLOCKED", exit_code=2, step_reached=1, ) # Each scenario: generate inputs + run COBOL cobol_runs = [] for scenario in scenarios: label = f" [{scenario.id}]" if is_multi else "" logger.info(f" Step 2 (generate inputs){label}...") r2 = self.step2_generate_inputs(scenario if is_multi else None) if not r2.success: return VerificationRun( program=self.program_id, runner="gixsql", status="BLOCKED", exit_code=2, step_reached=2, ) # 保存初始 DB 快照（Step2 初始化后、COBOL 运行前）供 Java 使用 initial_db = self._current_db_path or self.db_path scenario_label = self._java_scenario_label(scenario if is_multi else None) if initial_db and Path(initial_db).exists(): import tempfile, shutil as _shutil snap = Path(tempfile.gettempdir()) / f"v3_initial_db_{self.program_id}_{scenario.id}.db" _shutil.copy2(str(initial_db), str(snap)) self._java_initial_db_snapshots[scenario_label] = snap logger.info(f" 初始 DB 快照 -> {snap}") logger.info(f" Step 3 (run COBOL){label}...") r3 = self.step3_run_cobol(scenario if is_multi else None) if not r3.success: return VerificationRun( program=self.program_id, runner="gixsql", status="BLOCKED", exit_code=2, step_reached=3, ) cobol_runs.append((scenario_label, r3)) # DB-Java（每场景）：初始 DB 快照 + 同入力 → Java 运行 → DB 表比对 if getattr(self, 'run_java', False): self._run_db_java_scenario(scenario if is_multi else None) # Step 4: extract intermediate (last scenario wins for DB path) if 4 not in skip: logger.info(" Step 4 (extract)...") self.step4_extract_intermediate() if not self.skip_jvm: steps_remaining = [5, 6] for step_num in steps_remaining: if step_num in skip: continue logger.info(f" Step {step_num}...") if step_num == 5: self.step5_run_java() 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: self.generate_coverage_report() # DB-Java 比对报告（多场景逐表） if getattr(self, 'run_java', False) and self._db_java_results: try: self._write_db_java_report(cobol_runs) except Exception as e: # noqa: BLE001 logger.warning(f" DB-Java 报告生成失败: {e}") vr = VerificationRun( program=self.program_id, runner="gixsql", status="PASS", exit_code=0, step_reached=6 if not self.skip_jvm else 4, ) return vr # ── Internal helpers ── def _scan_assign_to(self) -> dict[str, str]: """Scan COBOL source for SELECT/ASSIGN-TO + OPEN direction. Returns {filename: direction} where direction is 'INPUT' or 'OUTPUT'. Works for both quoted (\"KIN08R01\") and bare (KIN01R01) ASSIGN. """ src_text = self.src_path.read_text(encoding="utf-8-sig") assign_map: dict[str, str] = {} # First pass: collect all SELECT/ASSIGN-TO mappings select_to_file: dict[str, str] = {} for m in re.finditer( r'SELECT\s+(\w+)\s+ASSIGN\s+TO\s+(?:EXTERNAL\s+)?"?([^"\s.]+)', src_text, re.IGNORECASE ): sel_name = m.group(1) fname = m.group(2).strip().rstrip('"') select_to_file[sel_name] = fname assign_map[fname] = "UNKNOWN" # Second pass: determine direction from OPEN statements. # COBOL allows multi-line OPEN where files listed without a direction # keyword inherit the last stated direction: # OPEN INPUT FILEA # FILEB <-- inherits INPUT # OUTPUT FILEC # Strategy: extract OPEN body (up to terminating '.'), collapse # whitespace, then parse direction→file pairs via splitting on # direction keyword boundaries. for m in re.finditer( r'OPEN\s+(.+?)\.', src_text, re.IGNORECASE | re.DOTALL ): full = re.sub(r'\s+', ' ', m.group(1)).strip() # Split on direction keyword boundaries: "INPUT X Y OUTPUT Z" # → ["INPUT X Y", "OUTPUT Z"] tokens = re.split(r'\s+(?=(?:INPUT|OUTPUT|I-O|EXTEND)\s)', full, flags=re.IGNORECASE) for seg in tokens: seg = seg.strip() if not seg: continue seg_m = re.match(r'(INPUT|OUTPUT|I-O|EXTEND)\s+([\w ]+)', seg, re.IGNORECASE) if not seg_m: continue direction = seg_m.group(1).upper() for fword in re.findall(r'\w+', seg_m.group(2)): if fword in select_to_file: fname = select_to_file[fword] if direction in ("INPUT", "I-O"): assign_map[fname] = "INPUT" else: assign_map[fname] = "OUTPUT" return assign_map def _init_database(self, db_path: Path): """Create tables from schema + COBOL EXEC SQL table definitions.""" self._create_tables(db_path) def _create_tables(self, db_path: Path): conn = sqlite3.connect(str(db_path)) for table in self.schema.db_tables: col_defs = [] pk_cols = [] for col in table.columns: col_defs.append(f"[{col.name}] {col.type}") if col.primary_key: pk_cols.append(f"[{col.name}]") if pk_cols: col_defs.append(f"PRIMARY KEY ({', '.join(pk_cols)})") # DROP 旧表确保 schema 与 YAML 定义一致（含 PRIMARY KEY） conn.execute(f"DROP TABLE IF EXISTS [{table.name}]") if table.sql_name and table.sql_name != table.name: conn.execute(f"DROP TABLE IF EXISTS [{table.sql_name}]") ddl = f"CREATE TABLE IF NOT EXISTS [{table.name}] (\n " + \ ",\n ".join(col_defs) + "\n)" conn.execute(ddl) # If sql_name differs, also create the COBOL-visible SQL table name if table.sql_name and table.sql_name != table.name: conn.execute(ddl.replace(f"[{table.name}]", f"[{table.sql_name}]")) conn.commit() conn.close() logger.info(f" DB initialized: {db_path}") def _merge_schema_columns(self, declared_columns: dict) -> dict: """YAML スキーマのカラム型を declared_columns にマージする。 EXEC SQL DECLARE TABLE がないプログラムでも正しい型が使われるようにする。""" import re for t in self.schema.db_tables: name = t.name.upper() if name not in declared_columns: declared_columns[name] = [] existing = {c['name'].upper() for c in declared_columns[name]} for c in t.columns: if c.name.upper() in existing: continue raw = c.type.upper() if raw.startswith('CHAR('): m = re.search(r'\((\d+)\)', raw) col = {'name': c.name, 'db_type': 'CHAR', 'size': int(m.group(1)) if m else 1} elif raw.startswith('VARCHAR('): m = re.search(r'\((\d+)\)', raw) col = {'name': c.name, 'db_type': 'VARCHAR', 'size': int(m.group(1)) if m else 50} elif raw in ('INTEGER',): col = {'name': c.name, 'db_type': 'INTEGER'} elif raw in ('SMALLINT',): col = {'name': c.name, 'db_type': 'SMALLINT'} elif raw.startswith('DECIMAL(') or raw.startswith('NUMERIC('): m = re.search(r'\((\d+)\s*,?\s*(\d+)?\)', raw) col = {'name': c.name, 'db_type': 'DECIMAL', 'precision': int(m.group(1)) if m else 6, 'scale': int(m.group(2)) if m and m.group(2) else 0} elif raw in ('DATE', 'TIMESTAMP'): col = {'name': c.name, 'db_type': 'DATE'} else: col = {'name': c.name, 'db_type': 'CHAR', 'size': 20} declared_columns[name].append(col) return declared_columns def _insert_pk_map(self) -> dict[str, list[str]]: """Map SQL table name → primary-key column names from the YAML schema. Used to generate PK-collision pre-seed rows for INSERT statements so the duplicate-key error path (SQLCODE = -803) is reachable at runtime. """ pk_map = {} for t in self.schema.db_tables: cols = [c.name for c in t.columns if c.primary_key] if cols: for name in {t.name, t.name.replace('_', '-'), t.sql_name}: if name: pk_map[name] = cols return pk_map 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 cbd = [str(d) for d in self.copybook_dirs] src_resolved = resolve_copybooks(src_text, ".", extra_search_paths=cbd) src_resolved = resolve_sql_includes(src_resolved, ".") preprocessed = preprocess(src_resolved) 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) branch_tree, assignments = build_branch_tree_fallback(proc_div, fields_dict) # Merge SQL assignments from original source sql_assigns = extract_sql_assignments(src_text) for tgt, asgn_list in sql_assigns.items(): for asgn in asgn_list: assignments.setdefault(tgt, []).append(asgn) branch_paths = mcdc_enum_paths(branch_tree, fields_dict) data_div2, declared_columns = strip_exec_sql_from_data_div(data_div) sql_meta = collect_sql_meta(assignments, declared_columns) if not sql_meta: logger.info(" No SQL metadata found, skipping DB population") return declared_columns = self._merge_schema_columns(declared_columns) db_input = build_db_input( branch_paths, fields_dict, assignments, sql_meta, declared_columns, records=records, insert_pk=self._insert_pk_map(), ) if not db_input: 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] # -- DAILY_RECORDS date enrichment: replace counter dates with valid YYYYMMDD -- if 'DAILY_RECORDS' in db_input: dr_rows = db_input['DAILY_RECORDS'] for i, row in enumerate(dr_rows): day = (i % 31) + 1 row['TARGET_DATE'] = f'202607{day:02d}' # -- MONTHLY_ABSENCE YEAR_MONTH enrichment: match command-line YEARMONTH -- if 'MONTHLY_ABSENCE' in db_input: ym = '202607' for row in db_input['MONTHLY_ABSENCE']: row['YEAR_MONTH'] = ym # -- INSURANCE-RATES ↔ EMP-MASTER SEARCH/EVALUATE coordination -- # Programs load all rate rows effective for the runtime YEAR-MONTH # (WHERE EFFECTIVE-FROM <= :ym AND EFFECTIVE-TO >= :ym), SEARCH the # internal WRK-RATE-ENTRY table against each employee's BASE-SALARY, # then EVALUATE DEPT-CODE. For the SEARCH to find a match (→ EVALUATE), # some EMP BASE_SALARY must fall inside a loaded rate's # MONTHLY_FROM..TO, and DEPT-CODE must span the EVALUATE ranges. # Gated on the EFFECTIVE window pattern (SHA02MNC-style) so programs # querying rates by other keys (e.g. SHA06TWM GRADE-CODE lookup) are # untouched. Table-name driven, not program-ID hardcoded. if ('INSURANCE-RATES' in db_input and 'EMP-MASTER' in db_input and any('EFFECTIVE-FROM' in str(m.get('where', '')).upper() for m in sql_meta if m.get('table') == 'INSURANCE-RATES')): rate_rows = db_input.get('INSURANCE-RATES', []) emp_rows = db_input.get('EMP-MASTER', []) if rate_rows and emp_rows: # First loaded rate (lowest GRADE_CODE, ORDER BY GRADE_CODE) # gets a MONTHLY band covering the target salaries. Other # employees' salaries stay OUTSIDE the band → SEARCH AT END # (W02 error log) so both SEARCH branches are runtime-covered. band_lo = 40000 band_hi = 40500 rate_rows[0]['MONTHLY_FROM'] = str(band_lo) rate_rows[0]['MONTHLY_TO'] = str(band_hi) # EMP: first rows inside the band, DEPT-CODE spanning the # EVALUATE ranges (1-10 / 11-20 / 21-30 / OTHER). dept_vals = ['1', '11', '21', '99'] for i, row in enumerate(emp_rows[:4]): row['DEPT_CODE'] = dept_vals[i] row['BASE_SALARY'] = str(band_lo + i * 100) # -- 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 # -- DB 属性区间对齐（通用）-- # 补全 DB 种子键（INSURANCE-RATES 的 GRADE / EMP-MASTER 的 EMP-ID）， # 并将部分 EMP-MASTER 属性（BIRTH-DATE / DEPENDENT-COUNT / REGION-CODE） # 对齐到 flat R02 RULE-TBL 的 AGE/DEPENDENTS/REGION 区间，使 # 第 2 段階ルールマッチング（2020RULESCOL）命中経路到達可能。 self._coordinate_db_rule_matching(db_input, records, fields_dict) # DB 种子值数字化：DB SELECT 种子列若对应 COBOL 输出 FD 的 PIC 9 # （数字）字段，但值形如 'G0000001'（字母+数字），剥离字母转纯数字， # 使 MOVE 到 PIC 9 输出合法（W01/W02 EMP-ID/CHG-DATE 正确显示）。 self._coordinate_seed_numeric_types(db_input, fields_dict) conn = sqlite3.connect(str(db_path)) for table_name, rows in db_input.items(): if not rows: logger.info(f" Table {table_name}: 0 initial rows (will be created at runtime)") continue # Normalize DB2 hyphenated identifiers -> underscores (schema uses underscores) db_table = table_name.replace('-', '_') # Debug logger.info(f" Table {table_name}: {len(rows)} rows, cols={list(rows[0].keys()) if rows else []}") # Query DB column types for type-aware value conversion col_types = {} try: pragma_cols = conn.execute( f"PRAGMA table_info([{db_table}])" ).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() remapped_rows = [] for row in rows: new_row = {} for k, v in row.items(): k_norm = k.replace('-', '_') if k_norm.upper() in valid_cols: new_row[k_norm] = v if new_row: remapped_rows.append(new_row) rows = remapped_rows if not rows: 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 [{db_table}] ({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.replace('-', '_')}]") logger.info(f" Table {table_name}: all rows deleted (scenario={scenario.id})") # -- Per-scenario table drop (e.g. OPEN CURSOR failure scenario) -- # Drops the table so a subsequent SQL OPEN/query fails (SQLCODE != 0), # covering the SQL-error branch. Generic: any program may declare # drop_tables to exercise its table-not-found error paths. if scenario and scenario.drop_tables: for table_name in scenario.drop_tables: conn.execute(f"DROP TABLE IF EXISTS [{table_name.replace('-', '_')}]") logger.info(f" Table {table_name}: dropped (scenario={scenario.id})") conn.commit() conn.close() logger.info(f" DB populated: {db_path}") def _coordinate_db_rule_matching(self, db_input, records, data_fields): """DB 属性区间对齐（通用，无程序硬编码）。 适用：DB 从 EMP-MASTER 取 属性（BIRTH-DATE / DEPENDENT-COUNT / REGION-CODE），再与 flat R02 RULE-TBL 的 AGE-FROM/TO、 DEPENDENTS-FROM/TO、REGION-CODE 区间做 M:N 照合するプログラム （SHA06TWM 等）。生成データでは DB 属性と R02 区间が独立合成され 数量级/値域がずれ、照合命中が発生しない。 本関数： 1) 補全 INSURANCE-RATES 种子鍵：R01 の GRADE-CODE と DB GRADE_CODE の差を埋める（DB-ERR → 主経路）。 2) 補全 EMP-MASTER 种子：R01 の EMP-ID と DB EMP_ID の差を埋める。 3) 属性区间对齐：DB EMP-MASTER の一部行の属性を R02 RULE-TBL の 区间内値に設定し（AGE≈70 / DEP≈85 / REGION=G1 等）、照合命中を 発生させる。他行は区间外を維持し no-data/不照合分支を保持。 検出はテーブル名 + R02 区间フィールド名パターン（AGE-FROM / DEPENDENTS-FROM / REGION-CODE）で行う。プログラム名ハードコードなし。 """ if not db_input or not records: return # 1) 从 records 提取 R02 RULE-TBL 区间（AGE/DEPENDENTS/REGION + 调整率） rule_age_from = rule_age_to = None rule_dep_from = rule_dep_to = None rule_region = None for rec in records: v_af = str(rec.get('R02AGE-FROM', '')).strip() v_at = str(rec.get('R02AGE-TO', '')).strip() v_df = str(rec.get('R02DEPENDENTS-FROM', '')).strip() v_dt = str(rec.get('R02DEPENDENTS-TO', '')).strip() v_rg = str(rec.get('R02REGION-CODE', '')).strip() if v_af.isdigit() and v_at.isdigit() and v_df.isdigit() and v_dt.isdigit() and v_rg: rule_age_from, rule_age_to = int(v_af), int(v_at) rule_dep_from, rule_dep_to = int(v_df), int(v_dt) rule_region = v_rg break # R02 区间模式未检测到 → 不做对齐（避免误伤其他程序） if rule_age_from is None or not rule_region: return # 2) 属性区间对齐：将 EMP-MASTER 已有行的属性设为 RULE 区间内值。 # 仅对齐部分行（保留反例 → no-data/不照合分支维持），不补全 DB 键 # （缺失 GRADE/EMP-ID 记录继续走 DB-ERR → SQLCODE≠0 分支覆盖）。 # AGE≈(from+to)/2 → BIRTH-DATE ≈ 運営日付(20260802) - age*365 # DEPENDENTS≈(from+to)/2, REGION = RULE-TBL REGION if 'EMP-MASTER' in db_input: emp_rows = db_input['EMP-MASTER'] mid_age = (rule_age_from + rule_age_to) // 2 mid_dep = (rule_dep_from + rule_dep_to) // 2 birth_date = _calc_birth_date(mid_age) aligned = 0 for row in emp_rows: # 仅对齐部分行（保留反例） if aligned >= 4: break if 'EMP_ID' not in row or 'BIRTH_DATE' not in row: continue row['BIRTH_DATE'] = birth_date row['DEPENDENT_COUNT'] = str(mid_dep) row['REGION_CODE'] = rule_region aligned += 1 if aligned: logger.info( f" DB 属性区间对齐: {aligned} 条 EMP-MASTER 属性→" f"BIRTH={birth_date}(AGE~{mid_age}) DEP={mid_dep} REG={rule_region}" ) def _coordinate_seed_numeric_types(self, db_input, data_fields): """DB 种子值数字化（通用，无程序硬编码）。 适用：DB SELECT 种子列的值形如 'G0000001'（字母+数字，来自 alpha 合 成序列），但对应 COBOL 输出 FD 字段是 PIC 9（数字，如 SHA07REC 的 EMP-ID PIC 9(008)）。运行时 MOVE 字母值到 PIC 9 非法 → 输出为空/0。 本関数：输出 FD（W01/W02 等）中 PIC 9 类型字段的 base 名（EMP-ID、 CHG-DATE、CHG-ID），对 DB 种子表中列名匹配的列，若值含非数字字符 则剥离非数字、左补零对齐 PIC 长度，转纯数字。字符字段（INSURER / PREV / REASON / CHG-TYPE）不触碰。 検出はフィールド名パターン（PIC 9 + 出力 FD）+ 値パターン（[A-Z]\\d+） で行う。プログラム名ハードコードなし。 """ if not db_input or not data_fields: return # 1) 输出 FD 前缀集合（W01/W02 等 OUTPUT FD）中 PIC 9 字段的 base 名 output_pref = set() pic9_bases = {} # base 名（大写，去连字符）→ 数字位数 for f in data_fields: if not isinstance(f, dict) or not f.get('pic') or f.get('is_88'): continue name = f['name'] m = re.match(r'^(W\d{2})(.*)$', name) if not m: continue pref, rest = m.group(1), m.group(2) pic = str(f.get('pic', '')) if re.match(r'^9\((\d+)\)$', pic): base = rest.lstrip('-').upper().replace('-', '_') digits = int(re.match(r'^9\((\d+)\)$', pic).group(1)) output_pref.add(pref) pic9_bases.setdefault(base, digits) if not pic9_bases: return # 2) 对每个 SELECT 种子表，数字化匹配的列 for table, rows in db_input.items(): if not rows: continue for col in list(rows[0].keys()): col_base = col.upper().replace('-', '_') if col_base not in pic9_bases: continue digits = pic9_bases[col_base] fixed = 0 for row in rows: if col not in row: continue v = str(row[col]).strip() if not v or v.isdigit(): continue # 形如 'G0000001' → 剥离非数字 → '0000001' → 左补零到 digits num = ''.join(ch for ch in v if ch.isdigit()) if not num: continue new_val = num.zfill(digits)[:digits] if new_val != v: row[col] = new_val fixed += 1 if fixed: logger.info( f" DB 种子值数字化: {table}.{col} {fixed} 条→纯数字" f"（PIC 9({digits}) 输出对齐）" ) def _deduplicate_r01_pk(self, recs: list[dict]) -> int: """Ensure all R01 records have unique (EMP_ID, DATE) pairs. After all patching, some records may share the same (EMP_ID, DATE), causing PK violation in DAILY_RECORDS INSERT -> ABEND -> 3000STPSOR not reached. Adjusts the day field for colliding records. """ groups = {} for i, rec in enumerate(recs): eid = rec.get('R01EMP-ID', '') dt = rec.get('R01DATE', '') if not eid or not eid.strip() or eid == '00000000': continue if not dt or len(dt) < 8: continue ym = dt[:6] groups.setdefault((eid, ym), []).append((i, dt[6:8])) fixed = 0 for (eid, ym), entries in groups.items(): if len(entries) <= 1: continue used_days = set(d for _, d in entries) if len(used_days) == len(entries): continue for idx, day in entries: rec = recs[idx] if sum(1 for _, d in entries if d == day) == 1: continue for dd in range(1, 32): nd = f"{dd:02d}" if nd not in used_days: used_days.add(nd) rec['R01DATE'] = ym + nd line = rec.get('R01LINE', '') if line: parts = line.split(',') if len(parts) >= 2: parts[1] = nd.ljust(8) rec['R01LINE'] = ','.join(parts) fixed += 1 logger.info(f" Dedup PK: rec[{idx}] (eid={eid} ym={ym}) day {day}->{nd}") break if fixed: logger.info(f" Dedup PK: {fixed} record(s) adjusted") return fixed def _inject_sql_error_rows(self, db_path: Path, records: list[dict] | None = None): """Insert duplicate-PK rows to trigger SQL error handling paths in COBOL. PK 冲突行的 PK 必须与"运行时实际会被 INSERT"的记录一致（如 R01 记录）， 否则程序 INSERT 时不会冲突。优先用测试记录的合成行（跳过会被清空 EMP-ID 的 records[0] 等特殊记录）；表已有数据时逐行注入，而非固定取 rows[0]。 """ conn = sqlite3.connect(str(db_path)) for table in self.schema.db_tables: pk_cols = [c.name for c in table.columns if c.primary_key] if not pk_cols: continue col_names = [c.name for c in table.columns] try: synthetic = self._make_synthetic_error_rows(table, records) if synthetic: rows = synthetic else: # fallback: 表已有行 rows = conn.execute(f"SELECT * FROM [{table.name}] LIMIT 2").fetchall() if not rows: continue quoted = ", ".join(f"[{c}]" for c in col_names) ph = ", ".join("?" for _ in col_names) for row in rows: vals = tuple(str(row[i]) if c in pk_cols else "X" for i, c in enumerate(col_names)) conn.execute(f"INSERT OR IGNORE INTO [{table.name}] ({quoted}) VALUES ({ph})", vals) logger.info(f" SQL error test row injected into {table.name}") except Exception as e: logger.debug(f" SQL error row injection skipped: {e}") conn.commit() conn.close() def _inject_extra_seed_rows(self, db_path: Path, scenario): """seed_extra_rows: 为 SELECT 型程序注入额外行（大结果集覆盖表头重出等分支）。 config: {table_name: count}。从该表已有 seed 行推导月份（date 列前 6 位， 如 DAILY_RECORDS 的 TARGET_DATE=202607xx），用唯一 EMP_ID + 当月日期 注入 count 行。通用实现：按表名注入，无程序硬编码。 """ extra = getattr(scenario, 'seed_extra_rows', None) if not extra: return conn = sqlite3.connect(str(db_path)) try: for table_name, count in extra.items(): table = next((t for t in self.schema.db_tables if t.name == table_name), None) if not table or not count or count <= 0: continue # 从现有 seed 行推导月份（PK 列中形如 YYYYMMDD 的值前 6 位） sample = conn.execute(f"SELECT * FROM [{table_name}] LIMIT 1").fetchall() month = None for row in sample: for i, c in enumerate(table.columns): if c.primary_key: v = str(row[i]) if len(v) >= 6 and v[:4].isdigit() and v[4:6].isdigit(): month = v[:6] break if month: break if not month: logger.warning(f" seed_extra_rows: {table_name} 无月份可推导, 跳过") continue col_names = [c.name for c in table.columns] quoted = ", ".join(f"[{c}]" for c in col_names) ph = ", ".join("?" for _ in col_names) inserted = 0 for i in range(count): emp = f"SEED{i + 1:04d}" vals = [] for c in table.columns: if c.name == 'EMP_ID': vals.append(emp) elif c.name == 'TARGET_DATE': vals.append(month + '01') elif c.name == 'YEAR_MONTH': vals.append(month) else: vals.append('0') try: conn.execute( f"INSERT OR IGNORE INTO [{table_name}] ({quoted}) " f"VALUES ({ph})", vals) inserted += 1 except Exception as e: logger.debug(f" seed_extra_rows inject skipped: {e}") conn.commit() logger.info( f" seed_extra_rows: {table_name} 注入 {inserted} 条（月 {month}）" ) finally: conn.close() def _inject_aggregation_boundaries(self, recs: list[dict]): """聚合边界数据注入（通用，无程序硬编码）。 目标分支（R01 集計型 DB 程序）： - AGG-ANNUAL-H ON SIZE ERROR：同 (EMP, 年月) 的 2+ 条记录设 *ANNUAL-H 为 PIC 最大值 → 累加溢出。 - AGG-COUNT < 100 的 ELSE：注入使不同 (EMP, 年月) 组合 >= 101 → 表满警告。 R01 记录字段按名称模式（R01*EMP-ID / R01*DATE / R01*ANNUAL-H）自动识别， 未命中即 no-op，不影响其他程序。 """ if not recs or len(recs) < 5: return first = recs[0] emp_f = date_f = hours_f = None for k in first: u = k.upper() if u.startswith('R01'): if u.endswith('EMP-ID') and not emp_f: emp_f = k elif 'ANNUAL' in u and ('-H' in u or 'HOURS' in u) and not hours_f: hours_f = k elif u.endswith('DATE') and 'WORK' not in u and 'APPL' not in u and not date_f: date_f = k if not (emp_f and date_f and hours_f): return # dup_eid = TARGET 最后批次（每批 8）的 EMP，保证被 T 卡片命中。 # 只考虑数字型 EMP（9(008) 字段的合法值）；字母型 EMP（如 'U0000031'） # 对数字字段非法，写文件时会被转成 SPACE 而跳过。 all_ids = sorted({str(r.get(emp_f, '')).strip() for r in recs if str(r.get(emp_f, '')).strip().isdigit() and str(r.get(emp_f, '')).strip() != '00000000'}) n = len(all_ids) if n < 2: return dup_eid = all_ids[n - (n % 8 or 8)] # 1) overflow：同 (EMP, 月) 的 2+ 条记录设 *ANNUAL-H 为 PIC 最大值 max_h = '9' * len(str(first.get(hours_f, ''))) src_idx = next((i for i, r in enumerate(recs) if str(r.get(emp_f, '')).strip() == dup_eid), None) if src_idx is not None and max_h: dup_date = str(recs[src_idx].get(date_f, '')) dup_ym = dup_date[:6] recs[src_idx][hours_f] = max_h used_days = {dup_date[6:8]} if len(dup_date) >= 8 else set() changed = 0 for j in range(max(1, len(recs) - 3), len(recs)): if j == src_idx: continue rec = recs[j] rec[emp_f] = dup_eid orig = str(rec.get(date_f, '')) day = (orig[6:8] if orig and orig[6:8] not in used_days else f"{len(used_days) + 1:02d}") used_days.add(day) rec[date_f] = dup_ym + day rec[hours_f] = max_h changed += 1 if changed >= 1: logger.info(f" Agg overflow: {changed + 1} 条 {dup_eid} 同月 max={max_h}") # 2) agg-full：保证 dup_eid 有 >=110 个不同月（其被 T 卡片命中聚合）， # 使 AGG-COUNT 超过 100 → 触发 AGG-COUNT < 100 的 ELSE（表满警告） distinct = set() for r in recs: e = str(r.get(emp_f, '')).strip() d = str(r.get(date_f, '')) if e and e != '00000000' and len(d) >= 6: distinct.add((e, d[:6])) if dup_eid: used_ym = {d[:6] for (e, d) in distinct if e == dup_eid} template = dict(recs[1] if len(recs) > 1 else recs[0]) target = 110 added = 0 ym = 200001 while len(used_ym) < target: ys = f"{ym:06d}" if ys not in used_ym: nr = dict(template) nr[emp_f] = dup_eid nr[date_f] = ys + '15' recs.append(nr) used_ym.add(ys) distinct.add((dup_eid, ys)) added += 1 ym += 1 if ym > 209912: break if added: logger.info( f" Agg table full: 追加 {added} 条 {dup_eid} 不同月（{dup_eid} 月数 {len(used_ym)}）" ) def _seed_matching_monthly_rows(self, db_path: Path, records: list[dict] | None, max_seed: int = 1, r01_dir: Path | None = None): """Pre-populate MONTHLY_ABSENCE with rows matching actual R01 record data. Reads the generated R01 flat file (200-byte fixed records, KIN07REC layout), extracts unique (EMP_ID, YEAR_MONTH) pairs, and inserts a SUBSET of them. This ensures some AGG entries find HV-CNT > 0 (UPDATE, DP#27) and others find HV-CNT = 0 (INSERT, DP#28).""" r01_path = (r01_dir or self.work_dir / "input") / "KIN08R01" if not r01_path.exists(): logger.info(" R01 file not found, skipping MONTHLY_ABSENCE seed") return conn = sqlite3.connect(str(db_path)) monthly_table = None for t in self.schema.db_tables: if t.name == "MONTHLY_ABSENCE": monthly_table = t break if not monthly_table: conn.close() return col_names = [c.name for c in monthly_table.columns] quoted = ", ".join(f"[{c}]" for c in col_names) ph = ", ".join("?" for _ in col_names) seen = set() pairs = [] rows_inserted = 0 # KIN07REC layout (each record is 200 bytes): # EMP-ID PIC 9(008) offset 0, 8 bytes # DATE PIC 9(008) offset 8, 8 bytes # ... remaining fields (not needed) rec_size = 200 with open(str(r01_path), 'rb') as f: data = f.read() num_recs = len(data) // rec_size for i in range(num_recs): off = i * rec_size emp_id = data[off:off+8].decode('ascii', errors='replace').strip() date = data[off+8:off+16].decode('ascii', errors='replace').strip() year_month = date[:6] if len(date) >= 6 else date if not emp_id or not year_month or emp_id == '00000000': continue key = (emp_id, year_month) if key in seen: continue seen.add(key) pairs.append((emp_id, year_month)) # Sort by EMP_ID for deterministic behavior, then seed only `max_seed` pairs pairs.sort(key=lambda x: x[0]) conn = sqlite3.connect(str(db_path)) monthly_table = None for t in self.schema.db_tables: if t.name == "MONTHLY_ABSENCE": monthly_table = t break if not monthly_table: conn.close() return col_names = [c.name for c in monthly_table.columns] quoted = ", ".join(f"[{c}]" for c in col_names) ph = ", ".join("?" for _ in col_names) for seed_idx, (emp_id, year_month) in enumerate(pairs): if seed_idx >= max_seed: break vals = { "EMP_ID": emp_id, "YEAR_MONTH": year_month, "ANNUAL_LEAVE_H": "0", "PERSONAL_LEAVE_H": "0", "OFFICIAL_LEAVE_H": "0", "SICK_LEAVE_H": "0", "UNAPPROVED_ABSENT_H": "0", "UPDATED_AT": "2026-01-01 00:00:00", } row = tuple(vals.get(c, "") for c in col_names) try: conn.execute(f"INSERT OR IGNORE INTO [MONTHLY_ABSENCE] ({quoted}) VALUES ({ph})", row) rows_inserted += 1 except Exception: pass conn.commit() conn.close() if rows_inserted: logger.info(f" MONTHLY_ABSENCE: {rows_inserted}/{len(pairs)} matching rows seeded (DP#27 F + DP#28 F)") elif pairs: logger.info(f" MONTHLY_ABSENCE: 0 seeded — all AGG entries will INSERT (DP#28 F)") def _make_synthetic_error_rows(self, table, records: list[dict] | None) -> list[tuple] | None: """Build synthetic error rows from test record data. 冲突行的 PK 必须与运行时 INSERT 的实际值一致。运行时主机变量由输入记录 赋值（MOVE R01EMP-ID TO HV-EMP-ID 等），故优先取输入记录字段 （R01EMP-ID / R01DATE），YEAR_MONTH 由 R01DATE[:6] 推导，而非取值 尚未赋值的 WS 合成值（HV-* 在运行前是垃圾值，如 'A0000001'）。 """ if not records or len(records) < 2: return None pk_cols = [c.name for c in table.columns if c.primary_key] if not pk_cols: return None # 列名 → 候选记录字段（输入记录字段优先，其次主机变量） hv_map = { 'EMP_ID': ('R01EMP-ID', 'HV-EMP-ID', ''), 'TARGET_DATE': ('R01DATE', 'HV-TARGET-DATE', ''), 'YEAR_MONTH': ('R01DATE', 'HV-YEAR-MONTH', ''), 'TIME_IN': ('R01TIME-IN', 'HV-TIME-IN', ''), 'TIME_OUT': ('R01TIME-OUT', 'HV-TIME-OUT', ''), 'ANNUAL_LEAVE_H': ('R01ANNUAL-H', 'HV-ANNUAL-H', ''), 'PERSONAL_LEAVE_H': ('R01PERSONAL-H', 'HV-PERSONAL-H', ''), 'OFFICIAL_LEAVE_H': ('R01OFFICIAL-H', 'HV-OFFICIAL-H', ''), 'SICK_LEAVE_H': ('R01SICK-H', 'HV-SICK-H', ''), 'UNAPPROVED_ABSENT_H': ('R01ABSENT-H', 'HV-ABSENT-H', ''), } def _first(rec, keys): for k in keys: if k and k in rec: return str(rec[k]).strip() return '' result = [] picked = 0 for rec in records: # 跳过会被清空 EMP-ID / 无效键的特殊记录（records[0] 等）， # 只选运行时确实会被 INSERT 的记录作为冲突 PK。 emp = _first(rec, hv_map.get('EMP_ID', ())) if not emp or emp == '00000000': continue r01date = _first(rec, ('R01DATE', 'HV-TARGET-DATE')) vals = [] for col in table.columns: val = None if col.name == 'YEAR_MONTH': val = r01date[:6] if len(r01date) >= 6 else '' elif col.name in hv_map: val = _first(rec, hv_map[col.name]) or None if val is None: val = ' ' if col.name in pk_cols else '' vals.append(str(val)) result.append(tuple(vals)) picked += 1 if picked >= 2: break return result if result else None