from __future__ import annotations import copy import hashlib import importlib.util import json from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[1] TEMPLATE = ROOT / "docs/acceptance/P3_WP01_ENTERPRISE_ACCEPTANCE.json" VALIDATOR_PATH = ROOT / "scripts/validate_phase3_wp01_acceptance.py" SPEC = importlib.util.spec_from_file_location("phase3_wp01_validator", VALIDATOR_PATH) assert SPEC and SPEC.loader validator = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(validator) def _load_template() -> dict: return json.loads(TEMPLATE.read_text(encoding="utf-8")) def _bind_evidence(value: object, root: Path, counter: list[int]) -> None: if isinstance(value, dict): for key, child in value.items(): if key == "evidence_refs" and isinstance(child, list): for reference in child: counter[0] += 1 name = f"evidence-{counter[0]:03d}.json" content = json.dumps( {"evidence_id": counter[0], "enterprise_formal": True}, sort_keys=True, ).encode() (root / name).write_bytes(content) reference.update( { "path": name, "sha256": hashlib.sha256(content).hexdigest(), "evidence_level": "ENTERPRISE_FORMAL", } ) else: _bind_evidence(child, root, counter) elif isinstance(value, list): for child in value: _bind_evidence(child, root, counter) def _accepted_bundle(tmp_path: Path) -> Path: payload = copy.deepcopy(_load_template()) payload["declared_status"] = "ENTERPRISE_ACCEPTANCE_COMPLETE" payload["execution_context"].update( { "enterprise_name": "Synthetic Enterprise", "environment_id": "preprod-synthetic-01", "environment_type": "PREPRODUCTION", "executed_commit": "95023e59bfb9f00be4ed61f7a8307fac56fbcee9", "started_at": "2026-08-03T09:00:00+08:00", "completed_at": "2026-08-05T18:00:00+08:00", } ) payload["source"].update( { "source_id": "enterprise-source-synthetic-01", "source_type": "POSTGRESQL", "readonly": True, "credentials_in_package": False, "network_zone": "enterprise-preproduction", } ) payload["sample_summary"].update( { "classification": "DESENSITIZED_ENTERPRISE_SAMPLE", "object_count": 100, "row_count": 1000, "sha256": hashlib.sha256(b"synthetic-enterprise-sample-summary").hexdigest(), "raw_data_in_evidence": False, } ) payload["release_artifact"].update( { "version": "v0.3.0-phase2", "artifact_name": "dataops-phase2-synthetic.tar.gz", "sha256": hashlib.sha256(b"synthetic-release-artifact").hexdigest(), } ) for index, case in enumerate(payload["uat_cases"], start=1): case.update( { "result": "ENTERPRISE_PASS", "environment_id": "preprod-synthetic-01", "executed_at": f"2026-08-03T1{index}:00:00+08:00", "executed_by_person_id": f"executor-{index}", "approved_by_person_id": f"approver-{index}", } ) metric_values = { "core_object_onboarding_rate": (95, 100), "incremental_change_accuracy": (99, 100), "responsibility_coverage": (95, 100), "quality_issue_closure_rate": (85, 100), "incident_evidence_coverage": (10, 10), "data_product_evidence_coverage": (10, 10), } for metric in payload["metrics"]: numerator, denominator = metric_values[metric["id"]] metric.update( { "status": "ENTERPRISE_PASS", "numerator": numerator, "denominator": denominator, "value": numerator / denominator * 100, "window_start": "2026-08-03T09:00:00+08:00", "window_end": "2026-08-03T18:00:00+08:00", "approved_by_person_id": f"metric-approver-{metric['id']}", } ) rehearsal = payload["preproduction_rehearsal"] rehearsal.update( { "status": "ENTERPRISE_PASS", "environment_id": "preprod-synthetic-01", "executed_at": "2026-08-03T14:00:00+08:00", "executed_by_person_id": "preprod-operator-1", "approved_by_person_id": "preprod-approver-1", "rto_minutes": 25, "rpo_minutes": 10, } ) for step in rehearsal["steps"]: step["status"] = "ENTERPRISE_PASS" for index, record in enumerate(payload["training_records"], start=1): record.update( { "status": "ENTERPRISE_PASS", "participants": [f"trainee-{index}-1", f"trainee-{index}-2"], "trainer_person_id": f"trainer-{index}", "completed_at": f"2026-08-04T1{index}:00:00+08:00", "exercise_passed": True, "retraining_required": False, "retraining_completed": False, } ) payload["defect_gate"].update( { "status": "ENTERPRISE_PASS", "p0_open": 0, "p1_open": 0, "reviewed_by_business_person_id": "business-defect-reviewer", "reviewed_by_technical_person_id": "technical-defect-reviewer", "reviewed_at": "2026-08-04T15:00:00+08:00", } ) payload["defect_gate"]["p1_disposition"].update( {"status": "NOT_REQUIRED", "covered_p1_count": 0} ) for index, signoff in enumerate(payload["signoffs"], start=1): signoff.update( { "status": "SIGNED", "person_id": f"signer-{index}", "person_name": f"Synthetic Signer {index}", "signed_at": f"2026-08-05T1{index}:00:00+08:00", } ) _bind_evidence(payload, tmp_path, [0]) package = tmp_path / "enterprise-acceptance.json" package.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") return package def _replace_first_case_evidence(package: Path, content: bytes) -> None: payload = json.loads(package.read_text()) reference = payload["uat_cases"][0]["evidence_refs"][0] (package.parent / reference["path"]).write_bytes(content) reference["sha256"] = hashlib.sha256(content).hexdigest() package.write_text(json.dumps(payload), encoding="utf-8") def _set_nested(payload: object, path: tuple[object, ...], value: object) -> None: current = payload for part in path[:-1]: current = current[part] # type: ignore[index] current[path[-1]] = value # type: ignore[index] def test_checked_in_template_is_engineering_ready_and_blocked_external() -> None: result = validator.validate_acceptance_bundle(TEMPLATE) assert result["status"] == "BLOCKED" assert result["declared_status"] == "ENGINEERING_READY_BLOCKED_EXTERNAL" assert result["blockers"] assert not result["failures"] def test_complete_synthetic_enterprise_bundle_is_accepted(tmp_path: Path) -> None: package = _accepted_bundle(tmp_path) result = validator.validate_acceptance_bundle(package) assert result == { "status": "ACCEPTED", "declared_status": "ENTERPRISE_ACCEPTANCE_COMPLETE", "blockers": [], "failures": [], } def test_low_metric_and_local_pass_cannot_be_accepted(tmp_path: Path) -> None: package = _accepted_bundle(tmp_path) payload = json.loads(package.read_text()) payload["metrics"][0]["numerator"] = 94 payload["metrics"][0]["value"] = 94 payload["uat_cases"][0]["result"] = "PASS_LOCAL" package.write_text(json.dumps(payload), encoding="utf-8") result = validator.validate_acceptance_bundle(package) assert result["status"] == "FAILED" assert any("threshold" in failure for failure in result["failures"]) assert any("PASS_LOCAL" in failure for failure in result["failures"]) @pytest.mark.parametrize("invalid_path", ["../escape.json", "/tmp/absolute.json"]) def test_evidence_path_escape_is_rejected(tmp_path: Path, invalid_path: str) -> None: package = _accepted_bundle(tmp_path) payload = json.loads(package.read_text()) payload["uat_cases"][0]["evidence_refs"][0]["path"] = invalid_path package.write_text(json.dumps(payload), encoding="utf-8") result = validator.validate_acceptance_bundle(package) assert result["status"] == "FAILED" assert any("evidence path" in failure for failure in result["failures"]) def test_missing_or_digest_mismatched_evidence_is_rejected(tmp_path: Path) -> None: package = _accepted_bundle(tmp_path) payload = json.loads(package.read_text()) first = payload["uat_cases"][0]["evidence_refs"][0] (tmp_path / first["path"]).unlink() payload["uat_cases"][1]["evidence_refs"][0]["sha256"] = "0" * 64 package.write_text(json.dumps(payload), encoding="utf-8") result = validator.validate_acceptance_bundle(package) assert result["status"] == "FAILED" assert any("missing evidence" in failure for failure in result["failures"]) assert any("digest mismatch" in failure for failure in result["failures"]) def test_placeholder_file_cannot_pretend_to_be_enterprise_evidence(tmp_path: Path) -> None: package = _accepted_bundle(tmp_path) payload = json.loads(package.read_text()) reference = payload["uat_cases"][0]["evidence_refs"][0] content = b"TBD_EXTERNAL" (tmp_path / reference["path"]).write_bytes(content) reference["sha256"] = hashlib.sha256(content).hexdigest() package.write_text(json.dumps(payload), encoding="utf-8") result = validator.validate_acceptance_bundle(package) assert result["status"] == "FAILED" assert any("placeholder evidence" in failure for failure in result["failures"]) @pytest.mark.parametrize( "content", [ b"status: TBD_EXTERNAL\nThe execution steps are documented below.", json.dumps({"status": "TBD_EXTERNAL", "enterprise_formal": True}).encode(), b"| field | value |\n|---|---|\n| result | TBD_EXTERNAL |", b"field,value\nresult,BLOCKED_EXTERNAL", ], ) def test_structured_unfinished_state_cannot_be_formal_evidence( tmp_path: Path, content: bytes ) -> None: package = _accepted_bundle(tmp_path) payload = json.loads(package.read_text()) reference = payload["uat_cases"][0]["evidence_refs"][0] (tmp_path / reference["path"]).write_bytes(content) reference["sha256"] = hashlib.sha256(content).hexdigest() package.write_text(json.dumps(payload), encoding="utf-8") result = validator.validate_acceptance_bundle(package) assert result["status"] == "FAILED" assert any("placeholder evidence" in failure for failure in result["failures"]) def test_checked_in_execution_guide_cannot_be_disguised_as_formal_evidence( tmp_path: Path, ) -> None: package = _accepted_bundle(tmp_path) payload = json.loads(package.read_text()) reference = payload["uat_cases"][0]["evidence_refs"][0] content = (ROOT / "docs/acceptance/P3_WP01_EXECUTION_GUIDE.md").read_bytes() (tmp_path / reference["path"]).write_bytes(content) reference["sha256"] = hashlib.sha256(content).hexdigest() package.write_text(json.dumps(payload), encoding="utf-8") result = validator.validate_acceptance_bundle(package) assert result["status"] == "FAILED" assert any("placeholder evidence" in failure for failure in result["failures"]) def test_formal_evidence_may_describe_historical_local_or_blocked_results(tmp_path: Path) -> None: package = _accepted_bundle(tmp_path) payload = json.loads(package.read_text()) reference = payload["uat_cases"][0]["evidence_refs"][0] content = b"Historical PASS_LOCAL and BLOCKED_EXTERNAL results were superseded by this formal run." (tmp_path / reference["path"]).write_bytes(content) reference["sha256"] = hashlib.sha256(content).hexdigest() package.write_text(json.dumps(payload), encoding="utf-8") assert validator.validate_acceptance_bundle(package)["status"] == "ACCEPTED" @pytest.mark.parametrize( "content", [ json.dumps( {"previous_status": "BLOCKED_EXTERNAL", "status": "ENTERPRISE_PASS"} ).encode(), json.dumps( {"historical_status": "PASS_LOCAL", "status": "ENTERPRISE_PASS"} ).encode(), json.dumps( {"prior_status": "BLOCKED_EXTERNAL", "status": "ENTERPRISE_PASS"} ).encode(), json.dumps( {"superseded_status": "PASS_LOCAL", "status": "ENTERPRISE_PASS"} ).encode(), json.dumps( {"history": {"status": "BLOCKED_EXTERNAL"}, "status": "ENTERPRISE_PASS"} ).encode(), b"| previous_status | notes |\n|---|---|\n| BLOCKED_EXTERNAL | superseded |", b"previous_status,notes\nPASS_LOCAL,superseded", b"\xff\xfe\x00\x89binary-evidence", ], ) def test_structured_historical_or_binary_evidence_is_allowed( tmp_path: Path, content: bytes ) -> None: package = _accepted_bundle(tmp_path) _replace_first_case_evidence(package, content) assert validator.validate_acceptance_bundle(package)["status"] == "ACCEPTED" @pytest.mark.parametrize( "content", [ json.dumps({"current_status": "BLOCKED_EXTERNAL"}).encode(), json.dumps( {"status": "ENTERPRISE_PASS", "measurement": "TBD_EXTERNAL"} ).encode(), json.dumps({"priority": "TBD_EXTERNAL"}).encode(), json.dumps({"prioritization": "TBD_EXTERNAL"}).encode(), json.dumps({"原因": "TBD_EXTERNAL"}, ensure_ascii=False).encode(), json.dumps({"原值": "TBD_EXTERNAL"}, ensure_ascii=False).encode(), b"| status | notes |\n|---|---|\n| BLOCKED_EXTERNAL | still open |", b"| metric | measured_value |\n|---|---|\n| onboarding | TBD_EXTERNAL |", b"| priority | TBD_EXTERNAL |", b"state,notes\nPASS_LOCAL,still local", b"metric,measured_value\nonboarding,TBD_EXTERNAL", b"priority,TBD_EXTERNAL", ], ) def test_structured_current_state_evidence_is_rejected( tmp_path: Path, content: bytes ) -> None: package = _accepted_bundle(tmp_path) _replace_first_case_evidence(package, content) result = validator.validate_acceptance_bundle(package) assert result["status"] == "FAILED" assert any("placeholder evidence" in failure for failure in result["failures"]) @pytest.mark.parametrize( ("path", "invalid_value"), [ (("source_cases",), {"case": "P2-WP13-UAT-021"}), (("signoffs",), {"role": "product_owner"}), (("preproduction_rehearsal", "rto_target_minutes"), "30"), (("training_records", 0, "participants"), [None]), (("signoffs", 0, "person_name"), 123), (("metrics", 0, "numerator"), float("nan")), (("metrics", 0, "value"), float("inf")), (("preproduction_rehearsal", "rto_minutes"), float("nan")), (("preproduction_rehearsal", "rpo_minutes"), float("inf")), ], ) def test_malformed_types_and_non_finite_numbers_fail_without_raising( tmp_path: Path, path: tuple[object, ...], invalid_value: object ) -> None: package = _accepted_bundle(tmp_path) payload = json.loads(package.read_text()) _set_nested(payload, path, invalid_value) package.write_text(json.dumps(payload), encoding="utf-8") result = validator.validate_acceptance_bundle(package) assert result["status"] == "FAILED" assert result["failures"] def test_self_approval_and_five_party_proxy_signing_are_rejected(tmp_path: Path) -> None: package = _accepted_bundle(tmp_path) payload = json.loads(package.read_text()) payload["uat_cases"][0]["approved_by_person_id"] = payload["uat_cases"][0][ "executed_by_person_id" ] payload["signoffs"][1]["person_id"] = payload["signoffs"][0]["person_id"] package.write_text(json.dumps(payload), encoding="utf-8") result = validator.validate_acceptance_bundle(package) assert result["status"] == "FAILED" assert any("self-approval" in failure for failure in result["failures"]) assert any("proxy signing" in failure for failure in result["failures"]) def test_open_p0_defect_blocks_acceptance(tmp_path: Path) -> None: package = _accepted_bundle(tmp_path) payload = json.loads(package.read_text()) payload["defect_gate"]["p0_open"] = 1 package.write_text(json.dumps(payload), encoding="utf-8") result = validator.validate_acceptance_bundle(package) assert result["status"] == "FAILED" assert any("p0_open" in failure for failure in result["failures"]) def test_open_p1_with_dual_written_decision_is_accepted(tmp_path: Path) -> None: package = _accepted_bundle(tmp_path) payload = json.loads(package.read_text()) payload["defect_gate"]["p1_open"] = 2 payload["defect_gate"]["open_p1_ids"] = ["P1-UAT-021-001", "P1-UAT-023-002"] payload["defect_gate"]["p1_disposition"].update( { "status": "WRITTEN_DECISION_APPROVED", "covered_p1_count": 2, "covered_p1_ids": ["P1-UAT-023-002", "P1-UAT-021-001"], "business_person_id": "p1-business-approver", "technical_person_id": "p1-technical-approver", "decided_at": "2026-08-04T14:30:00+08:00", } ) package.write_text(json.dumps(payload), encoding="utf-8") assert validator.validate_acceptance_bundle(package)["status"] == "ACCEPTED" def test_open_p1_without_written_decision_is_rejected(tmp_path: Path) -> None: package = _accepted_bundle(tmp_path) payload = json.loads(package.read_text()) payload["defect_gate"]["p1_open"] = 1 package.write_text(json.dumps(payload), encoding="utf-8") result = validator.validate_acceptance_bundle(package) assert result["status"] == "FAILED" assert any("WRITTEN_DECISION_APPROVED" in failure for failure in result["failures"]) def test_open_p1_ids_must_be_unique_and_exactly_covered(tmp_path: Path) -> None: package = _accepted_bundle(tmp_path) payload = json.loads(package.read_text()) payload["defect_gate"]["p1_open"] = 2 payload["defect_gate"]["open_p1_ids"] = ["P1-UAT-021-001", "P1-UAT-021-001"] payload["defect_gate"]["p1_disposition"].update( { "status": "WRITTEN_DECISION_APPROVED", "covered_p1_count": 2, "covered_p1_ids": ["P1-UAT-021-001", "TBD_EXTERNAL"], "business_person_id": "p1-business-approver", "technical_person_id": "p1-technical-approver", "decided_at": "2026-08-04T14:30:00+08:00", } ) package.write_text(json.dumps(payload), encoding="utf-8") result = validator.validate_acceptance_bundle(package) assert result["status"] == "FAILED" assert any("unique P1 ids" in failure for failure in result["failures"]) assert any("invalid or placeholder P1 id" in failure for failure in result["failures"]) @pytest.mark.parametrize( ("field_path", "wrong_value", "message"), [ (("release_artifact", "version"), "v0.3.0-phase2-hotfix", "release_artifact.version"), (("execution_context", "executed_commit"), "5267d8bd8b30a58ed5fb6ec855ea579e7ce34d69", "executed_commit"), ], ) def test_release_version_and_tag_commit_are_exact( tmp_path: Path, field_path: tuple[str, str], wrong_value: str, message: str ) -> None: package = _accepted_bundle(tmp_path) payload = json.loads(package.read_text()) payload[field_path[0]][field_path[1]] = wrong_value package.write_text(json.dumps(payload), encoding="utf-8") result = validator.validate_acceptance_bundle(package) assert result["status"] == "FAILED" assert any(message in failure for failure in result["failures"]) def test_reversed_context_or_metric_window_is_rejected(tmp_path: Path) -> None: package = _accepted_bundle(tmp_path) payload = json.loads(package.read_text()) payload["execution_context"]["started_at"] = "2026-08-05T19:00:00+08:00" payload["metrics"][0]["window_start"] = "2026-08-03T19:00:00+08:00" package.write_text(json.dumps(payload), encoding="utf-8") result = validator.validate_acceptance_bundle(package) assert result["status"] == "FAILED" assert any("started_at must not be after completed_at" in failure for failure in result["failures"]) assert any("window_start must not be after window_end" in failure for failure in result["failures"]) def test_case_and_rehearsal_environment_must_match_context(tmp_path: Path) -> None: package = _accepted_bundle(tmp_path) payload = json.loads(package.read_text()) payload["uat_cases"][0]["environment_id"] = "preprod-other" payload["preproduction_rehearsal"]["environment_id"] = "preprod-other" package.write_text(json.dumps(payload), encoding="utf-8") result = validator.validate_acceptance_bundle(package) assert result["status"] == "FAILED" assert sum("must match execution_context.environment_id" in failure for failure in result["failures"]) == 2 def test_signoff_cannot_precede_acceptance_completion_events(tmp_path: Path) -> None: package = _accepted_bundle(tmp_path) payload = json.loads(package.read_text()) payload["signoffs"][0]["signed_at"] = "2026-08-03T10:00:00+08:00" package.write_text(json.dumps(payload), encoding="utf-8") result = validator.validate_acceptance_bundle(package) assert result["status"] == "FAILED" assert any("must not precede completion event" in failure for failure in result["failures"]) def test_incomplete_rehearsal_or_training_is_rejected(tmp_path: Path) -> None: package = _accepted_bundle(tmp_path) payload = json.loads(package.read_text()) payload["preproduction_rehearsal"]["steps"][2]["status"] = "BLOCKED_EXTERNAL" payload["training_records"][0]["exercise_passed"] = False package.write_text(json.dumps(payload), encoding="utf-8") result = validator.validate_acceptance_bundle(package) assert result["status"] == "FAILED" assert any("rehearsal" in failure for failure in result["failures"]) assert any("training" in failure for failure in result["failures"])