DataOps-platform/tests/data_research/test_device_observability.py

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