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"