DataOps-platform/tests/data_research/test_governance_metrics.py

from __future__ import annotations

import pytest


class MemoryGovernanceMetricRepository:
    def __init__(self):
        self.detail_calls = []

    def summary_counts(self, access):
        assert access.business_domain_uids == ("domain-a", "domain-b")
        return {
            "asset_completeness": (2, 3),
            "responsibility_coverage": (3, 3),
            "entity_mapping": (1, 3),
            "issue_closure": (1, 2),
            "issue_recurrence": (0, 0),
        }

    def list_details(self, access, *, metric, state, page, page_size):
        self.detail_calls.append(
            (access, metric, state, page, page_size)
        )
        return (
            [
                {
                    "asset_uid": "asset-1",
                    "asset_name": "冷却泵",
                    "missing_fields": ["organization"],
                }
            ],
            1,
        )


def test_summary_exposes_fixed_definitions_counts_and_rates():
    from app.core.data_research.governance_metrics import (
        GovernanceMetricAccess,
        GovernanceMetricsService,
    )

    service = GovernanceMetricsService(MemoryGovernanceMetricRepository())
    result = service.summary(
        GovernanceMetricAccess(
            global_access=False,
            business_domain_uids=("domain-b", "domain-a", "domain-a"),
        )
    )

    assert result["scope"] == {
        "global_access": False,
        "business_domain_uids": ["domain-a", "domain-b"],
    }
    assert [item["code"] for item in result["metrics"]] == [
        "asset_completeness",
        "responsibility_coverage",
        "entity_mapping",
        "issue_closure",
        "issue_recurrence",
    ]
    assert result["metrics"][0] == {
        "code": "asset_completeness",
        "name": "台账完整率",
        "definition": (
            "名称、位置、组织、责任人及授权有效来源映射均完整的在用设备"
            "占可见在用设备的比例"
        ),
        "numerator": 2,
        "denominator": 3,
        "rate": 0.666667,
        "status": "available",
        "states": ["complete", "incomplete"],
    }
    assert result["metrics"][1]["rate"] == 1.0
    assert result["metrics"][2]["rate"] == 0.333333
    assert result["metrics"][3]["rate"] == 0.5
    assert result["metrics"][4]["rate"] is None
    assert result["metrics"][4]["status"] == "no_data"


def test_detail_validates_metric_state_and_pagination():
    from app.core.data_research.errors import GovernanceMetricInvalid
    from app.core.data_research.governance_metrics import (
        GovernanceMetricAccess,
        GovernanceMetricsService,
    )

    repository = MemoryGovernanceMetricRepository()
    service = GovernanceMetricsService(repository)
    access = GovernanceMetricAccess(global_access=True)

    result = service.details(
        access,
        metric="asset_completeness",
        state="incomplete",
        page="2",
        page_size="25",
    )

    assert result == {
        "metric": {
            "code": "asset_completeness",
            "name": "台账完整率",
            "definition": (
                "名称、位置、组织、责任人及授权有效来源映射均完整的在用设备"
                "占可见在用设备的比例"
            ),
        },
        "state": "incomplete",
        "records": [
            {
                "asset_uid": "asset-1",
                "asset_name": "冷却泵",
                "missing_fields": ["organization"],
            }
        ],
        "page": 2,
        "page_size": 25,
        "total": 1,
    }
    assert repository.detail_calls[-1][1:] == (
        "asset_completeness",
        "incomplete",
        2,
        25,
    )

    with pytest.raises(GovernanceMetricInvalid, match="unsupported metric"):
        service.details(access, metric="ranking", state="top")
    with pytest.raises(GovernanceMetricInvalid, match="unsupported state"):
        service.details(
            access,
            metric="asset_completeness",
            state="mapped",
        )
    with pytest.raises(GovernanceMetricInvalid, match="page must"):
        service.details(
            access,
            metric="asset_completeness",
            state="complete",
            page=0,
        )
    with pytest.raises(GovernanceMetricInvalid, match="page_size must"):
        service.details(
            access,
            metric="asset_completeness",
            state="complete",
            page_size=101,
        )


def test_access_contract_is_immutable_and_normalizes_domains():
    from dataclasses import FrozenInstanceError

    from app.core.data_research.governance_metrics import GovernanceMetricAccess

    access = GovernanceMetricAccess(
        global_access=False,
        business_domain_uids=(" domain-b ", "", "domain-a", "domain-a"),
    )

    assert access.business_domain_uids == ("domain-a", "domain-b")
    with pytest.raises(FrozenInstanceError):
        access.global_access = True