ai-ticket-platform/backend/apps/system/tests/test_blueprint_readiness.py

import json

import pytest
from django.core.management import call_command
from django.core.management.base import CommandError
from django.db import DatabaseError

from apps.system import blueprint_readiness
from apps.system.blueprint_readiness import (
    EXTERNAL_GATE_SPECS,
    build_blueprint_readiness_report,
)


def passed_evaluation_report():
    return {"status": "passed", "exit_ready": True, "reasons": [], "metrics": {}}


def complete_external_evidence():
    common = {
        "status": "passed",
        "verified_at": "2026-07-27T09:30:00Z",
        "reviewer": "acceptance-reviewer",
        "artifact_refs": ["evidence/run-001.json"],
    }
    return {
        "schema_version": 1,
        "gates": {
            "stage3_retrieval": {
                **common,
                "environment": "postgresql",
                "embedding_deployment": "embedding-prod",
                "rerank_deployment": "rerank-prod",
                "sample_count": 50,
                "thresholds_met": True,
                "metrics": {"recall_at_5": 0.92, "p95_latency_ms": 180},
                "hnsw": {"m": 16, "ef_construction": 64, "ef_search": 40},
            },
            "stage4_distillation": {
                **common,
                "sample_count": 20,
                "independent_review": True,
                "anonymization_review_passed": True,
                "published": True,
                "retrieved_after_publish": True,
                "services": ["postgresql", "embedding", "rerank", "text_model"],
            },
            "stage4_documents": {
                **common,
                "failure_recovery_verified": True,
                "formats": ["PDF", "TXT", "LOG", "DOCX", "XLSX", "PPTX"],
                "services": ["minio", "clamav", "postgresql", "embedding"],
            },
            "stage6_channel": {
                **common,
                "provider": "enterprise-chat",
                "signature_verified": True,
                "inbound_verified": True,
                "outbound_verified": True,
                "receipt_verified": True,
                "idempotency_verified": True,
            },
            "stage6_identity": {
                **common,
                "oidc_provider": "enterprise-idp",
                "ldap_directory": "enterprise-directory",
                "oidc_login_verified": True,
                "group_claim_verified": True,
                "failure_fallback_verified": True,
                "ldap_directory_sync_verified": True,
            },
            "stage6_production_ops": {
                **common,
                "alert_receiver": "on-call",
                "backup_schedule": "daily",
                "offsite_storage": "encrypted-vault",
                "alerts_verified": True,
                "backup_schedule_installed": True,
                "offsite_encrypted_copy_verified": True,
            },
        },
    }


def recovery_evidence():
    return {
        "status": "passed",
        "backup_id": "20260727T092235Z",
        "duration_seconds": 106.57,
        "database_smoke": "ticket_and_attachment_restored",
        "object_storage_smoke": "marker_restored",
        "django_check": "passed",
    }


def test_complete_blueprint_evidence_passes_all_gates():
    report = build_blueprint_readiness_report(
        stage2_report=passed_evaluation_report(),
        stage3_report=passed_evaluation_report(),
        external_evidence=complete_external_evidence(),
        recovery_evidence=recovery_evidence(),
    )

    assert report["exit_ready"] is True
    assert report["status"] == "passed"
    assert report["summary"] == {
        "passed_gate_count": 9,
        "total_gate_count": 9,
        "remaining_gate_count": 0,
    }
    assert {gate["code"] for gate in report["gates"]} == {
        "stage2_intake",
        "stage3_support",
        *EXTERNAL_GATE_SPECS,
        "stage6_recovery",
    }


def test_external_evidence_rejects_sensitive_keys_and_incomplete_contract():
    evidence = complete_external_evidence()
    evidence["gates"]["stage6_identity"]["client_secret"] = "must-not-be-recorded"
    report = build_blueprint_readiness_report(
        stage2_report=passed_evaluation_report(),
        stage3_report=passed_evaluation_report(),
        external_evidence=evidence,
        recovery_evidence=recovery_evidence(),
    )

    external_gates = [
        gate for gate in report["gates"] if gate["source"] == "external_evidence_manifest"
    ]
    assert all(gate["status"] == "not_ready" for gate in external_gates)
    assert all("不得包含密码" in gate["reasons"][0] for gate in external_gates)


def test_database_schema_failure_becomes_not_ready_gate(monkeypatch):
    def unavailable_report(*, include_synthetic):
        raise DatabaseError("schema unavailable")

    monkeypatch.setattr(
        blueprint_readiness,
        "build_stage_2_acceptance_report",
        unavailable_report,
    )
    report = build_blueprint_readiness_report(
        stage3_report=passed_evaluation_report(),
        external_evidence=complete_external_evidence(),
        recovery_evidence=recovery_evidence(),
    )

    stage2_gate = next(gate for gate in report["gates"] if gate["code"] == "stage2_intake")
    assert stage2_gate["status"] == "not_ready"
    assert "数据库迁移" in stage2_gate["reasons"][0]


@pytest.mark.django_db
def test_blueprint_command_writes_not_ready_report_and_can_fail_ci(tmp_path):
    output = tmp_path / "blueprint-readiness.json"
    missing_recovery = tmp_path / "missing-recovery.json"

    call_command(
        "report_blueprint_readiness",
        output=str(output),
    )
    report = json.loads(output.read_text(encoding="utf-8"))
    assert report["status"] == "not_ready"
    assert report["exit_ready"] is False
    recovery_gate = next(gate for gate in report["gates"] if gate["code"] == "stage6_recovery")
    assert recovery_gate["status"] == "passed"

    with pytest.raises(CommandError, match="尚未满足"):
        call_command(
            "report_blueprint_readiness",
            recovery_evidence=str(missing_recovery),
            fail_on_not_ready=True,
        )