DataOps-platform/tests/integration/test_device_observability_postgres.py

from __future__ import annotations

import os
import uuid
from datetime import UTC, datetime

import pytest
from sqlalchemy import text

pytestmark = pytest.mark.integration


def test_device_observability_persists_graph_and_root_cause(monkeypatch):
    platform_url = os.environ.get("TEST_DATABASE_URL")
    if not platform_url:
        pytest.skip("TEST_DATABASE_URL is required")

    monkeypatch.setenv("DATABASE_URL", platform_url)
    from app import create_app, db
    from app.core.data_research.device_observability import (
        DeviceObservabilityConflict,
        DeviceObservabilityService,
    )
    from app.core.data_research.device_observability_repository import (
        SqlAlchemyDeviceObservabilityRepository,
    )
    from app.models.data_research import DeviceAsset, IngestionSource

    app = create_app()
    app.config.update(TESTING=True)
    suffix = uuid.uuid4().hex[:10]
    actor_uid = str(uuid.uuid4())
    source_uid = str(uuid.uuid4())
    device_uid = str(uuid.uuid4())
    now = datetime.now(UTC).replace(microsecond=0)

    try:
        with app.app_context():
            db.session.execute(
                text(
                    """
                    INSERT INTO public.users (
                        id, username, display_name, password_hash, status
                    ) VALUES (
                        CAST(:id AS uuid), :username, :username,
                        'integration-test', 'active'
                    )
                    """
                ),
                {
                    "id": actor_uid,
                    "username": f"wp09-actor-{suffix}",
                },
            )
            db.session.add(
                IngestionSource(
                    uid=source_uid,
                    source_type="database",
                    name=f"WP09 source {suffix}",
                    config={"database_type": "postgresql"},
                    permission_scope={},
                    status="active",
                    created_by=actor_uid,
                )
            )
            db.session.add(
                DeviceAsset(
                    uid=device_uid,
                    asset_type="device",
                    name=f"WP09 pump {suffix}",
                    status="active",
                    current_version=1,
                    content_hash="9" * 64,
                    attributes={},
                    created_by=actor_uid,
                    updated_by=actor_uid,
                )
            )
            db.session.commit()

            repository = SqlAlchemyDeviceObservabilityRepository(db.session)
            service = DeviceObservabilityService(
                repository,
                commit=db.session.commit,
                rollback=db.session.rollback,
            )
            imported = service.import_evidence(
                {
                    "source_uid": source_uid,
                    "events": [
                        {
                            "source_entity": "alarm_events",
                            "source_code": f"AL-{suffix}",
                            "event_type": "alarm",
                            "asset_uid": device_uid,
                            "title": "轴承温度高",
                            "severity": "warning",
                            "status": "observed",
                            "occurred_at": now.isoformat(),
                            "evidence": {"sample_ref": f"S-{suffix}"},
                        },
                        {
                            "source_entity": "fault_events",
                            "source_code": f"FT-{suffix}",
                            "event_type": "fault",
                            "asset_uid": device_uid,
                            "title": "轴承故障",
                            "severity": "error",
                            "status": "observed",
                            "occurred_at": now.isoformat(),
                            "evidence": {"fault_code": "F-001"},
                        },
                        {
                            "source_entity": "downtime_events",
                            "source_code": f"DT-{suffix}",
                            "event_type": "downtime",
                            "asset_uid": device_uid,
                            "title": "泵停机",
                            "severity": "critical",
                            "status": "resolved",
                            "occurred_at": now.isoformat(),
                            "ended_at": now.isoformat(),
                            "evidence": {"duration_minutes": 18},
                        },
                    ],
                    "relations": [
                        {
                            "from": {
                                "kind": "event",
                                "source_entity": "alarm_events",
                                "source_code": f"AL-{suffix}",
                            },
                            "relation_type": "indicates",
                            "to": {
                                "kind": "event",
                                "source_entity": "fault_events",
                                "source_code": f"FT-{suffix}",
                            },
                            "evidence": {"window_minutes": 5},
                        },
                        {
                            "from": {
                                "kind": "event",
                                "source_entity": "fault_events",
                                "source_code": f"FT-{suffix}",
                            },
                            "relation_type": "triggered",
                            "to": {
                                "kind": "event",
                                "source_entity": "downtime_events",
                                "source_code": f"DT-{suffix}",
                            },
                            "evidence": {"operator_log_ref": f"OP-{suffix}"},
                        },
                    ],
                },
                actor_uid,
            )
            assert imported.created_events == 3
            assert imported.created_relations == 2

            duplicate = service.import_evidence(
                {
                    "source_uid": source_uid,
                    "events": [
                        {
                            "source_entity": "alarm_events",
                            "source_code": f"AL-{suffix}",
                            "event_type": "alarm",
                            "asset_uid": device_uid,
                            "title": "轴承温度高",
                            "severity": "warning",
                            "status": "observed",
                            "occurred_at": now.isoformat(),
                            "evidence": {"sample_ref": f"S-{suffix}"},
                        }
                    ],
                    "relations": [],
                },
                actor_uid,
            )
            assert duplicate.existing_events == 1
            with pytest.raises(DeviceObservabilityConflict):
                service.import_evidence(
                    {
                        "source_uid": source_uid,
                        "events": [
                            {
                                "source_entity": "alarm_events",
                                "source_code": f"AL-{suffix}",
                                "event_type": "alarm",
                                "asset_uid": device_uid,
                                "title": "冲突告警标题",
                                "severity": "warning",
                                "status": "observed",
                                "occurred_at": now.isoformat(),
                                "evidence": {},
                            }
                        ],
                        "relations": [],
                    },
                    actor_uid,
                )

            records, total = repository.search_events(
                {"event_type": "downtime", "asset_uid": device_uid},
                page=1,
                page_size=20,
            )
            assert total == 1
            assert records[0].title == "泵停机"
            graph = service.graph("event", records[0].uid, max_hops=3)
            assert len(graph.nodes) == 3
            assert [item.relation_type for item in graph.relations] == [
                "triggered",
                "indicates",
            ]
            analysis = service.root_cause(
                "event",
                records[0].uid,
                max_hops=3,
            )
            assert analysis.analysis_status == "supported_candidates"
            assert [item.event_type for item in analysis.candidates] == [
                "fault",
                "alarm",
            ]
    finally:
        with app.app_context():
            db.session.execute(
                text(
                    "DELETE FROM public.device_evidence_relations "
                    "WHERE created_by = CAST(:actor_uid AS uuid)"
                ),
                {"actor_uid": actor_uid},
            )
            db.session.execute(
                text(
                    "DELETE FROM public.device_operational_events "
                    "WHERE created_by = CAST(:actor_uid AS uuid)"
                ),
                {"actor_uid": actor_uid},
            )
            db.session.execute(
                text(
                    "DELETE FROM public.device_assets "
                    "WHERE uid = CAST(:uid AS uuid)"
                ),
                {"uid": device_uid},
            )
            db.session.execute(
                text(
                    "DELETE FROM public.ingestion_sources "
                    "WHERE uid = CAST(:uid AS uuid)"
                ),
                {"uid": source_uid},
            )
            db.session.execute(
                text(
                    "DELETE FROM public.users "
                    "WHERE id = CAST(:id AS uuid)"
                ),
                {"id": actor_uid},
            )
            db.session.commit()