hangshuo652/cobol-java-v3
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: SQL between/hostvar-key alignment, class-condition parsing, gcov merge across scenario runs

Browse Source
This commit is contained in:
hangshuo652
2026-08-09 17:43:00 +08:00
parent f331c8fa2a
commit 273a3f8211
31 changed files with 3789 additions and 272 deletions
Download Patch File Download Diff File Expand all files Collapse all files
orchestrator_db.py
+653 -74
View File
@@ -34,6 +34,33 @@ from runners.gixsql_runner import GixsqlCobolRunner, GixsqlTableData
logger = logging.getLogger(__name__)
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 管线単体実行結果"""
@@ -86,6 +113,7 @@ class GixsqlOrchestrator:
self.java_input_path: Optional[Path] = None
self._current_db_path: Optional[Path] = None # scenario-specific DB path
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
@@ -148,6 +176,20 @@ class GixsqlOrchestrator:
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:
@@ -235,10 +277,27 @@ class GixsqlOrchestrator:
# 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', '')
@@ -253,10 +312,8 @@ class GixsqlOrchestrator:
emp_id = rec.get('HV-EMP-ID', '')
if not emp_id or emp_id == '00000000':
emp_id = f"EMP{str(i).zfill(5)}"
# First record: empty EMP-ID to trigger R01EMP-ID = SPACE path (DP#12)
if i == 0:
rec['R01LINE'] = f"{' '*8},{parts[1]}"
rec['R01EMP-ID'] = ' ' * len(emp_id)
else:
rec['R01LINE'] = f"{emp_id.ljust(8)},{parts[1]}"
rec['R01EMP-ID'] = emp_id
@@ -286,20 +343,25 @@ class GixsqlOrchestrator:
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 different day
# to avoid PK conflict in DAILY_RECORDS INSERT.
# 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', '')
if orig_date and len(orig_date) >= 8:
rec['R01DATE'] = dup_ym + orig_date[6:8]
else:
rec['R01DATE'] = dup_ym + '01'
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:
@@ -307,6 +369,9 @@ class GixsqlOrchestrator:
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.
@@ -367,6 +432,9 @@ class GixsqlOrchestrator:
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
@@ -384,6 +452,7 @@ class GixsqlOrchestrator:
"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],
@@ -406,7 +475,7 @@ class GixsqlOrchestrator:
data_fields = parse_data_division(data_div) if data_div else []
fdict = []
for f in data_fields:
fdict.append({
_entry = {
'name': f.name, 'level': f.level, 'pic': f.pic,
'pic_info': {
'type': f.pic_info.type if f.pic_info else 'unknown',
@@ -419,7 +488,11 @@ class GixsqlOrchestrator:
'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)
@@ -460,12 +533,14 @@ class GixsqlOrchestrator:
# 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)
sql_meta, declared_columns, records=recs,
insert_pk=self._insert_pk_map())
# Write main JSON（シナリオ毎に分離）
json_outdir = output_root / "main" / "json"
@@ -551,14 +626,19 @@ 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))
# 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.
cwd_kin = cwd / "kin"
if cwd_kin.exists():
cwd_kin.unlink()
shutil.copy2(str(db_path), str(cwd_kin))
# .gcda は CWD（= run_dir）に書き出されるので、実行後に gcov/run_{id}/ に移動する
# 各シナリオ実行前に前回の .gcda を削除（GnuCOBOL は累積書込みを行うため）
@@ -569,15 +649,27 @@ class GixsqlOrchestrator:
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 / "logs"
@@ -642,12 +734,7 @@ class GixsqlOrchestrator:
if not dst.exists():
shutil.copy2(str(f), str(dst))
merged_data: dict[int, int] = {}
for sd in run_dirs:
data = run_gcov(f"{self.program_id}_pp", str(sd))
if data:
for line, count in data.items():
merged_data[line] = max(merged_data.get(line, 0), count)
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
@@ -674,18 +761,16 @@ class GixsqlOrchestrator:
# 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
# Also merge subprogram gcov data from each scenario
from cobol_testgen.gcov import run_gcov as _run_gcov
# 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 / "gcov"
self._sub_gcov_data = {}
for sub in self.schema.subprograms:
sub_merged: dict[int, int] = {}
for sd in sorted(gcov_dir.glob("run_*")):
sub_data = _run_gcov(sub, str(sd))
if sub_data:
for line, cnt in sub_data.items():
sub_merged[line] = max(sub_merged.get(line, 0), cnt)
sub_merged = _merge_run_dirs_gcov(gcov_dir, sub)
if sub_merged:
gcov_data.update(sub_merged)
self._sub_gcov_data[sub] = sub_merged
else:
# Single-run: collect .gcno/.gcda and run gcov
gcov_dir = self.runtime_dir / "gcov"
@@ -726,10 +811,12 @@ class GixsqlOrchestrator:
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:
gcov_data.update(sd)
self._sub_gcov_data[sub] = sd
# 4. Static branch tree from step2
st = self.generated_structure
@@ -758,6 +845,7 @@ class GixsqlOrchestrator:
},
'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:
@@ -1011,7 +1099,7 @@ class GixsqlOrchestrator:
# 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+"?([^"\s.]+)',
r'SELECT\s+(\w+)\s+ASSIGN\s+TO\s+(?:EXTERNAL\s+)?"?([^"\s.]+)',
src_text, re.IGNORECASE
):
sel_name = m.group(1)
@@ -1078,6 +1166,58 @@ class GixsqlOrchestrator:
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 初期行を生成し挿入する。"""
@@ -1094,7 +1234,7 @@ class GixsqlOrchestrator:
data_fields = parse_data_division(data_div) if data_div else []
fields_dict = []
for f in data_fields:
fields_dict.append({
_entry = {
'name': f.name, 'level': f.level, 'pic': f.pic,
'pic_info': {
'type': f.pic_info.type if f.pic_info else 'unknown',
@@ -1107,7 +1247,11 @@ class GixsqlOrchestrator:
'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)
@@ -1127,10 +1271,12 @@ class GixsqlOrchestrator:
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")
@@ -1165,6 +1311,50 @@ class GixsqlOrchestrator:
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():
@@ -1173,12 +1363,26 @@ class GixsqlOrchestrator:
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 []}")
@@ -1186,7 +1390,7 @@ class GixsqlOrchestrator:
col_types = {}
try:
pragma_cols = conn.execute(
f"PRAGMA table_info([{table_name}])"
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}
@@ -1197,8 +1401,9 @@ class GixsqlOrchestrator:
for row in rows:
new_row = {}
for k, v in row.items():
if k.upper() in valid_cols:
new_row[k] = v
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
@@ -1225,20 +1430,219 @@ class GixsqlOrchestrator:
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})"
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}]")
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."""
"""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]
@@ -1246,18 +1650,18 @@ class GixsqlOrchestrator:
continue
col_names = [c.name for c in table.columns]
try:
rows = conn.execute(f"SELECT * FROM [{table.name}] LIMIT 2").fetchall()
if len(rows) < 1:
# For empty tables, generate synthetic error rows from test record data
synthetic = self._make_synthetic_error_rows(table, records)
if synthetic:
rows = synthetic
else:
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(rows[0][i]) if c in pk_cols else "X" for i, c in enumerate(col_names))
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:
@@ -1265,6 +1669,163 @@ class GixsqlOrchestrator:
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):
@@ -1353,42 +1914,60 @@ class GixsqlOrchestrator:
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 for an empty table from test record data."""
"""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
# Map COBOL host-variable names to table column names
# KIN08DBU DAILY_RECORDS: EMP_ID=HV-EMP-ID, TARGET_DATE=HV-TARGET-DATE
# KIN08DBU MONTHLY_ABSENCE: EMP_ID=HV-EMP-ID, YEAR_MONTH=HV-YEAR-MONTH
# 列名 → 候选记录字段（输入记录字段优先，其次主机变量）
hv_map = {
'EMP_ID': ('HV-EMP-ID', 'R01EMP-ID', ''),
'TARGET_DATE': ('HV-TARGET-DATE', ''),
'YEAR_MONTH': ('HV-YEAR-MONTH', ''),
'TIME_IN': ('HV-TIME-IN', ''),
'TIME_OUT': ('HV-TIME-OUT', ''),
'ANNUAL_LEAVE_H': ('HV-ANNUAL-H', ''),
'PERSONAL_LEAVE_H': ('HV-PERSONAL-H', ''),
'OFFICIAL_LEAVE_H': ('HV-OFFICIAL-H', ''),
'SICK_LEAVE_H': ('HV-SICK-H', ''),
'UNAPPROVED_ABSENT_H': ('HV-ABSENT-H', ''),
'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 = []
for idx in range(min(2, len(records))):
rec = records[idx]
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 in hv_map:
for key in hv_map[col.name]:
if key and key in rec:
val = rec[key]
break
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) if val is not None else '')
vals.append(str(val))
result.append(tuple(vals))
picked += 1
if picked >= 2:
break
return result if result else None