DataOps-platform/tests/data_research/test_governance_metrics_api.py

from __future__ import annotations


class FakeGovernanceMetricsService:
    def __init__(self):
        self.calls = []

    def summary(self, access):
        self.calls.append(("summary", access))
        return {
            "scope": {
                "global_access": access.global_access,
                "business_domain_uids": list(access.business_domain_uids),
            },
            "metrics": [
                {
                    "code": "asset_completeness",
                    "name": "台账完整率",
                    "definition": "固定口径",
                    "numerator": 2,
                    "denominator": 3,
                    "rate": 0.666667,
                    "status": "available",
                    "states": ["complete", "incomplete"],
                }
            ],
        }

    def details(self, access, **filters):
        self.calls.append(("details", access, filters))
        return {
            "metric": {
                "code": filters["metric"],
                "name": "台账完整率",
                "definition": "固定口径",
            },
            "state": filters["state"],
            "records": [{"asset_uid": "asset-1", "asset_name": "循环泵"}],
            "page": int(filters["page"]),
            "page_size": int(filters["page_size"]),
            "total": 1,
        }


def _client(monkeypatch, *, role="viewer"):
    from app import create_app
    from app.api.data_development import routes
    from app.core.data_research.governance_metrics import GovernanceMetricAccess

    service = FakeGovernanceMetricsService()
    monkeypatch.setattr(
        "app.core.system.permissions.authenticate_request",
        lambda: {
            "id": "00000000-0000-7000-8000-000000000111",
            "roles": [role],
        },
    )
    monkeypatch.setattr(
        routes,
        "get_governance_metrics_service",
        lambda: service,
        raising=False,
    )
    monkeypatch.setattr(
        routes,
        "_governance_metric_access",
        lambda: GovernanceMetricAccess(
            global_access=role == "admin",
            business_domain_uids=(
                () if role == "admin" else ("domain-a",)
            ),
        ),
        raising=False,
    )
    app = create_app()
    app.config.update(TESTING=True)
    return app.test_client(), service


def test_viewer_reads_summary_with_server_derived_scope(monkeypatch):
    http, service = _client(monkeypatch)

    response = http.get(
        "/api/development/v1/governance-metrics/summary"
    )

    assert response.status_code == 200
    data = response.get_json()["data"]
    assert data["scope"] == {
        "global_access": False,
        "business_domain_uids": ["domain-a"],
    }
    assert data["metrics"][0]["rate"] == 0.666667
    assert service.calls[0][0] == "summary"


def test_viewer_traces_metric_to_bounded_details(monkeypatch):
    http, service = _client(monkeypatch)

    response = http.get(
        "/api/development/v1/governance-metrics/details"
        "?metric=asset_completeness&state=incomplete&page=2&page_size=25"
    )

    assert response.status_code == 200
    data = response.get_json()["data"]
    assert data["records"] == [
        {"asset_uid": "asset-1", "asset_name": "循环泵"}
    ]
    assert service.calls[-1][2] == {
        "metric": "asset_completeness",
        "state": "incomplete",
        "page": "2",
        "page_size": "25",
    }


def test_invalid_metric_returns_safe_400_contract(monkeypatch):
    from app.core.data_research.errors import GovernanceMetricInvalid

    http, service = _client(monkeypatch)

    def invalid(*_args, **_kwargs):
        raise GovernanceMetricInvalid("unsupported metric")

    service.details = invalid
    response = http.get(
        "/api/development/v1/governance-metrics/details"
        "?metric=ranking&state=top"
    )

    assert response.status_code == 400
    assert response.get_json()["error"]["code"] == (
        "GOVERNANCE_METRIC_INVALID"
    )


def test_unexpected_repository_failure_uses_existing_safe_error(monkeypatch):
    http, service = _client(monkeypatch)

    def unavailable(*_args, **_kwargs):
        raise RuntimeError("database secret")

    service.summary = unavailable
    response = http.get(
        "/api/development/v1/governance-metrics/summary"
    )

    assert response.status_code == 500
    payload = response.get_json()
    assert payload["message"] == "数据采集任务处理失败"
    assert "database secret" not in repr(payload)


def test_admin_scope_is_global_and_anonymous_is_denied(monkeypatch):
    admin_http, service = _client(monkeypatch, role="admin")
    admin_response = admin_http.get(
        "/api/development/v1/governance-metrics/summary"
    )

    assert admin_response.status_code == 200
    assert service.calls[0][1].global_access is True

    monkeypatch.setattr(
        "app.core.system.permissions.authenticate_request",
        lambda: None,
    )
    from app import create_app

    anonymous_response = create_app().test_client().get(
        "/api/development/v1/governance-metrics/summary"
    )
    assert anonymous_response.status_code == 401