From f2e304a508a9ff3d048de5bf625ead287ed5dd70 Mon Sep 17 00:00:00 2001 From: kjh2064 Date: Sun, 21 Jun 2026 23:40:53 +0900 Subject: [PATCH] fix: clarify platform transition wbs failure notes --- ...est_validate_platform_transition_wbs_v1.py | 125 ++++++++++++++++++ tools/validate_platform_transition_wbs_v1.py | 39 ++++++ 2 files changed, 164 insertions(+) create mode 100644 tests/unit/test_validate_platform_transition_wbs_v1.py diff --git a/tests/unit/test_validate_platform_transition_wbs_v1.py b/tests/unit/test_validate_platform_transition_wbs_v1.py new file mode 100644 index 0000000..af27520 --- /dev/null +++ b/tests/unit/test_validate_platform_transition_wbs_v1.py @@ -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] diff --git a/tools/validate_platform_transition_wbs_v1.py b/tools/validate_platform_transition_wbs_v1.py index 06f1381..3e69704 100644 --- a/tools/validate_platform_transition_wbs_v1.py +++ b/tools/validate_platform_transition_wbs_v1.py @@ -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