kjh2064/QuantEngineByItz
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: clarify platform transition wbs failure notes

Browse Source
This commit is contained in:
김재현
2026-06-21 23:40:53 +09:00
parent 3002149fce
commit f2e304a508
2 changed files with 164 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_validate_platform_transition_wbs_v1.py
+125
View File
@@ -0,0 +1,125 @@
from __future__ import annotations
import json
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
import tools.validate_platform_transition_wbs_v1 as validator
def test_validate_platform_transition_wbs_reports_failure_notes(monkeypatch):
spec = {
"phase_5_platform_transition": {
"P1_kis_core_api_collector": {
"success_criteria": {
"expected_success_value": {},
"evidence_artifacts": [],
"verification_commands": [],
}
},
"P2_sqlite_canonical_store": {
"success_criteria": {
"expected_success_value": {},
"evidence_artifacts": [],
"verification_commands": [],
}
},
"P3_ci_scheduler_cutover": {
"success_criteria": {
"expected_success_value": {},
"evidence_artifacts": [],
"verification_commands": [],
}
},
"P4_gas_thin_adapter_minimize": {
"success_criteria": {
"expected_success_value": {},
"evidence_artifacts": [],
"verification_commands": [],
}
},
"P5_postgresql_upgrade_path": {
"success_criteria": {
"expected_success_value": {},
"evidence_artifacts": [],
"verification_commands": [],
}
},
}
}
monkeypatch.setattr(
validator,
"_load_spec",
lambda: spec,
)
monkeypatch.setattr(
validator,
"_read_text",
lambda path: "Phase 5 데이터 플랫폼 전환 WBS 성공값 P1 KIS core collector P2 SQLite canonical store P3 CI scheduler cutover P4 GAS thin adapter minimize P5 PostgreSQL upgrade path",
)
monkeypatch.setattr(
validator,
"_check_p1",
lambda: {
"gate": "FAIL",
"expected_success_value": {},
"evidence": {"summary_path": "Temp/test_kis_data_collection.json", "db_path": "Temp/test_kis_data_collection.db"},
"errors": ["summary_status=None"],
},
)
monkeypatch.setattr(
validator,
"_check_p2",
lambda: {
"gate": "FAIL",
"expected_success_value": {},
"evidence": {"db_path": "Temp/test_kis_data_collection.db"},
"errors": ["sqlite_round_trip_missing"],
},
)
monkeypatch.setattr(
validator,
"_check_p3",
lambda: {
"gate": "PASS",
"expected_success_value": {},
"evidence": {"workflow_path": ".gitea/workflows/kis_data_collection.yml"},
"errors": [],
},
)
monkeypatch.setattr(
validator,
"_check_p4",
lambda: {
"gate": "FAIL",
"expected_success_value": {},
"evidence": {"validation_path": "Temp/gas_thin_adapter_validation_v1.json"},
"errors": ["gate=None", "function_inventory_coverage_pct<100"],
},
)
monkeypatch.setattr(
validator,
"_check_p5",
lambda: {
"gate": "PASS",
"expected_success_value": {},
"evidence": {},
"errors": [],
},
)
rc = validator.main()
payload = json.loads((ROOT / "Temp" / "platform_transition_wbs_v1.json").read_text(encoding="utf-8"))
assert rc == 1
assert payload["gate"] == "FAIL"
assert payload["message"].startswith("Platform transition WBS check failed")
assert len(payload["failure_notes"]) == 3
assert "P1 failed" in payload["failure_notes"][0]
assert "P2 failed" in payload["failure_notes"][1]
assert "P4 failed" in payload["failure_notes"][2]
tools/validate_platform_transition_wbs_v1.py
+39
View File
@@ -86,6 +86,36 @@ def _check_p1() -> dict[str, Any]:
}
def _humanize_check_failure(key: str, result: dict[str, Any]) -> str:
evidence = result.get("evidence") if isinstance(result.get("evidence"), dict) else {}
errors = result.get("errors") if isinstance(result.get("errors"), list) else []
if key == "P1_kis_core_api_collector":
summary = evidence.get("summary_path", "")
db = evidence.get("db_path", "")
return (
"P1 failed: missing or empty KIS collector evidence. "
f"Expected {summary} and {db} to exist with collection_runs>=1 and collection_snapshots>=1. "
"Fix: run the KIS collection job first, or restore the collector artifacts before this validator."
)
if key == "P2_sqlite_canonical_store":
db = evidence.get("db_path", "")
return (
"P2 failed: SQLite round-trip evidence is missing. "
f"Expected {db} to contain collection_runs and collection_snapshots. "
"Fix: regenerate the collector DB or restore Temp/test_kis_data_collection.db before this validator."
)
if key == "P4_gas_thin_adapter_minimize":
validation_path = evidence.get("validation_path", "")
return (
"P4 failed: thin-adapter validation output is missing or incomplete. "
f"Expected {validation_path} with gate=PASS and function_inventory_coverage_pct=100.0. "
"Fix: run tools/validate_gas_thin_adapter_v1.py before the platform-transition gate."
)
if errors:
return f"{key} failed: " + ", ".join(str(item) for item in errors)
return f"{key} failed: evidence gate did not pass."
def _check_p2() -> dict[str, Any]:
from src.quant_engine.data_collection_backend_v1 import CollectionStoreSpec, normalize_store_spec
@@ -272,6 +302,7 @@ def main() -> int:
}
missing_criteria: list[str] = []
failure_notes: list[str] = []
for key, result in checks.items():
spec_row = phase.get(key) or {}
criteria = spec_row.get("success_criteria") or {}
@@ -285,6 +316,7 @@ def main() -> int:
missing_criteria.append(f"{key}.verification_commands")
if result["gate"] != "PASS":
missing_criteria.append(f"{key}.evidence_gate")
failure_notes.append(_humanize_check_failure(key, result))
roadmap_mentions = [
"Phase 5 데이터 플랫폼 전환 WBS 성공값",
@@ -299,14 +331,21 @@ def main() -> int:
payload = {
"formula_id": "PLATFORM_TRANSITION_WBS_V1",
"gate": "PASS" if not missing_criteria and not roadmap_missing else "FAIL",
"message": (
"Platform transition WBS check passed."
if not missing_criteria and not roadmap_missing
else "Platform transition WBS check failed. See failure_notes for the exact missing evidence and recovery step."
),
"spec_path": str(SPEC_PATH),
"roadmap_doc_path": str(ROADMAP_DOC_PATH),
"missing_criteria": missing_criteria,
"failure_notes": failure_notes,
"roadmap_missing": roadmap_missing,
"checks": checks,
}
out = ROOT / "Temp" / "platform_transition_wbs_v1.json"
out.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
print(payload["message"])
print(json.dumps(payload, ensure_ascii=False, indent=2))
return 0 if payload["gate"] == "PASS" else 1