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()