DataOps-platform/tests/data_research/test_device_quality_api.py

from __future__ import annotations

from datetime import UTC, datetime, timedelta

import pytest

VERSION_UID = "01900000-0000-7000-8000-000000000801"
RUN_UID = "01900000-0000-7000-8000-000000000802"
ASSET_UID = "01900000-0000-7000-8000-000000000803"
SOURCE_UID = "01900000-0000-7000-8000-000000000804"
MAPPING_UID = "01900000-0000-7000-8000-000000000805"


class FakeDeviceQualityService:
    def __init__(self):
        from app.core.data_research.device_quality import (
            DEFAULT_DEVICE_QUALITY_RULES,
            DeviceQualityAssetScoreRecord,
            DeviceQualityPolicyVersionRecord,
            DeviceQualityRuleResultRecord,
            DeviceQualityRunRecord,
            DeviceQualityViolationRecord,
            validate_device_quality_rules,
        )

        now = datetime(2026, 7, 29, 17, 0, tzinfo=UTC)
        self.version = DeviceQualityPolicyVersionRecord(
            uid=VERSION_UID,
            profile_uid="01900000-0000-7000-8000-000000000806",
            version=1,
            status="published",
            rules=validate_device_quality_rules(
                DEFAULT_DEVICE_QUALITY_RULES
            ),
            content_hash="a" * 64,
            created_by="editor-1",
            created_at=now,
            published_by="admin-1",
            published_at=now,
        )
        self.run_record = DeviceQualityRunRecord(
            uid=RUN_UID,
            policy_version_uid=VERSION_UID,
            policy_hash="a" * 64,
            source_uid=SOURCE_UID,
            status="success",
            total_assets=1,
            total_violations=1,
            score=85.0,
            created_by="editor-1",
            created_at=now,
        )
        self.rule_result = DeviceQualityRuleResultRecord(
            uid="01900000-0000-7000-8000-000000000807",
            run_uid=RUN_UID,
            rule_code="asset_context_complete",
            severity="error",
            weight=15.0,
            status="violated",
            evaluated_count=1,
            violation_count=1,
            sampled_count=1,
            pass_rate=0.0,
            weighted_score=0.0,
            created_at=now,
        )
        self.violation = DeviceQualityViolationRecord(
            uid="01900000-0000-7000-8000-000000000808",
            run_uid=RUN_UID,
            rule_code="asset_context_complete",
            severity="error",
            asset_uid=ASSET_UID,
            field_name="location",
            source_uid=SOURCE_UID,
            source_mapping_uid=MAPPING_UID,
            message="设备台账位置、组织或责任人不完整",
            evidence={
                "asset_uid": ASSET_UID,
                "source_mapping_uid": MAPPING_UID,
                "missing_fields": ["location"],
            },
            created_at=now,
            expires_at=now + timedelta(days=30),
        )
        self.asset_score = DeviceQualityAssetScoreRecord(
            uid="01900000-0000-7000-8000-000000000809",
            run_uid=RUN_UID,
            asset_uid=ASSET_UID,
            asset_type="device",
            status="violated",
            evaluated_rule_count=3,
            violation_count=1,
            score=62.5,
            created_at=now,
        )
        self.actions = []

    def profile(self):
        self.actions.append(("profile",))
        return {
            "profile_uid": self.version.profile_uid,
            "name": "设备台账与故障质量策略",
            "latest_version": self.version,
            "active_version": self.version,
        }

    def bootstrap(self, *, actor_uid):
        self.actions.append(("bootstrap", actor_uid))
        return self.version

    def revise(self, *, rules, expected_version, actor_uid):
        self.actions.append(
            ("revise", rules, expected_version, actor_uid)
        )
        return self.version

    def versions(self):
        self.actions.append(("versions",))
        return [self.version]

    def publish(self, version_uid, *, actor_uid):
        self.actions.append(("publish", version_uid, actor_uid))
        return self.version

    def run(self, *, actor_uid, source_uid=None):
        self.actions.append(("run", actor_uid, source_uid))
        return self.run_record

    def runs(self, *, page, page_size):
        self.actions.append(("runs", page, page_size))
        return [self.run_record], 1

    def get_run(self, run_uid):
        self.actions.append(("get_run", run_uid))
        return self.run_record, [self.rule_result]

    def violations(self, run_uid, *, rule_code, page, page_size):
        self.actions.append(
            ("violations", run_uid, rule_code, page, page_size)
        )
        return [self.violation], 1

    def asset_scores(self, run_uid, *, page, page_size):
        self.actions.append(("asset_scores", run_uid, page, page_size))
        return [self.asset_score], 1


@pytest.fixture()
def client(monkeypatch):
    from flask import request

    from app import create_app
    from app.api.data_development import routes
    from app.core.system import permissions

    service = FakeDeviceQualityService()

    def identity():
        role = request.headers.get(
            "Authorization",
            "",
        ).removeprefix("Bearer ")
        if role not in {"viewer", "editor", "admin"}:
            return None
        return {"id": f"{role}-1", "roles": [role]}

    monkeypatch.setattr(permissions, "authenticate_request", identity)
    monkeypatch.setattr(
        routes,
        "get_device_quality_service",
        lambda: service,
        raising=False,
    )
    app = create_app()
    app.config.update(TESTING=True)
    return app.test_client(), service


def test_viewer_reads_profile_runs_and_evidence_but_cannot_bootstrap(client):
    http, service = client
    headers = {"Authorization": "Bearer viewer"}

    profile = http.get(
        "/api/development/v1/device-quality/profile",
        headers=headers,
    )
    versions = http.get(
        "/api/development/v1/device-quality/profile/versions",
        headers=headers,
    )
    runs = http.get(
        "/api/development/v1/device-quality/runs?page=1&page_size=20",
        headers=headers,
    )
    detail = http.get(
        f"/api/development/v1/device-quality/runs/{RUN_UID}",
        headers=headers,
    )
    violations = http.get(
        f"/api/development/v1/device-quality/runs/{RUN_UID}/violations"
        "?rule_code=asset_context_complete&page=1&page_size=20",
        headers=headers,
    )
    scores = http.get(
        f"/api/development/v1/device-quality/runs/{RUN_UID}/asset-scores"
        "?page=1&page_size=20",
        headers=headers,
    )
    denied = http.post(
        "/api/development/v1/device-quality/bootstrap",
        headers=headers,
        json={},
    )

    assert profile.status_code == 200
    assert profile.get_json()["data"]["active_version"]["version"] == 1
    assert len(profile.get_json()["data"]["active_version"]["rules"]) == 7
    assert versions.get_json()["data"]["total"] == 1
    assert runs.get_json()["data"]["records"][0]["score"] == 85.0
    assert detail.get_json()["data"]["rule_results"][0]["pass_rate"] == 0.0
    assert violations.get_json()["data"]["records"][0]["asset_uid"] == (
        ASSET_UID
    )
    assert violations.get_json()["data"]["records"][0]["evidence"] == {
        "asset_uid": ASSET_UID,
        "missing_fields": ["location"],
        "source_mapping_uid": MAPPING_UID,
    }
    assert scores.get_json()["data"]["records"][0]["score"] == 62.5
    assert denied.status_code == 403
    assert service.actions[0] == ("profile",)


def test_editor_can_bootstrap_revise_and_execute_but_cannot_publish(client):
    http, service = client
    headers = {"Authorization": "Bearer editor"}

    bootstrapped = http.post(
        "/api/development/v1/device-quality/bootstrap",
        headers=headers,
        json={},
    )
    revised = http.post(
        "/api/development/v1/device-quality/profile/versions",
        headers=headers,
        json={
            "expected_version": 1,
            "rules": list(service.version.rules),
        },
    )
    executed = http.post(
        "/api/development/v1/device-quality/runs",
        headers=headers,
        json={"source_uid": SOURCE_UID},
    )
    denied = http.post(
        f"/api/development/v1/device-quality/profile/versions/"
        f"{VERSION_UID}/publish",
        headers=headers,
        json={},
    )

    assert bootstrapped.status_code == 201
    assert revised.status_code == 201
    assert executed.status_code == 201
    assert executed.get_json()["data"]["policy_version_uid"] == VERSION_UID
    assert denied.status_code == 403
    assert ("bootstrap", "editor-1") in service.actions
    assert ("run", "editor-1", SOURCE_UID) in service.actions


def test_admin_publishes_exact_version_through_runtime_gate(client):
    http, service = client
    response = http.post(
        f"/api/development/v1/device-quality/profile/versions/"
        f"{VERSION_UID}/publish",
        headers={"Authorization": "Bearer admin"},
        json={},
    )

    assert response.status_code == 200
    assert response.get_json()["data"]["status"] == "published"
    assert service.actions[-1] == ("publish", VERSION_UID, "admin-1")