|
|
@@ -0,0 +1,642 @@
|
|
|
+#!/usr/bin/env python3
|
|
|
+"""Fail-closed validator for the P3-WP01 enterprise acceptance package."""
|
|
|
+
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import argparse
|
|
|
+import csv
|
|
|
+import hashlib
|
|
|
+import io
|
|
|
+import json
|
|
|
+import math
|
|
|
+import re
|
|
|
+from datetime import datetime
|
|
|
+from pathlib import Path
|
|
|
+from typing import Any
|
|
|
+
|
|
|
+BLOCKED = "ENGINEERING_READY_BLOCKED_EXTERNAL"
|
|
|
+COMPLETE = "ENTERPRISE_ACCEPTANCE_COMPLETE"
|
|
|
+EXPECTED_VERSION = "v0.3.0-phase2"
|
|
|
+EXPECTED_COMMIT = "95023e59bfb9f00be4ed61f7a8307fac56fbcee9"
|
|
|
+PLACEHOLDERS = {"", "TBD", "TBD_EXTERNAL", "UNKNOWN", "N/A", "PLACEHOLDER"}
|
|
|
+PURE_EVIDENCE_PLACEHOLDERS = PLACEHOLDERS | {"BLOCKED_EXTERNAL", "PASS_LOCAL"}
|
|
|
+UNFINISHED_STATE_VALUES = {
|
|
|
+ "TBD_EXTERNAL",
|
|
|
+ "BLOCKED_EXTERNAL",
|
|
|
+ "PASS_LOCAL",
|
|
|
+ "ENGINEERING_READY_BLOCKED_EXTERNAL",
|
|
|
+}
|
|
|
+INCOMPLETE_EVIDENCE_VALUES = PURE_EVIDENCE_PLACEHOLDERS | UNFINISHED_STATE_VALUES
|
|
|
+CASE_IDS = {f"P2-WP13-UAT-{number:03d}" for number in range(21, 25)}
|
|
|
+METRIC_TARGETS = {
|
|
|
+ "core_object_onboarding_rate": 95,
|
|
|
+ "incremental_change_accuracy": 99,
|
|
|
+ "responsibility_coverage": 95,
|
|
|
+ "quality_issue_closure_rate": 85,
|
|
|
+ "incident_evidence_coverage": 100,
|
|
|
+ "data_product_evidence_coverage": 100,
|
|
|
+}
|
|
|
+REHEARSAL_STEPS = {"install", "upgrade", "backup", "candidate_rollback", "rto_rpo"}
|
|
|
+TRAINING_IDS = {"user_training", "operations_training"}
|
|
|
+SIGNOFF_ROLES = {
|
|
|
+ "product_owner",
|
|
|
+ "business_owner",
|
|
|
+ "technical_owner",
|
|
|
+ "security_owner",
|
|
|
+ "operations_owner",
|
|
|
+}
|
|
|
+SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
|
|
|
+P1_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}$")
|
|
|
+CURRENT_STATUS_RE = re.compile(
|
|
|
+ r"^\s*(?:[-*]\s*)?(?:\*\*)?(?:status|state|状态|当前状态)(?:\*\*)?\s*[:：=]\s*"
|
|
|
+ r"[`'\"*]*\s*(?:TBD_EXTERNAL|BLOCKED_EXTERNAL|PASS_LOCAL|ENGINEERING_READY_BLOCKED_EXTERNAL)"
|
|
|
+ r"\s*[`'\"*]*(?:\s|[。；;,.]|$)",
|
|
|
+ re.IGNORECASE | re.MULTILINE,
|
|
|
+)
|
|
|
+HISTORICAL_ENGLISH_TOKENS = {
|
|
|
+ "history",
|
|
|
+ "historical",
|
|
|
+ "previous",
|
|
|
+ "prior",
|
|
|
+ "superseded",
|
|
|
+}
|
|
|
+HISTORICAL_CHINESE_KEYS = {"原状态", "先前状态", "前一状态"}
|
|
|
+
|
|
|
+
|
|
|
+def _missing(value: Any) -> bool:
|
|
|
+ return value is None or (isinstance(value, str) and value.strip().upper() in PLACEHOLDERS)
|
|
|
+
|
|
|
+
|
|
|
+def _reject_json_constant(value: str) -> None:
|
|
|
+ raise ValueError(f"non-finite JSON number is prohibited: {value}")
|
|
|
+
|
|
|
+
|
|
|
+def _is_finite_number(value: Any) -> bool:
|
|
|
+ return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value)
|
|
|
+
|
|
|
+
|
|
|
+def _parse_iso_time(value: Any) -> datetime | None:
|
|
|
+ if not isinstance(value, str) or _missing(value):
|
|
|
+ return None
|
|
|
+ try:
|
|
|
+ parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
|
+ except ValueError:
|
|
|
+ return None
|
|
|
+ return parsed if parsed.tzinfo is not None else None
|
|
|
+
|
|
|
+
|
|
|
+def _normalized_cell(value: str) -> str:
|
|
|
+ return value.strip().strip("`'\"* ").upper()
|
|
|
+
|
|
|
+
|
|
|
+def _is_historical_key(value: Any) -> bool:
|
|
|
+ if not isinstance(value, str):
|
|
|
+ return False
|
|
|
+ key = value.strip().lower()
|
|
|
+ if key in HISTORICAL_ENGLISH_TOKENS:
|
|
|
+ return True
|
|
|
+ if any(
|
|
|
+ key.startswith(f"{token}{separator}")
|
|
|
+ for token in HISTORICAL_ENGLISH_TOKENS
|
|
|
+ for separator in ("_", "-", " ")
|
|
|
+ ):
|
|
|
+ return True
|
|
|
+ if any(key.endswith(f"{separator}history") for separator in ("_", "-", " ")):
|
|
|
+ return True
|
|
|
+ if key == "历史" or key.startswith("历史"):
|
|
|
+ return True
|
|
|
+ return key in HISTORICAL_CHINESE_KEYS or any(
|
|
|
+ key.startswith(f"{token}{separator}")
|
|
|
+ for token in HISTORICAL_CHINESE_KEYS
|
|
|
+ for separator in ("_", "-", " ")
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _json_contains_incomplete_value(value: Any, *, historical_context: bool = False) -> bool:
|
|
|
+ if isinstance(value, dict):
|
|
|
+ for key, child in value.items():
|
|
|
+ child_is_historical = historical_context or _is_historical_key(key)
|
|
|
+ if _json_contains_incomplete_value(child, historical_context=child_is_historical):
|
|
|
+ return True
|
|
|
+ return False
|
|
|
+ if isinstance(value, list):
|
|
|
+ return any(
|
|
|
+ _json_contains_incomplete_value(child, historical_context=historical_context)
|
|
|
+ for child in value
|
|
|
+ )
|
|
|
+ return (
|
|
|
+ not historical_context
|
|
|
+ and isinstance(value, str)
|
|
|
+ and value.strip().upper() in INCOMPLETE_EVIDENCE_VALUES
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _is_separator_row(cells: list[str]) -> bool:
|
|
|
+ return bool(cells) and all(not cell or re.fullmatch(r":?-{3,}:?", cell.strip()) for cell in cells)
|
|
|
+
|
|
|
+
|
|
|
+def _table_contains_current_placeholder(rows: list[list[str]]) -> bool:
|
|
|
+ rows = [[cell.strip() for cell in row] for row in rows if any(cell.strip() for cell in row)]
|
|
|
+ if not rows:
|
|
|
+ return False
|
|
|
+ header = rows[0]
|
|
|
+ historical_columns = {
|
|
|
+ index for index, cell in enumerate(header) if _is_historical_key(_normalized_cell(cell))
|
|
|
+ }
|
|
|
+ for cells in rows:
|
|
|
+ if _is_separator_row(cells):
|
|
|
+ continue
|
|
|
+ if cells and _is_historical_key(_normalized_cell(cells[0])):
|
|
|
+ continue
|
|
|
+ for index, cell in enumerate(cells):
|
|
|
+ if (
|
|
|
+ index not in historical_columns
|
|
|
+ and _normalized_cell(cell) in INCOMPLETE_EVIDENCE_VALUES
|
|
|
+ ):
|
|
|
+ return True
|
|
|
+ return False
|
|
|
+
|
|
|
+
|
|
|
+def _is_placeholder_evidence(evidence_bytes: bytes) -> bool:
|
|
|
+ """Reject structured current-state placeholders without guessing about binary evidence."""
|
|
|
+ try:
|
|
|
+ evidence_text = evidence_bytes.decode("utf-8")
|
|
|
+ except UnicodeDecodeError:
|
|
|
+ return False
|
|
|
+ if evidence_text.strip().upper() in INCOMPLETE_EVIDENCE_VALUES:
|
|
|
+ return True
|
|
|
+ try:
|
|
|
+ parsed_json = json.loads(evidence_text)
|
|
|
+ except json.JSONDecodeError:
|
|
|
+ parsed_json = None
|
|
|
+ if parsed_json is not None and _json_contains_incomplete_value(parsed_json):
|
|
|
+ return True
|
|
|
+ if CURRENT_STATUS_RE.search(evidence_text):
|
|
|
+ return True
|
|
|
+ markdown_rows = [line.split("|")[1:-1] if line.strip().startswith("|") and line.strip().endswith("|") else line.split("|") for line in evidence_text.splitlines() if "|" in line]
|
|
|
+ if _table_contains_current_placeholder(markdown_rows):
|
|
|
+ return True
|
|
|
+ try:
|
|
|
+ csv_rows = list(csv.reader(io.StringIO(evidence_text)))
|
|
|
+ except csv.Error:
|
|
|
+ csv_rows = []
|
|
|
+ return any(len(row) > 1 for row in csv_rows) and _table_contains_current_placeholder(csv_rows)
|
|
|
+
|
|
|
+
|
|
|
+def _records_by_id(records: Any, key: str, expected: set[str], failures: list[str], label: str) -> dict[str, dict]:
|
|
|
+ if not isinstance(records, list):
|
|
|
+ failures.append(f"{label} must be a list")
|
|
|
+ return {}
|
|
|
+ mapped: dict[str, dict] = {}
|
|
|
+ for record in records:
|
|
|
+ if (
|
|
|
+ not isinstance(record, dict)
|
|
|
+ or not isinstance(record.get(key), str)
|
|
|
+ or _missing(record.get(key))
|
|
|
+ ):
|
|
|
+ failures.append(f"{label} contains an invalid record")
|
|
|
+ continue
|
|
|
+ record_id = record[key]
|
|
|
+ if record_id in mapped:
|
|
|
+ failures.append(f"{label} contains duplicate {record_id}")
|
|
|
+ mapped[record_id] = record
|
|
|
+ if set(mapped) != expected:
|
|
|
+ failures.append(f"{label} ids must be exactly {sorted(expected)}")
|
|
|
+ return mapped
|
|
|
+
|
|
|
+
|
|
|
+def _validate_acceptance_bundle(package_path: Path, evidence_root: Path | None = None) -> dict[str, Any]:
|
|
|
+ package_path = Path(package_path).resolve()
|
|
|
+ failures: list[str] = []
|
|
|
+ blockers: list[str] = []
|
|
|
+ try:
|
|
|
+ payload = json.loads(
|
|
|
+ package_path.read_text(encoding="utf-8"), parse_constant=_reject_json_constant
|
|
|
+ )
|
|
|
+ except (OSError, json.JSONDecodeError, ValueError) as exc:
|
|
|
+ return {"status": "FAILED", "declared_status": None, "blockers": [], "failures": [f"cannot read package: {exc}"]}
|
|
|
+ if not isinstance(payload, dict):
|
|
|
+ return {"status": "FAILED", "declared_status": None, "blockers": [], "failures": ["package root must be an object"]}
|
|
|
+
|
|
|
+ declared = payload.get("declared_status")
|
|
|
+ if not isinstance(declared, str) or declared not in {BLOCKED, COMPLETE}:
|
|
|
+ failures.append(f"invalid declared_status: {declared!r}")
|
|
|
+ enterprise_claim = declared == COMPLETE
|
|
|
+ root = Path(evidence_root).resolve() if evidence_root else (
|
|
|
+ package_path.parent.parent.resolve() if package_path.parent.name == "acceptance" else package_path.parent.resolve()
|
|
|
+ )
|
|
|
+
|
|
|
+ def pending_or_fail(label: str) -> None:
|
|
|
+ message = f"{label} is missing or placeholder"
|
|
|
+ (failures if enterprise_claim else blockers).append(message)
|
|
|
+
|
|
|
+ def require(value: Any, label: str) -> bool:
|
|
|
+ if _missing(value):
|
|
|
+ pending_or_fail(label)
|
|
|
+ return False
|
|
|
+ return True
|
|
|
+
|
|
|
+ def require_string(value: Any, label: str) -> bool:
|
|
|
+ if value is None or (
|
|
|
+ isinstance(value, str)
|
|
|
+ and value.strip().upper() in INCOMPLETE_EVIDENCE_VALUES
|
|
|
+ ):
|
|
|
+ pending_or_fail(label)
|
|
|
+ return False
|
|
|
+ if not isinstance(value, str) or not value.strip():
|
|
|
+ failures.append(f"{label} must be a non-empty, non-placeholder string")
|
|
|
+ return False
|
|
|
+ return True
|
|
|
+
|
|
|
+ def object_field(container: dict[str, Any], key: str, label: str) -> dict[str, Any]:
|
|
|
+ value = container.get(key)
|
|
|
+ if not isinstance(value, dict):
|
|
|
+ failures.append(f"{label} must be an object")
|
|
|
+ return {}
|
|
|
+ return value
|
|
|
+
|
|
|
+ def require_time(value: Any, label: str) -> datetime | None:
|
|
|
+ if not require_string(value, label):
|
|
|
+ return None
|
|
|
+ parsed = _parse_iso_time(value)
|
|
|
+ if parsed is None:
|
|
|
+ failures.append(f"{label} must be an ISO-8601 timestamp with timezone")
|
|
|
+ return parsed
|
|
|
+
|
|
|
+ def validate_refs(refs: Any, label: str) -> None:
|
|
|
+ if not isinstance(refs, list):
|
|
|
+ failures.append(f"{label}.evidence_refs must be a list")
|
|
|
+ return
|
|
|
+ if not refs:
|
|
|
+ pending_or_fail(f"{label}.evidence_refs")
|
|
|
+ return
|
|
|
+ for index, ref in enumerate(refs):
|
|
|
+ ref_label = f"{label}.evidence_refs[{index}]"
|
|
|
+ if not isinstance(ref, dict):
|
|
|
+ failures.append(f"{ref_label} must be an object")
|
|
|
+ continue
|
|
|
+ raw_path = ref.get("path")
|
|
|
+ if not require_string(raw_path, f"{ref_label}.path"):
|
|
|
+ continue
|
|
|
+ relative = Path(raw_path)
|
|
|
+ if relative.is_absolute() or ".." in relative.parts:
|
|
|
+ failures.append(f"{ref_label} evidence path escapes evidence root")
|
|
|
+ continue
|
|
|
+ resolved = (root / relative).resolve()
|
|
|
+ try:
|
|
|
+ resolved.relative_to(root)
|
|
|
+ except ValueError:
|
|
|
+ failures.append(f"{ref_label} evidence path escapes evidence root")
|
|
|
+ continue
|
|
|
+ digest = ref.get("sha256")
|
|
|
+ level = ref.get("evidence_level")
|
|
|
+ if not resolved.is_file():
|
|
|
+ if enterprise_claim or not _missing(digest):
|
|
|
+ failures.append(f"{ref_label} missing evidence file: {raw_path}")
|
|
|
+ else:
|
|
|
+ blockers.append(f"{ref_label} external evidence file is not yet captured")
|
|
|
+ if require_string(digest, f"{ref_label}.sha256") and not SHA256_RE.fullmatch(digest):
|
|
|
+ failures.append(f"{ref_label}.sha256 must be a lowercase SHA-256 digest")
|
|
|
+ elif isinstance(digest, str) and SHA256_RE.fullmatch(digest) and resolved.is_file():
|
|
|
+ evidence_bytes = resolved.read_bytes()
|
|
|
+ actual = hashlib.sha256(evidence_bytes).hexdigest()
|
|
|
+ if actual != digest:
|
|
|
+ failures.append(f"{ref_label} digest mismatch")
|
|
|
+ if _is_placeholder_evidence(evidence_bytes):
|
|
|
+ failures.append(f"{ref_label} placeholder evidence cannot satisfy enterprise acceptance")
|
|
|
+ if require_string(level, f"{ref_label}.evidence_level") and level != "ENTERPRISE_FORMAL":
|
|
|
+ failures.append(f"{ref_label} evidence level {level!r} is not ENTERPRISE_FORMAL")
|
|
|
+
|
|
|
+ if payload.get("schema_version") != "1.0" or payload.get("work_package") != "P3-WP01":
|
|
|
+ failures.append("schema_version/work_package contract mismatch")
|
|
|
+ source_cases = payload.get("source_cases")
|
|
|
+ if (
|
|
|
+ not isinstance(source_cases, list)
|
|
|
+ or any(not isinstance(item, str) or _missing(item) for item in source_cases)
|
|
|
+ or set(source_cases) != CASE_IDS
|
|
|
+ ):
|
|
|
+ failures.append("source_cases must contain UAT-021 through UAT-024 exactly")
|
|
|
+
|
|
|
+ context = object_field(payload, "execution_context", "execution_context")
|
|
|
+ for field in ("enterprise_name", "environment_id"):
|
|
|
+ require_string(context.get(field), f"execution_context.{field}")
|
|
|
+ if require_string(context.get("environment_type"), "execution_context.environment_type") and context.get("environment_type") != "PREPRODUCTION":
|
|
|
+ failures.append("execution_context.environment_type must be PREPRODUCTION")
|
|
|
+ commit = context.get("executed_commit")
|
|
|
+ if require_string(commit, "execution_context.executed_commit") and commit != EXPECTED_COMMIT:
|
|
|
+ failures.append(f"execution_context.executed_commit must equal {EXPECTED_COMMIT}")
|
|
|
+ context_started = require_time(context.get("started_at"), "execution_context.started_at")
|
|
|
+ context_completed = require_time(context.get("completed_at"), "execution_context.completed_at")
|
|
|
+ if context_started and context_completed and context_started > context_completed:
|
|
|
+ failures.append("execution_context.started_at must not be after completed_at")
|
|
|
+
|
|
|
+ completion_events: list[tuple[str, datetime]] = []
|
|
|
+
|
|
|
+ def bind_time(value: Any, label: str, *, completion_event: bool = False) -> datetime | None:
|
|
|
+ parsed = require_time(value, label)
|
|
|
+ if parsed and context_started and parsed < context_started:
|
|
|
+ failures.append(f"{label} must be within execution_context time range")
|
|
|
+ if parsed and context_completed and parsed > context_completed:
|
|
|
+ failures.append(f"{label} must be within execution_context time range")
|
|
|
+ if parsed and completion_event:
|
|
|
+ completion_events.append((label, parsed))
|
|
|
+ return parsed
|
|
|
+
|
|
|
+ source = object_field(payload, "source", "source")
|
|
|
+ for field in ("source_id", "source_type", "network_zone"):
|
|
|
+ require_string(source.get(field), f"source.{field}")
|
|
|
+ if source.get("readonly") is None:
|
|
|
+ pending_or_fail("source.readonly")
|
|
|
+ elif source.get("readonly") is not True:
|
|
|
+ failures.append("source.readonly must be true")
|
|
|
+ if source.get("credentials_in_package") is not False:
|
|
|
+ failures.append("source.credentials_in_package must be false")
|
|
|
+
|
|
|
+ sample = object_field(payload, "sample_summary", "sample_summary")
|
|
|
+ if require_string(sample.get("classification"), "sample_summary.classification") and sample.get("classification") != "DESENSITIZED_ENTERPRISE_SAMPLE":
|
|
|
+ failures.append("sample_summary.classification must be DESENSITIZED_ENTERPRISE_SAMPLE")
|
|
|
+ for field in ("object_count", "row_count"):
|
|
|
+ value = sample.get(field)
|
|
|
+ if not require(value, f"sample_summary.{field}"):
|
|
|
+ continue
|
|
|
+ if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
|
|
|
+ failures.append(f"sample_summary.{field} must be a positive integer")
|
|
|
+ sample_digest = sample.get("sha256")
|
|
|
+ if require_string(sample_digest, "sample_summary.sha256") and not SHA256_RE.fullmatch(sample_digest):
|
|
|
+ failures.append("sample_summary.sha256 must be a lowercase SHA-256 digest")
|
|
|
+ if sample.get("raw_data_in_evidence") is not False:
|
|
|
+ failures.append("sample_summary.raw_data_in_evidence must be false")
|
|
|
+
|
|
|
+ artifact = object_field(payload, "release_artifact", "release_artifact")
|
|
|
+ version = artifact.get("version")
|
|
|
+ if require_string(version, "release_artifact.version") and version != EXPECTED_VERSION:
|
|
|
+ failures.append(f"release_artifact.version must equal {EXPECTED_VERSION}")
|
|
|
+ require_string(artifact.get("artifact_name"), "release_artifact.artifact_name")
|
|
|
+ artifact_digest = artifact.get("sha256")
|
|
|
+ if require_string(artifact_digest, "release_artifact.sha256") and not SHA256_RE.fullmatch(artifact_digest):
|
|
|
+ failures.append("release_artifact.sha256 must be a lowercase SHA-256 digest")
|
|
|
+ validate_refs(artifact.get("evidence_refs"), "release_artifact")
|
|
|
+
|
|
|
+ cases = _records_by_id(payload.get("uat_cases"), "id", CASE_IDS, failures, "uat_cases")
|
|
|
+ for case_id, case in cases.items():
|
|
|
+ result = case.get("result")
|
|
|
+ if result == "PASS_LOCAL":
|
|
|
+ failures.append(f"{case_id} PASS_LOCAL cannot satisfy enterprise acceptance")
|
|
|
+ result_valid = False
|
|
|
+ else:
|
|
|
+ result_valid = require_string(result, f"{case_id}.result")
|
|
|
+ if result_valid and result != "ENTERPRISE_PASS":
|
|
|
+ if enterprise_claim:
|
|
|
+ failures.append(f"{case_id} must be ENTERPRISE_PASS")
|
|
|
+ else:
|
|
|
+ blockers.append(f"{case_id} awaits enterprise execution")
|
|
|
+ for field in ("environment_id", "executed_by_person_id", "approved_by_person_id"):
|
|
|
+ require_string(case.get(field), f"{case_id}.{field}")
|
|
|
+ if isinstance(case.get("environment_id"), str) and isinstance(context.get("environment_id"), str) and case.get("environment_id") != context.get("environment_id"):
|
|
|
+ failures.append(f"{case_id}.environment_id must match execution_context.environment_id")
|
|
|
+ bind_time(case.get("executed_at"), f"{case_id}.executed_at", completion_event=True)
|
|
|
+ if case.get("executed_by_person_id") and case.get("executed_by_person_id") == case.get("approved_by_person_id"):
|
|
|
+ failures.append(f"{case_id} self-approval is prohibited")
|
|
|
+ validate_refs(case.get("evidence_refs"), case_id)
|
|
|
+
|
|
|
+ metrics = _records_by_id(payload.get("metrics"), "id", set(METRIC_TARGETS), failures, "metrics")
|
|
|
+ for metric_id, metric in metrics.items():
|
|
|
+ target = METRIC_TARGETS[metric_id]
|
|
|
+ comparator_valid = require_string(metric.get("comparator"), f"{metric_id}.comparator")
|
|
|
+ unit_valid = require_string(metric.get("unit"), f"{metric_id}.unit")
|
|
|
+ target_value = metric.get("target")
|
|
|
+ target_ready = require(target_value, f"{metric_id}.target")
|
|
|
+ target_valid = target_ready and _is_finite_number(target_value)
|
|
|
+ if target_ready and not target_valid:
|
|
|
+ failures.append(f"{metric_id}.target must be a finite number")
|
|
|
+ if not (comparator_valid and unit_valid and target_valid) or metric.get("comparator") != "GTE" or target_value != target or metric.get("unit") != "percent":
|
|
|
+ failures.append(f"{metric_id} threshold contract must be GTE {target} percent")
|
|
|
+ metric_status_valid = require_string(metric.get("status"), f"{metric_id}.status")
|
|
|
+ if metric_status_valid and metric.get("status") != "ENTERPRISE_PASS":
|
|
|
+ (failures if enterprise_claim else blockers).append(f"{metric_id} awaits enterprise measurement")
|
|
|
+ numerator, denominator, value = metric.get("numerator"), metric.get("denominator"), metric.get("value")
|
|
|
+ values_ready = True
|
|
|
+ for item, field in ((numerator, "numerator"), (denominator, "denominator"), (value, "value")):
|
|
|
+ if not require(item, f"{metric_id}.{field}"):
|
|
|
+ values_ready = False
|
|
|
+ elif not _is_finite_number(item):
|
|
|
+ failures.append(f"{metric_id}.{field} must be a finite number")
|
|
|
+ values_ready = False
|
|
|
+ if values_ready:
|
|
|
+ if numerator < 0 or denominator <= 0 or value < 0:
|
|
|
+ failures.append(f"{metric_id} metric values are invalid")
|
|
|
+ else:
|
|
|
+ calculated = numerator / denominator * 100
|
|
|
+ if abs(calculated - value) > 1e-6:
|
|
|
+ failures.append(f"{metric_id} value does not match numerator/denominator")
|
|
|
+ if value < target:
|
|
|
+ failures.append(f"{metric_id} threshold {target}% is not met")
|
|
|
+ window_start = bind_time(metric.get("window_start"), f"{metric_id}.window_start")
|
|
|
+ window_end = bind_time(metric.get("window_end"), f"{metric_id}.window_end", completion_event=True)
|
|
|
+ if window_start and window_end and window_start > window_end:
|
|
|
+ failures.append(f"{metric_id}.window_start must not be after window_end")
|
|
|
+ require_string(metric.get("approved_by_person_id"), f"{metric_id}.approved_by_person_id")
|
|
|
+ validate_refs(metric.get("evidence_refs"), metric_id)
|
|
|
+
|
|
|
+ rehearsal = object_field(payload, "preproduction_rehearsal", "preproduction_rehearsal")
|
|
|
+ rehearsal_status_valid = require_string(rehearsal.get("status"), "preproduction_rehearsal.status")
|
|
|
+ if rehearsal_status_valid and rehearsal.get("status") != "ENTERPRISE_PASS":
|
|
|
+ (failures if enterprise_claim else blockers).append("preproduction rehearsal is incomplete")
|
|
|
+ for field in ("environment_id", "executed_by_person_id", "approved_by_person_id"):
|
|
|
+ require_string(rehearsal.get(field), f"preproduction_rehearsal.{field}")
|
|
|
+ if isinstance(rehearsal.get("environment_id"), str) and isinstance(context.get("environment_id"), str) and rehearsal.get("environment_id") != context.get("environment_id"):
|
|
|
+ failures.append("preproduction_rehearsal.environment_id must match execution_context.environment_id")
|
|
|
+ bind_time(rehearsal.get("executed_at"), "preproduction_rehearsal.executed_at", completion_event=True)
|
|
|
+ if rehearsal.get("executed_by_person_id") and rehearsal.get("executed_by_person_id") == rehearsal.get("approved_by_person_id"):
|
|
|
+ failures.append("preproduction rehearsal self-approval is prohibited")
|
|
|
+ steps = _records_by_id(rehearsal.get("steps"), "id", REHEARSAL_STEPS, failures, "preproduction rehearsal steps")
|
|
|
+ for step_id, step in steps.items():
|
|
|
+ step_status_valid = require_string(step.get("status"), f"preproduction rehearsal step {step_id}.status")
|
|
|
+ if step_status_valid and step.get("status") != "ENTERPRISE_PASS":
|
|
|
+ (failures if enterprise_claim else blockers).append(f"preproduction rehearsal step {step_id} is incomplete")
|
|
|
+ for observed, target_field in (("rto_minutes", "rto_target_minutes"), ("rpo_minutes", "rpo_target_minutes")):
|
|
|
+ target_value = rehearsal.get(target_field)
|
|
|
+ target_valid = require(target_value, f"preproduction_rehearsal.{target_field}")
|
|
|
+ if target_valid and (not _is_finite_number(target_value) or target_value <= 0):
|
|
|
+ failures.append(f"preproduction_rehearsal.{target_field} must be a positive finite number")
|
|
|
+ target_valid = False
|
|
|
+ value = rehearsal.get(observed)
|
|
|
+ value_valid = require(value, f"preproduction_rehearsal.{observed}")
|
|
|
+ if value_valid and (not _is_finite_number(value) or value < 0):
|
|
|
+ failures.append(f"preproduction_rehearsal.{observed} must be a non-negative finite number")
|
|
|
+ value_valid = False
|
|
|
+ if target_valid and value_valid and value > target_value:
|
|
|
+ failures.append(f"preproduction rehearsal {observed} exceeds its target")
|
|
|
+ validate_refs(rehearsal.get("evidence_refs"), "preproduction_rehearsal")
|
|
|
+
|
|
|
+ trainings = _records_by_id(payload.get("training_records"), "id", TRAINING_IDS, failures, "training_records")
|
|
|
+ for training_id, record in trainings.items():
|
|
|
+ training_status_valid = require_string(record.get("status"), f"training {training_id}.status")
|
|
|
+ if training_status_valid and record.get("status") != "ENTERPRISE_PASS":
|
|
|
+ (failures if enterprise_claim else blockers).append(f"training {training_id} is incomplete")
|
|
|
+ participants = record.get("participants")
|
|
|
+ if not isinstance(participants, list):
|
|
|
+ failures.append(f"training {training_id}.participants must be a list")
|
|
|
+ elif not participants:
|
|
|
+ pending_or_fail(f"training {training_id}.participants")
|
|
|
+ else:
|
|
|
+ for index, participant in enumerate(participants):
|
|
|
+ require_string(participant, f"training {training_id}.participants[{index}]")
|
|
|
+ require_string(record.get("trainer_person_id"), f"training {training_id}.trainer_person_id")
|
|
|
+ bind_time(record.get("completed_at"), f"training {training_id}.completed_at", completion_event=True)
|
|
|
+ if record.get("exercise_passed") is not True:
|
|
|
+ (failures if enterprise_claim else blockers).append(f"training {training_id} exercise is incomplete")
|
|
|
+ if record.get("retraining_required") is None:
|
|
|
+ pending_or_fail(f"training {training_id}.retraining_required")
|
|
|
+ elif not isinstance(record.get("retraining_required"), bool):
|
|
|
+ failures.append(f"training {training_id}.retraining_required must be boolean")
|
|
|
+ if record.get("retraining_completed") is None:
|
|
|
+ pending_or_fail(f"training {training_id}.retraining_completed")
|
|
|
+ elif not isinstance(record.get("retraining_completed"), bool):
|
|
|
+ failures.append(f"training {training_id}.retraining_completed must be boolean")
|
|
|
+ if record.get("retraining_required") is True and record.get("retraining_completed") is not True:
|
|
|
+ failures.append(f"training {training_id} required retraining is incomplete")
|
|
|
+ validate_refs(record.get("evidence_refs"), f"training {training_id}")
|
|
|
+
|
|
|
+ defects = object_field(payload, "defect_gate", "defect_gate")
|
|
|
+ defect_status_valid = require_string(defects.get("status"), "defect_gate.status")
|
|
|
+ if defect_status_valid and defects.get("status") != "ENTERPRISE_PASS":
|
|
|
+ (failures if enterprise_claim else blockers).append("defect gate awaits enterprise review")
|
|
|
+ p0_open = defects.get("p0_open")
|
|
|
+ if require(p0_open, "defect_gate.p0_open") and (
|
|
|
+ not isinstance(p0_open, int) or isinstance(p0_open, bool) or p0_open != 0
|
|
|
+ ):
|
|
|
+ failures.append("defect_gate.p0_open must be 0")
|
|
|
+ p1_open = defects.get("p1_open")
|
|
|
+ p1_ready = require(p1_open, "defect_gate.p1_open")
|
|
|
+ if p1_ready and (not isinstance(p1_open, int) or isinstance(p1_open, bool) or p1_open < 0):
|
|
|
+ failures.append("defect_gate.p1_open must be a non-negative integer")
|
|
|
+ p1_ready = False
|
|
|
+ open_p1_ids = defects.get("open_p1_ids")
|
|
|
+
|
|
|
+ def valid_nonnegative_count(value: Any, label: str) -> bool:
|
|
|
+ if not isinstance(value, int) or isinstance(value, bool) or value < 0:
|
|
|
+ failures.append(f"{label} must be a non-negative integer")
|
|
|
+ return False
|
|
|
+ return True
|
|
|
+
|
|
|
+ def valid_p1_ids(value: Any, label: str) -> bool:
|
|
|
+ if not isinstance(value, list):
|
|
|
+ failures.append(f"{label} must be a list")
|
|
|
+ return False
|
|
|
+ if any(not isinstance(item, str) or _missing(item) or not P1_ID_RE.fullmatch(item) for item in value):
|
|
|
+ failures.append(f"{label} contains an invalid or placeholder P1 id")
|
|
|
+ return False
|
|
|
+ if len(value) != len(set(value)):
|
|
|
+ failures.append(f"{label} must contain unique P1 ids")
|
|
|
+ return False
|
|
|
+ return True
|
|
|
+
|
|
|
+ disposition = object_field(defects, "p1_disposition", "defect_gate.p1_disposition")
|
|
|
+ disposition_status = disposition.get("status")
|
|
|
+ if not require_string(disposition_status, "defect_gate.p1_disposition.status"):
|
|
|
+ pass
|
|
|
+ elif p1_ready and p1_open == 0:
|
|
|
+ if open_p1_ids != []:
|
|
|
+ failures.append("defect_gate.open_p1_ids must be empty when p1_open is 0")
|
|
|
+ if disposition_status != "NOT_REQUIRED":
|
|
|
+ failures.append("defect_gate.p1_disposition.status must be NOT_REQUIRED when p1_open is 0")
|
|
|
+ covered_count = disposition.get("covered_p1_count")
|
|
|
+ if not valid_nonnegative_count(covered_count, "defect_gate.p1_disposition.covered_p1_count") or covered_count != 0:
|
|
|
+ failures.append("defect_gate.p1_disposition.covered_p1_count must be 0 when p1_open is 0")
|
|
|
+ if disposition.get("covered_p1_ids") != []:
|
|
|
+ failures.append("defect_gate.p1_disposition.covered_p1_ids must be empty when p1_open is 0")
|
|
|
+ elif p1_ready and p1_open > 0:
|
|
|
+ open_ids_valid = valid_p1_ids(open_p1_ids, "defect_gate.open_p1_ids")
|
|
|
+ if open_ids_valid and len(open_p1_ids) != p1_open:
|
|
|
+ failures.append("defect_gate.open_p1_ids count must equal p1_open")
|
|
|
+ if disposition_status != "WRITTEN_DECISION_APPROVED":
|
|
|
+ failures.append("open P1 defects require WRITTEN_DECISION_APPROVED disposition")
|
|
|
+ covered_count = disposition.get("covered_p1_count")
|
|
|
+ if not valid_nonnegative_count(covered_count, "defect_gate.p1_disposition.covered_p1_count") or covered_count != p1_open:
|
|
|
+ failures.append("P1 written decision must cover every open P1 defect")
|
|
|
+ covered_p1_ids = disposition.get("covered_p1_ids")
|
|
|
+ covered_ids_valid = valid_p1_ids(covered_p1_ids, "defect_gate.p1_disposition.covered_p1_ids")
|
|
|
+ if open_ids_valid and covered_ids_valid and set(covered_p1_ids) != set(open_p1_ids):
|
|
|
+ failures.append("P1 written decision covered_p1_ids must exactly match open_p1_ids")
|
|
|
+ disposition_business = disposition.get("business_person_id")
|
|
|
+ disposition_technical = disposition.get("technical_person_id")
|
|
|
+ require_string(disposition_business, "defect_gate.p1_disposition.business_person_id")
|
|
|
+ require_string(disposition_technical, "defect_gate.p1_disposition.technical_person_id")
|
|
|
+ if disposition_business and disposition_business == disposition_technical:
|
|
|
+ failures.append("P1 written decision business and technical approvers must be distinct")
|
|
|
+ bind_time(disposition.get("decided_at"), "defect_gate.p1_disposition.decided_at", completion_event=True)
|
|
|
+ validate_refs(disposition.get("evidence_refs"), "defect_gate.p1_disposition")
|
|
|
+ business_reviewer = defects.get("reviewed_by_business_person_id")
|
|
|
+ technical_reviewer = defects.get("reviewed_by_technical_person_id")
|
|
|
+ require_string(business_reviewer, "defect_gate.reviewed_by_business_person_id")
|
|
|
+ require_string(technical_reviewer, "defect_gate.reviewed_by_technical_person_id")
|
|
|
+ if business_reviewer and business_reviewer == technical_reviewer:
|
|
|
+ failures.append("defect gate business and technical reviewers must be distinct")
|
|
|
+ bind_time(defects.get("reviewed_at"), "defect_gate.reviewed_at", completion_event=True)
|
|
|
+ validate_refs(defects.get("evidence_refs"), "defect_gate")
|
|
|
+
|
|
|
+ signoffs = _records_by_id(payload.get("signoffs"), "role", SIGNOFF_ROLES, failures, "signoffs")
|
|
|
+ signer_ids: list[str] = []
|
|
|
+ signoff_times: list[tuple[str, datetime]] = []
|
|
|
+ for role, signoff in signoffs.items():
|
|
|
+ signoff_status_valid = require_string(signoff.get("status"), f"{role}.status")
|
|
|
+ if signoff_status_valid and signoff.get("status") != "SIGNED":
|
|
|
+ (failures if enterprise_claim else blockers).append(f"{role} signoff is not signed")
|
|
|
+ if require_string(signoff.get("person_id"), f"{role}.person_id"):
|
|
|
+ signer_ids.append(signoff["person_id"])
|
|
|
+ require_string(signoff.get("person_name"), f"{role}.person_name")
|
|
|
+ signed_at = bind_time(signoff.get("signed_at"), f"{role}.signed_at")
|
|
|
+ if signed_at:
|
|
|
+ signoff_times.append((role, signed_at))
|
|
|
+ validate_refs(signoff.get("evidence_refs"), role)
|
|
|
+ if len(signer_ids) != len(set(signer_ids)):
|
|
|
+ failures.append("five-party proxy signing is prohibited; each role needs a distinct person_id")
|
|
|
+ if completion_events:
|
|
|
+ latest_label, latest_completion = max(completion_events, key=lambda item: item[1])
|
|
|
+ for role, signed_at in signoff_times:
|
|
|
+ if signed_at < latest_completion:
|
|
|
+ failures.append(f"{role}.signed_at must not precede completion event {latest_label}")
|
|
|
+
|
|
|
+ failures = list(dict.fromkeys(failures))
|
|
|
+ blockers = list(dict.fromkeys(blockers))
|
|
|
+ if failures:
|
|
|
+ status = "FAILED"
|
|
|
+ elif blockers:
|
|
|
+ status = "FAILED" if enterprise_claim else "BLOCKED"
|
|
|
+ if enterprise_claim:
|
|
|
+ failures = [f"acceptance claim is incomplete: {blocker}" for blocker in blockers]
|
|
|
+ blockers = []
|
|
|
+ elif enterprise_claim:
|
|
|
+ status = "ACCEPTED"
|
|
|
+ else:
|
|
|
+ status = "FAILED"
|
|
|
+ failures = ["blocked package contains no external blockers; declared status is inconsistent"]
|
|
|
+ return {"status": status, "declared_status": declared, "blockers": blockers, "failures": failures}
|
|
|
+
|
|
|
+
|
|
|
+def validate_acceptance_bundle(package_path: Path, evidence_root: Path | None = None) -> dict[str, Any]:
|
|
|
+ """Validate a bundle and convert expected malformed-input errors into FAILED."""
|
|
|
+ try:
|
|
|
+ return _validate_acceptance_bundle(package_path, evidence_root)
|
|
|
+ except (KeyError, OverflowError, TypeError, ValueError) as exc:
|
|
|
+ return {
|
|
|
+ "status": "FAILED",
|
|
|
+ "declared_status": None,
|
|
|
+ "blockers": [],
|
|
|
+ "failures": [f"malformed acceptance package: {type(exc).__name__}: {exc}"],
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def main() -> int:
|
|
|
+ parser = argparse.ArgumentParser(description=__doc__)
|
|
|
+ parser.add_argument("package", type=Path, help="enterprise acceptance JSON package")
|
|
|
+ parser.add_argument("--evidence-root", type=Path)
|
|
|
+ args = parser.parse_args()
|
|
|
+ result = validate_acceptance_bundle(args.package, args.evidence_root)
|
|
|
+ print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
|
+ return {"ACCEPTED": 0, "FAILED": 2, "BLOCKED": 3}[result["status"]]
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ raise SystemExit(main())
|
马小龙