DataOps-platform/tests/data_research/test_device_observability_api.py

from __future__ import annotations

from datetime import UTC, datetime

import pytest

SOURCE_UID = "01900000-0000-7000-8000-000000000911"
DEVICE_UID = "01900000-0000-7000-8000-000000000912"
EVENT_UID = "01900000-0000-7000-8000-000000000913"
FAULT_UID = "01900000-0000-7000-8000-000000000914"
RELATION_UID = "01900000-0000-7000-8000-000000000915"


class FakeDeviceObservabilityService:
    def __init__(self):
        from app.core.data_research.device_observability import (
            EvidenceRelationRecord,
            GraphNode,
            GraphProjection,
            OperationalEventRecord,
            RootCauseAnalysis,
            RootCauseCandidate,
        )

        now = datetime(2026, 7, 29, 11, tzinfo=UTC)
        self.event = OperationalEventRecord(
            uid=EVENT_UID,
            source_uid=SOURCE_UID,
            source_entity="downtime_events",
            source_code="DT-001",
            event_type="downtime",
            asset_uid=DEVICE_UID,
            component_uid=None,
            title="一号泵停机",
            severity="critical",
            status="resolved",
            occurred_at=now,
            ended_at=now,
            evidence_refs=(
                {"type": "source_evidence", "data": {"duration": 18}},
            ),
            content_hash="a" * 64,
            created_by="editor-1",
            created_at=now,
        )
        self.fault_node = GraphNode(
            kind="event",
            uid=FAULT_UID,
            node_type="fault",
            label="轴承故障",
            occurred_at=now,
            evidence_refs=(),
        )
        self.anchor_node = GraphNode(
            kind="event",
            uid=EVENT_UID,
            node_type="downtime",
            label="一号泵停机",
            occurred_at=now,
            evidence_refs=self.event.evidence_refs,
        )
        self.relation = EvidenceRelationRecord(
            uid=RELATION_UID,
            from_kind="event",
            from_uid=FAULT_UID,
            relation_type="triggered",
            to_kind="event",
            to_uid=EVENT_UID,
            evidence_refs=(
                {"type": "source_evidence", "data": {"work_order": "WO-1"}},
            ),
            source="imported",
            created_by="editor-1",
            created_at=now,
        )
        self.graph_result = GraphProjection(
            anchor_kind="event",
            anchor_uid=EVENT_UID,
            nodes=(self.anchor_node, self.fault_node),
            relations=(self.relation,),
            truncated=False,
        )
        self.analysis = RootCauseAnalysis(
            analysis_status="supported_candidates",
            conclusion="发现有证据支持的根因候选，仍需设备专家确认",
            anchor=self.anchor_node,
            candidates=(
                RootCauseCandidate(
                    event_uid=FAULT_UID,
                    event_type="fault",
                    title="轴承故障",
                    occurred_at=now,
                    support_level="direct",
                    path_node_uids=(FAULT_UID, EVENT_UID),
                    path_relation_uids=(RELATION_UID,),
                    path_relation_types=("triggered",),
                    evidence_refs=self.relation.evidence_refs,
                ),
            ),
            graph=self.graph_result,
            limitations=("不等同于已确认根因",),
            generated_at=now,
        )
        self.actions = []

    def events(self, filters, *, page, page_size):
        self.actions.append(("events", filters, page, page_size))
        return [self.event], 1

    def import_evidence(self, payload, actor_uid):
        from app.core.data_research.device_observability import (
            ObservabilityImportResult,
        )

        self.actions.append(("import", payload, actor_uid))
        return ObservabilityImportResult(
            events=(self.event,),
            relations=(self.relation,),
            created_events=1,
            existing_events=0,
            created_relations=1,
            existing_relations=0,
        )

    def graph(self, anchor_kind, anchor_uid, *, max_hops):
        self.actions.append(("graph", anchor_kind, anchor_uid, max_hops))
        return self.graph_result

    def root_cause(self, anchor_kind, anchor_uid, *, max_hops):
        self.actions.append(
            ("root_cause", anchor_kind, anchor_uid, max_hops)
        )
        return self.analysis


@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 = FakeDeviceObservabilityService()

    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_observability_service",
        lambda: service,
        raising=False,
    )
    app = create_app()
    app.config.update(TESTING=True)
    return app.test_client(), service


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

    listing = http.get(
        "/api/development/v1/device-observability/events"
        f"?event_type=downtime&asset_uid={DEVICE_UID}&page=1&page_size=20",
        headers=headers,
    )
    graph = http.get(
        "/api/development/v1/device-observability/graph"
        f"?anchor_kind=event&anchor_uid={EVENT_UID}&max_hops=2",
        headers=headers,
    )
    analysis = http.get(
        "/api/development/v1/device-observability/root-cause"
        f"?anchor_kind=event&anchor_uid={EVENT_UID}&max_hops=3",
        headers=headers,
    )
    denied = http.post(
        "/api/development/v1/device-observability/import",
        headers=headers,
        json={"source_uid": SOURCE_UID, "events": [], "relations": []},
    )

    assert listing.status_code == 200
    assert listing.get_json()["data"]["records"][0]["event_type"] == (
        "downtime"
    )
    assert graph.status_code == 200
    assert graph.get_json()["data"]["relations"][0]["relation_type"] == (
        "triggered"
    )
    assert analysis.status_code == 200
    data = analysis.get_json()["data"]
    assert data["analysis_status"] == "supported_candidates"
    assert data["candidates"][0]["path_relation_types"] == ["triggered"]
    assert denied.status_code == 403
    assert service.actions[0][0] == "events"


def test_editor_imports_bounded_operational_evidence(client):
    http, service = client
    response = http.post(
        "/api/development/v1/device-observability/import",
        headers={"Authorization": "Bearer editor"},
        json={
            "source_uid": SOURCE_UID,
            "events": [
                {
                    "source_entity": "downtime_events",
                    "source_code": "DT-001",
                    "event_type": "downtime",
                    "asset_uid": DEVICE_UID,
                    "title": "一号泵停机",
                    "severity": "critical",
                    "status": "resolved",
                    "occurred_at": "2026-07-29T11:00:00+00:00",
                    "evidence": {"duration": 18},
                }
            ],
            "relations": [],
        },
    )

    assert response.status_code == 201
    assert response.get_json()["data"]["created_events"] == 1
    assert service.actions[-1][0] == "import"
    assert service.actions[-1][2] == "editor-1"