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 function.attr == "route"
                    and isinstance(function.value, ast.Name)
                    and function.value.id == "bp"
                ):
                    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",
        "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_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",
            "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