hangshuo652
0203ead96b
Config-driven architecture: - coverage_dates: YAML-driven date replacement for LEAVE_RECORDS - row_overrides: per-scenario field overrides (e.g. STATUS='9') - delete_all_rows: per-scenario table emptying for empty-cursor tests gcda accumulation fix (V3 bug #7): - Delete .gcda before each scenario run to prevent GnuCOBOL accumulation - Force-copy scenario gcda (skip mtime check) - Copy scenario DB to CWD data/kin.db for CONNECT TO path resolution Multi-run gcov merge fix: - Always merge multi-run gcov data regardless of generate_coverage flag - Fix v3_root UnboundLocalError in cleanup path Other fixes: - _make_key_unique: skip WHERE-constrained columns to avoid PK conflict - incremental_supplement: support fields_dict for base record generation - check_coverage: use structure coverage data if available - orchestrator.py: filter _-prefixed fields in TestCase; merge Agent2Data cases
100 lines
3.3 KiB
Python
100 lines
3.3 KiB
Python
"""Program schema — per-program DB table definitions + subprogram list."""
|
|
|
|
from __future__ import annotations
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
|
|
@dataclass
|
|
class ColumnDef:
|
|
name: str
|
|
type: str # SQL type: "CHAR(6)", "NUMERIC(4)", "VARCHAR(30)"
|
|
primary_key: bool = False
|
|
nullable: bool = False
|
|
default: Optional[str] = None
|
|
cobol_field: Optional[str] = None # COBOL field name if different
|
|
|
|
|
|
@dataclass
|
|
class TableDef:
|
|
name: str
|
|
columns: list[ColumnDef] = field(default_factory=list)
|
|
create_if_missing: bool = True
|
|
sql_name: Optional[str] = None # COBOL SQL table name if different from YAML name
|
|
|
|
|
|
@dataclass
|
|
class SysinDef:
|
|
"""SYSIN card configuration for a single run scenario.
|
|
Each program defines its own sysin fields; period/modes are KIN08DBU specific.
|
|
"""
|
|
period: str | None = "202607"
|
|
include_invalid_period: bool = False
|
|
modes: list[str] = field(default_factory=lambda: ["NORMAL"])
|
|
|
|
|
|
@dataclass
|
|
class ScenarioDef:
|
|
"""A single run scenario: one set of inputs for one COBOL execution."""
|
|
id: str
|
|
sysin: SysinDef = field(default_factory=SysinDef)
|
|
inject_duplicate_pk: bool = False
|
|
row_overrides: dict[str, dict[str, str]] = field(default_factory=dict)
|
|
delete_all_rows: bool = False
|
|
|
|
|
|
@dataclass
|
|
class ProgramSchema:
|
|
program_id: str
|
|
db_tables: list[TableDef] = field(default_factory=list)
|
|
subprograms: list[str] = field(default_factory=list)
|
|
db_type: str = "SQLite"
|
|
db_name: str = "OVERTIME.DB"
|
|
runs: list[ScenarioDef] = field(default_factory=list)
|
|
coverage_dates: dict[str, list[dict[str, str]]] = field(default_factory=dict)
|
|
|
|
@classmethod
|
|
def from_yaml(cls, path: str | Path) -> ProgramSchema:
|
|
import yaml
|
|
with open(path, encoding="utf-8") as f:
|
|
raw = yaml.safe_load(f)
|
|
tables = []
|
|
for t in raw.get("db_tables", []):
|
|
cols = [ColumnDef(**c) for c in t.get("columns", [])]
|
|
tables.append(TableDef(
|
|
name=t["name"], columns=cols,
|
|
sql_name=t.get("sql_name"),
|
|
))
|
|
runs_raw = raw.get("runs", [])
|
|
runs = []
|
|
for r in runs_raw:
|
|
sysin_raw = r.get("sysin", {})
|
|
runs.append(ScenarioDef(
|
|
id=r["id"],
|
|
sysin=SysinDef(**sysin_raw),
|
|
inject_duplicate_pk=r.get("inject_duplicate_pk", False),
|
|
row_overrides=r.get("row_overrides", {}),
|
|
delete_all_rows=r.get("delete_all_rows", False),
|
|
))
|
|
return cls(
|
|
program_id=raw["program_id"],
|
|
db_tables=tables,
|
|
subprograms=raw.get("subprograms", []),
|
|
db_type=raw.get("db_type", "SQLite"),
|
|
db_name=raw.get("db_name", "OVERTIME.DB"),
|
|
runs=runs,
|
|
coverage_dates=raw.get("coverage_dates", {}),
|
|
)
|
|
|
|
|
|
def load_schema(program_id: str, search_dirs: list[str | Path] | None = None) -> ProgramSchema:
|
|
"""Load per-program YAML schema by program ID."""
|
|
if search_dirs is None:
|
|
search_dirs = [Path(__file__).parent / "programs"]
|
|
for d in search_dirs:
|
|
p = Path(d) / f"{program_id}.yaml"
|
|
if p.exists():
|
|
return ProgramSchema.from_yaml(p)
|
|
raise FileNotFoundError(f"Schema not found for {program_id} in {search_dirs}")
|