DataOps-platform/tests/test_architecture_artifacts.py

import ast
import subprocess
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
ARCH = ROOT / "docs" / "architecture"


def _route_count() -> int:
    count = 0
    for route_file in (ROOT / "app" / "api").glob("*/*.py"):
        tree = ast.parse(route_file.read_text(encoding="utf-8"))
        for node in ast.walk(tree):
            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue
            for decorator in node.decorator_list:
                if not isinstance(decorator, ast.Call):
                    continue
                function = decorator.func
                if (
                    isinstance(function, ast.Attribute)
                    and isinstance(function.value, ast.Name)
                    and function.value.id == "bp"
                    and (
                        function.attr == "route"
                        or (
                            route_file.parent.name == "knowledge_base"
                            and function.attr
                            in {"get", "post", "put", "patch", "delete"}
                        )
                    )
                ):
                    count += 1
    return count


def test_required_architecture_artifacts_exist():
    required = {
        "ARCHITECTURE_OVERVIEW.md",
        "OPENAPI.yaml",
        "FRONTEND_REMOVAL_INVENTORY.md",
        "DATA_MODEL.md",
        "DEPLOYMENT_SOURCE_OF_TRUTH.md",
        "ADR-001-authentication.md",
        "ADR-002-workflow-engine.md",
        "ADR-003-cross-store-consistency.md",
        "ADR-004-ai-first-kestra-orchestration.md",
        "NEXT_ITERATION_ROADMAP.md",
    }
    assert required <= {path.name for path in ARCH.glob("*")}


def test_openapi_route_inventory_matches_flask_source():
    source_count = _route_count()
    contract = (ARCH / "OPENAPI.yaml").read_text(encoding="utf-8")

    assert source_count > 100
    assert "openapi: 3.1.0" in contract
    assert f"x-route-count: {source_count}" in contract
    assert contract.count("operationId:") == source_count


def test_openapi_generator_is_reproducible(tmp_path):
    generated = tmp_path / "OPENAPI.yaml"
    subprocess.run(
        [
            sys.executable,
            str(ROOT / "scripts" / "generate_openapi.py"),
            "--output",
            str(generated),
        ],
        cwd=ROOT,
        check=True,
    )

    assert generated.read_bytes() == (ARCH / "OPENAPI.yaml").read_bytes()


def test_wp09_contract_and_migration_are_documented():
    contract = (ARCH / "OPENAPI.yaml").read_text(encoding="utf-8")
    data_model = (ARCH / "DATA_MODEL.md").read_text(encoding="utf-8")
    migration = (
        ROOT
        / "migrations/versions/20260729_350_device_observability.py"
    )

    for path in (
        "/api/development/v1/device-observability/events",
        "/api/development/v1/device-observability/import",
        "/api/development/v1/device-observability/graph",
        "/api/development/v1/device-observability/root-cause",
    ):
        assert path in contract
    assert migration.exists()
    assert "device_operational_events" in data_model
    assert "device_evidence_relations" in data_model
    assert "证据不足，无法确认根因" in data_model


def test_wp10_device_knowledge_contract_and_boundaries_are_documented():
    contract = (ARCH / "OPENAPI.yaml").read_text(encoding="utf-8")
    data_model = (ARCH / "DATA_MODEL.md").read_text(encoding="utf-8")
    golden_set = (
        ROOT
        / "docs/acceptance/WP10_DEVICE_KNOWLEDGE_GOLDEN_SET.json"
    )

    for path in (
        "/api/knowledge/search",
        "/api/knowledge/ask",
        "/api/knowledge/sources/{source_uid}",
        "/api/knowledge/admin/device-sources",
        "/api/knowledge/admin/device-sources/{source_uid}/scope",
        "/api/knowledge/admin/query-audits",
    ):
        assert path in contract
    assert "x-response-fields: [" in contract
    assert "citations" in contract
    assert "evidence" in contract
    assert "knowledge_query_audits" in data_model
    assert "authorized_sources" in data_model
    assert "LightRAG 仍是影子投影" in data_model
    assert golden_set.exists()


def test_wp11_governance_metric_contract_and_boundaries_are_documented():
    contract = (ARCH / "OPENAPI.yaml").read_text(encoding="utf-8")
    data_model = (ARCH / "DATA_MODEL.md").read_text(encoding="utf-8")

    for path in (
        "/api/development/v1/governance-metrics/summary",
        "/api/development/v1/governance-metrics/details",
    ):
        assert path in contract
    for field in ("scope", "metrics", "metric", "state", "records", "total"):
        assert field in contract
    for term in (
        "台账完整率",
        "责任覆盖率",
        "实体映射率",
        "问题闭环率",
        "问题复发率",
        "rate = null",
        "authorized_sources",
        "不依赖前端过滤",
    ):
        assert term in data_model


def test_wp12_security_audit_contract_and_boundaries_are_documented():
    contract = (ARCH / "OPENAPI.yaml").read_text(encoding="utf-8")
    data_model = (ARCH / "DATA_MODEL.md").read_text(encoding="utf-8")
    validation = ROOT / "docs/validation/WP12_SECURITY_AUDIT_EVIDENCE.md"

    for path in (
        "/api/system/governance-audit/security-checks",
        "/api/system/governance-audit/coverage",
        "/api/system/governance-audit/events",
        "/api/system/governance-audit/seals",
        "/api/system/governance-audit/seals/{seal_uid}/verify",
    ):
        assert path in contract
    for term in (
        "governance_audit_seals",
        "HMAC-SHA256",
        "authentication",
        "ingestion",
        "entity_resolution",
        "publication",
        "remediation",
        "knowledge_query",
        "不等于阻止数据库管理员修改",
        "50,000",
    ):
        assert term in data_model
    assert validation.exists()


def test_wp12_release_copy_matches_authoritative_backend_files():
    for relative in (
        "app/__init__.py",
        "app/api/system/__init__.py",
        "app/api/system/governance_audit.py",
        "app/config/config.py",
        "app/core/data_source/redaction.py",
        "app/core/system/governance_audit.py",
        "app/core/system/governance_audit_repository.py",
        "app/core/system/permissions.py",
    ):
        source = ROOT / relative
        release = ROOT / "deployment" / relative
        assert release.read_bytes() == source.read_bytes(), relative


def test_contract_ci_regenerates_and_checks_the_committed_inventory():
    workflow = (ROOT / ".github" / "workflows" / "contracts.yml").read_text(
        encoding="utf-8"
    )
    assert "scripts/generate_openapi.py" in workflow
    assert "git diff --exit-code" in workflow
    assert "tests/test_architecture_artifacts.py" in workflow


def test_architecture_decisions_capture_product_boundaries():
    combined = "\n".join(
        (ARCH / name).read_text(encoding="utf-8")
        for name in (
            "DATA_MODEL.md",
            "DEPLOYMENT_SOURCE_OF_TRUTH.md",
            "ADR-001-authentication.md",
            "ADR-002-workflow-engine.md",
            "ADR-003-cross-store-consistency.md",
            "ADR-004-ai-first-kestra-orchestration.md",
            "NEXT_ITERATION_ROADMAP.md",
        )
    )

    for required_term in (
        "管理员",
        "编辑者",
        "查看者",
        "DataFlow",
        "n8n Workflow",
        "同一环境只允许一个当前生效版本",
        "Qwen",
        "DeepSeek",
        "每日全量一致性巡检",
        "app/",
        "deployment/app/",
    ):
        assert required_term in combined


def test_removal_inventory_is_bound_to_the_recovery_backup():
    inventory = (ARCH / "FRONTEND_REMOVAL_INVENTORY.md").read_text(
        encoding="utf-8"
    )
    assert "DataOps-platform-backup-before-cleanup-20260716-170321.tar.gz" in inventory
    assert "866c6c4aff127f88497a5ddf39ac8d8da93008cc784ae6f276f80a7cd17d4ef0" in inventory
    assert "348" in inventory


def test_release_sync_and_package_include_versioned_migrations():
    sync = (ROOT / "deployment" / "sync_release.sh").read_text(encoding="utf-8")
    package = (ROOT / "deployment" / "package_release.sh").read_text(
        encoding="utf-8"
    )
    run_script = (ROOT / "scripts" / "run_dataops.sh").read_text(encoding="utf-8")

    assert 'copy_tree "${ROOT_DIR}/migrations"' in sync
    assert 'copy_file "${ROOT_DIR}/alembic.ini"' in sync
    assert 'copy_into_release "${SCRIPT_DIR}/migrations"' in package
    assert 'copy_into_release "${SCRIPT_DIR}/alembic.ini"' in package
    assert 'alembic" -c "${APP_DIR}/alembic.ini" upgrade head' in run_script