DataOps-platform/tests/test_p2_wp12_replication_package_contract.py

from __future__ import annotations

import hashlib
import json
import subprocess
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
PACKAGE_DIR = ROOT / "docs/phase2/p2-wp12-spare-parts-package"
BASE_COMMIT = "8ec8caa"


def _sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def test_replication_package_is_self_contained_and_hash_bound():
    manifest = json.loads((PACKAGE_DIR / "manifest.json").read_text(encoding="utf-8"))

    for key in ("template", "evidence", "snapshot", "delta", "acceptance_report"):
        binding = manifest["files"][key]
        path = PACKAGE_DIR / binding["path"]
        assert path.is_file()
        assert _sha256(path) == binding["sha256"]
    assert (PACKAGE_DIR / "domain-template.json").read_bytes() == (
        ROOT / "docs/phase2/P2_WP01_SPARE_PARTS_DOMAIN_TEMPLATE.json"
    ).read_bytes()


def test_cli_report_is_reproducible_and_contains_no_source_rows(tmp_path: Path):
    outputs = [tmp_path / "first.json", tmp_path / "second.json"]
    for output in outputs:
        completed = subprocess.run(
            [
                str(ROOT / ".venv/bin/python"),
                str(ROOT / "scripts/validate_domain_replication.py"),
                "--package-dir",
                str(PACKAGE_DIR),
                "--output",
                str(output),
            ],
            cwd=ROOT,
            capture_output=True,
            text=True,
        )
        assert completed.returncode == 0, completed.stderr

    assert outputs[0].read_bytes() == outputs[1].read_bytes()
    assert outputs[0].read_bytes() == (PACKAGE_DIR / "acceptance-report.json").read_bytes()
    report = json.loads(outputs[0].read_text(encoding="utf-8"))
    assert report["status"] == "passed"
    assert report["report_sha256"]
    serialized = outputs[0].read_text(encoding="utf-8")
    assert "MAT-0001" not in serialized
    assert "轴承" not in serialized


def test_package_uses_generic_platform_apis_and_documents_external_gates():
    evidence = json.loads((PACKAGE_DIR / "evidence.json").read_text(encoding="utf-8"))
    api_refs = {
        ref
        for receipt in evidence["receipts"]
        for ref in receipt["api_refs"]
    }
    required_prefixes = {
        "/api/meta/domain-templates",
        "/api/meta/active-metadata",
        "/api/development/v1/semantic-assets",
        "/api/rules/quality-operations",
        "/api/system/work-center",
        "/api/datafactory/observability",
        "/api/dataservice/governance/products",
        "/api/knowledge/agents",
    }
    assert required_prefixes <= api_refs
    assert all("/device-" not in ref for ref in api_refs)

    readme = (PACKAGE_DIR / "README.md").read_text(encoding="utf-8")
    for phrase in (
        "企业真实负责人",
        "企业只读数据源",
        "真实用户 UAT",
        "不能替代生产验收",
        "第三业务域",
    ):
        assert phrase in readme


def test_wp12_changes_do_not_modify_device_specific_services():
    completed = subprocess.run(
        ["git", "diff", "--name-only", BASE_COMMIT, "--"],
        cwd=ROOT,
        check=True,
        capture_output=True,
        text=True,
    )
    changed = set(completed.stdout.splitlines())
    status = subprocess.run(
        ["git", "status", "--porcelain"],
        cwd=ROOT,
        check=True,
        capture_output=True,
        text=True,
    )
    changed.update(
        line[3:]
        for line in status.stdout.splitlines()
        if line.startswith("?? ")
    )
    forbidden_prefixes = (
        "app/core/data_research/device_",
        "app/api/data_development/device_",
        "frontend/src/views/dataResearch/device",
    )
    assert not any(path.startswith(forbidden_prefixes) for path in changed)
    assert "app/core/governance/domain_replication.py" in changed


def test_replication_validator_has_no_domain_specific_imports():
    source = (ROOT / "app/core/governance/domain_replication.py").read_text(
        encoding="utf-8"
    )
    cli = (ROOT / "scripts/validate_domain_replication.py").read_text(
        encoding="utf-8"
    )
    assert "app.core.data_research.device" not in source + cli
    assert "spare_parts" not in source
    assert "material_code" not in source