from __future__ import annotations from dataclasses import replace from datetime import UTC, datetime import pytest from app.core.data_research.device_observability import ( DeviceObservabilityConflict, DeviceObservabilityInvalid, DeviceObservabilityNotFound, DeviceObservabilityService, EvidenceRelationRecord, GraphNode, GraphProjection, ) SOURCE_UID = "11111111-1111-4111-8111-111111111111" DEVICE_UID = "22222222-2222-4222-8222-222222222222" ISSUE_UID = "33333333-3333-4333-8333-333333333333" ACTOR_UID = "44444444-4444-4444-8444-444444444444" class MemoryRepository: def __init__(self): self.events = {} self.identities = {} self.relations = {} self.nodes = { ("asset", DEVICE_UID): GraphNode( kind="asset", uid=DEVICE_UID, node_type="device", label="一号泵", occurred_at=None, evidence_refs=(), ), ("quality_issue", ISSUE_UID): GraphNode( kind="quality_issue", uid=ISSUE_UID, node_type="quality_issue", label="QI-001", occurred_at=None, evidence_refs=(), ), } def source_exists(self, source_uid): return source_uid == SOURCE_UID def get_event_by_identity(self, source_uid, source_entity, source_code): uid = self.identities.get((source_uid, source_entity, source_code)) return self.events.get(uid) def add_event(self, record): self.events[record.uid] = record self.identities[ (record.source_uid, record.source_entity, record.source_code) ] = record.uid self.nodes[("event", record.uid)] = GraphNode( kind="event", uid=record.uid, node_type=record.event_type, label=record.title, occurred_at=record.occurred_at, evidence_refs=record.evidence_refs, ) def resolve_event_identity(self, source_uid, source_entity, source_code): return self.identities.get((source_uid, source_entity, source_code)) def node_exists(self, kind, uid): return (kind, uid) in self.nodes def get_relation(self, record): return self.relations.get(record.identity) def add_relation(self, record): self.relations[record.identity] = record def graph(self, anchor_kind, anchor_uid, max_hops, max_nodes, max_edges): assert max_hops <= 3 nodes = {f"{anchor_kind}:{anchor_uid}": self.nodes[(anchor_kind, anchor_uid)]} edges = [] frontier = {(anchor_kind, anchor_uid)} visited = set(frontier) for _ in range(max_hops): next_frontier = set() for relation in self.relations.values(): source = (relation.from_kind, relation.from_uid) target = (relation.to_kind, relation.to_uid) if source not in frontier and target not in frontier: continue if len(edges) >= max_edges: break edges.append(relation) for endpoint in (source, target): if endpoint not in visited and len(nodes) < max_nodes: visited.add(endpoint) next_frontier.add(endpoint) nodes[f"{endpoint[0]}:{endpoint[1]}"] = self.nodes[endpoint] frontier = next_frontier if not frontier: break return GraphProjection( anchor_kind=anchor_kind, anchor_uid=anchor_uid, nodes=tuple(nodes.values()), relations=tuple(edges), truncated=False, ) def service(repository=None, uid_values=None): values = iter( uid_values or ( "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa1", "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa2", "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa3", "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa4", "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa5", "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa6", "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa7", "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa8", ) ) return DeviceObservabilityService( repository or MemoryRepository(), uid_factory=lambda: next(values), ) def event(source_code, event_type, title, **extra): return { "source_entity": f"{event_type}_events", "source_code": source_code, "event_type": event_type, "asset_uid": DEVICE_UID, "title": title, "severity": extra.pop("severity", "warning"), "status": extra.pop("status", "observed"), "occurred_at": extra.pop( "occurred_at", datetime(2026, 7, 29, 8, 0, tzinfo=UTC).isoformat(), ), "evidence": extra.pop("evidence", {"source_row": source_code}), **extra, } def event_endpoint(event_type, source_code): return { "kind": "event", "source_entity": f"{event_type}_events", "source_code": source_code, } def test_import_is_idempotent_by_source_identity_and_rejects_conflicts(): repository = MemoryRepository() observability = service(repository) payload = { "source_uid": SOURCE_UID, "events": [event("AL-1", "alarm", "轴承温度高")], "relations": [], } created = observability.import_evidence(payload, ACTOR_UID) duplicate = observability.import_evidence(payload, ACTOR_UID) assert created.created_events == 1 assert isinstance(created.events[0].occurred_at, datetime) assert created.events[0].occurred_at.tzinfo is not None assert duplicate.existing_events == 1 with pytest.raises(DeviceObservabilityConflict): observability.import_evidence( { **payload, "events": [event("AL-1", "alarm", "轴承振动高")], }, ACTOR_UID, ) def test_import_rejects_secrets_and_invalid_event_windows(): observability = service() with pytest.raises(DeviceObservabilityInvalid, match="secret"): observability.import_evidence( { "source_uid": SOURCE_UID, "events": [ event( "AL-2", "alarm", "高温告警", evidence={"password": "must-not-persist"}, ) ], "relations": [], }, ACTOR_UID, ) with pytest.raises(DeviceObservabilityInvalid, match="ended_at"): observability.import_evidence( { "source_uid": SOURCE_UID, "events": [ event( "DT-1", "downtime", "设备停机", ended_at="2026-07-29T07:00:00+00:00", ) ], "relations": [], }, ACTOR_UID, ) def test_relations_resolve_new_event_identities_and_require_existing_nodes(): repository = MemoryRepository() observability = service(repository) result = observability.import_evidence( { "source_uid": SOURCE_UID, "events": [ event("AL-3", "alarm", "压力告警"), event("FT-3", "fault", "泵体故障"), ], "relations": [ { "from": event_endpoint("alarm", "AL-3"), "relation_type": "indicates", "to": event_endpoint("fault", "FT-3"), "evidence": {"rule": "same-device-window"}, } ], }, ACTOR_UID, ) assert result.created_relations == 1 relation = next(iter(repository.relations.values())) assert relation.from_kind == "event" assert relation.to_kind == "event" with pytest.raises(DeviceObservabilityConflict): observability.import_evidence( { "source_uid": SOURCE_UID, "events": [], "relations": [ { "from": event_endpoint("alarm", "AL-3"), "relation_type": "indicates", "to": event_endpoint("fault", "FT-3"), "evidence": {"rule": "different-evidence"}, } ], }, ACTOR_UID, ) with pytest.raises(DeviceObservabilityInvalid, match="does not exist"): observability.import_evidence( { "source_uid": SOURCE_UID, "events": [], "relations": [ { "from": {"kind": "asset", "uid": DEVICE_UID}, "relation_type": "part_of", "to": { "kind": "asset", "uid": "99999999-9999-4999-8999-999999999999", }, "evidence": {}, } ], }, ACTOR_UID, ) def test_graph_limits_hops_and_requires_a_real_anchor(): observability = service() with pytest.raises(DeviceObservabilityInvalid, match="max_hops"): observability.graph("asset", DEVICE_UID, max_hops=4) with pytest.raises(DeviceObservabilityNotFound, match="does not exist"): observability.graph( "asset", "99999999-9999-4999-8999-999999999999", max_hops=2, ) def test_root_cause_returns_evidence_paths_without_claiming_a_verdict(): repository = MemoryRepository() observability = service(repository) observability.import_evidence( { "source_uid": SOURCE_UID, "events": [ event("AL-4", "alarm", "轴承温度高"), event("FT-4", "fault", "轴承故障"), event("DT-4", "downtime", "一号泵停机"), ], "relations": [ { "from": event_endpoint("alarm", "AL-4"), "relation_type": "indicates", "to": event_endpoint("fault", "FT-4"), "evidence": {"window_minutes": 5}, }, { "from": event_endpoint("fault", "FT-4"), "relation_type": "triggered", "to": event_endpoint("downtime", "DT-4"), "evidence": {"work_order": "WO-4"}, }, ], }, ACTOR_UID, ) anchor_uid = repository.resolve_event_identity( SOURCE_UID, "downtime_events", "DT-4", ) result = observability.root_cause("event", anchor_uid, max_hops=3) assert result.analysis_status == "supported_candidates" assert result.conclusion == "发现有证据支持的根因候选，仍需设备专家确认" assert [item.event_type for item in result.candidates] == ["fault", "alarm"] assert result.candidates[0].path_relation_types == ("triggered",) assert result.candidates[1].path_relation_types == ( "indicates", "triggered", ) assert "不等同于已确认根因" in result.limitations[0] def test_root_cause_explicitly_reports_insufficient_evidence(): repository = MemoryRepository() observability = service(repository) imported = observability.import_evidence( { "source_uid": SOURCE_UID, "events": [event("DT-5", "downtime", "二号泵停机")], "relations": [ { "from": event_endpoint("downtime", "DT-5"), "relation_type": "occurred_on", "to": {"kind": "asset", "uid": DEVICE_UID}, "evidence": {"source": "downtime-log"}, } ], }, ACTOR_UID, ) anchor_uid = imported.events[0].uid result = observability.root_cause("event", anchor_uid) assert result.analysis_status == "insufficient_evidence" assert result.conclusion == "证据不足，无法确认根因" assert result.candidates == () def test_root_cause_does_not_treat_a_future_event_as_a_candidate_cause(): repository = MemoryRepository() observability = service(repository) imported = observability.import_evidence( { "source_uid": SOURCE_UID, "events": [ event( "DT-6", "downtime", "三号泵停机", occurred_at="2026-07-29T08:00:00+00:00", ), event( "FT-6", "fault", "停机后补录故障", occurred_at="2026-07-29T09:00:00+00:00", ), ], "relations": [ { "from": event_endpoint("fault", "FT-6"), "relation_type": "triggered", "to": event_endpoint("downtime", "DT-6"), "evidence": {"source": "late-manual-entry"}, } ], }, ACTOR_UID, ) anchor_uid = next( item.uid for item in imported.events if item.event_type == "downtime" ) result = observability.root_cause("event", anchor_uid) assert result.analysis_status == "insufficient_evidence" assert result.candidates == () def test_relation_identity_includes_direction_and_type(): base = EvidenceRelationRecord( uid="aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", from_kind="event", from_uid="bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb", relation_type="indicates", to_kind="event", to_uid="cccccccc-cccc-4ccc-8ccc-cccccccccccc", evidence_refs=(), source="imported", created_by=ACTOR_UID, created_at=None, ) assert base.identity != replace(base, relation_type="triggered").identity assert base.identity != replace( base, from_uid=base.to_uid, to_uid=base.from_uid, ).identity