DataOps-platform/tests/test_product_governance_contract.py

from pathlib import Path

from app.core.system.permissions import (
    DATA_PRODUCTS_MANAGE,
    DATA_PRODUCTS_OPERATE,
    DATA_PRODUCTS_READ,
    permission_for_request,
    permissions_for_roles,
)

ROOT = Path(__file__).resolve().parents[1]


def test_product_governance_migration_is_additive_versioned_and_audited():
    migration = (
        ROOT / "migrations/versions/20260802_440_product_governance.py"
    ).read_text(encoding="utf-8")
    assert 'revision = "20260802_440"' in migration
    assert 'down_revision = "20260802_430"' in migration
    for table in (
        "governed_data_products",
        "data_product_governance_events",
        "data_product_applications",
        "data_product_application_events",
        "data_product_contracts",
        "data_product_contract_versions",
        "data_product_certificates",
        "data_product_feedback",
    ):
        assert f"CREATE TABLE public.{table}" in migration
    assert "REFERENCES public.data_products(id)" in migration
    assert "REFERENCES public.governance_tasks(uid)" in migration
    assert "raise RuntimeError" in migration
    assert "DROP TABLE" not in migration.upper()


def test_product_governance_permissions_separate_read_operate_and_manage():
    assert DATA_PRODUCTS_READ in permissions_for_roles(["viewer"])
    assert DATA_PRODUCTS_OPERATE not in permissions_for_roles(["viewer"])
    assert DATA_PRODUCTS_OPERATE in permissions_for_roles(["editor"])
    assert DATA_PRODUCTS_MANAGE not in permissions_for_roles(["editor"])
    assert DATA_PRODUCTS_MANAGE in permissions_for_roles(["admin"])
    assert permission_for_request(
        "/api/dataservice/governance/products", "GET"
    ) == (DATA_PRODUCTS_READ,)
    assert permission_for_request(
        "/api/dataservice/governance/applications", "POST"
    ) == (DATA_PRODUCTS_OPERATE,)
    assert permission_for_request(
        "/api/dataservice/governance/feedback/x/transition", "POST"
    ) == (DATA_PRODUCTS_OPERATE,)
    assert permission_for_request(
        "/api/dataservice/governance/products/x/contracts", "POST"
    ) == (DATA_PRODUCTS_OPERATE,)


def test_api_and_frontend_cover_the_complete_product_governance_surface():
    api = (
        ROOT / "app/api/data_service/product_governance_routes.py"
    ).read_text(encoding="utf-8")
    client = (ROOT / "frontend/src/api/productGovernance.js").read_text(
        encoding="utf-8"
    )
    view = (
        ROOT / "frontend/src/views/dataService/productGovernance/index.vue"
    ).read_text(encoding="utf-8")
    routes = (ROOT / "frontend/src/router/routes.js").read_text(encoding="utf-8")

    for path in (
        "/governance/products",
        "/governance/applications",
        "/contracts",
        "/compatibility",
        "/certificates",
        "/feedback",
        "/governance/dashboard",
    ):
        assert path in api
    for operation in (
        "registerGovernedProduct",
        "transitionGovernedProduct",
        "createProductApplication",
        "submitProductApplication",
        "reconcileProductApplication",
        "fulfillProductApplication",
        "createProductContract",
        "checkProductContractCompatibility",
        "reviseProductContract",
        "publishProductContract",
        "terminateProductContract",
        "issueProductCertificate",
        "createProductFeedback",
        "transitionProductFeedback",
    ):
        assert operation in client
    for capability in (
        "产品治理与运营",
        "通用申请",
        "数据合同",
        "产品合格证",
        "用户反馈",
        "不会自动下发数据权限",
    ):
        assert capability in view
    assert "/dataService/productGovernance" in routes
    assert "data-products:read" in routes


def test_certificate_reads_canonical_evidence_and_never_grants_access():
    repository = (
        ROOT / "app/core/data_service/product_governance_repository.py"
    ).read_text(encoding="utf-8")
    service = (
        ROOT / "app/core/data_service/product_governance.py"
    ).read_text(encoding="utf-8")
    for evidence_table in (
        "quality_profile_runs",
        "quality_sla_events",
        "active_metadata_lineage",
        "data_rule_versions",
        "workflow_runs",
    ):
        assert evidence_table in repository
    assert "UnifiedWorkCenterService" in repository
    assert "approval_gateway.get_task" in service
    assert '"grants_data_access": False' in service