from datetime import datetime, timezone from uuid import uuid4 import pytest from pydantic import ValidationError from app.domain.schemas.workbench_runtime import WorkbenchPersistedAlert from app.services.workbench_runtime import ( build_workbench_runtime_snapshot, project_latest_analysis, project_sensor_location_map_payload, sensor_location_ids_for_map_projection, ) NOW = datetime(2026, 9, 17, 2, 30, tzinfo=timezone.utc) @pytest.fixture def anyio_backend(): return "asyncio" def test_workbench_alert_rejects_naive_datetimes(): with pytest.raises(ValidationError): WorkbenchPersistedAlert.model_validate({ "alert_id": "alert-1", "condition_id": None, "source": { "task_type": "scheduled_task", "task_id": "schedule-1", "run_id": "occurrence-1", }, "title": "test", "description": "test", "occurred_at": "2026-09-18T02:30:00", "severity": "warning", "status": "active", }) @pytest.mark.anyio async def test_snapshot_reports_not_configured_without_persisted_sections(): snapshot = await build_workbench_runtime_snapshot( clock=lambda: NOW, ) assert snapshot == { "availability": "not_configured", "source": "none", "scenario_availability": "not_configured", "artifact_availability": "not_configured", "schedule_availability": "not_configured", "work_order_availability": "not_configured", "alert_availability": "not_configured", "generated_at": NOW, "conditions": [], "alerts": [], "scenarios": [], "latest_analysis": None, "work_orders": [], } def analysis_run(**overrides): value = { "run_id": uuid4(), "name": "北区爆管定位", "run_type": "burst_location", "created_by": "alice", "created_at": NOW, "started_at": NOW, "status": "completed", "parameters": {}, } value.update(overrides) return value @pytest.mark.anyio async def test_snapshot_keeps_analysis_results_out_of_operational_conditions(): failed_analysis = analysis_run(status="failed") result_created_at = NOW.replace(minute=31) snapshot = await build_workbench_runtime_snapshot( analysis_runs=[failed_analysis], latest_analysis_results=[ { "result_type": "burst_location", "created_at": result_created_at, "payload": { "drawing_payload": { "type": "FeatureCollection", "features": [{ "type": "Feature", "properties": {"label": "后端结果"}, "geometry": { "type": "Point", "coordinates": [121.5, 30.9], }, }], } }, } ], clock=lambda: NOW, ) assert snapshot["source"] == "none" assert snapshot["scenario_availability"] == "ready" assert snapshot["artifact_availability"] == "ready" assert snapshot["schedule_availability"] == "not_configured" assert snapshot["work_order_availability"] == "not_configured" assert snapshot["conditions"] == [] assert snapshot["scenarios"] == [ { "id": f"analysis-run-{failed_analysis['run_id']}", "run_id": str(failed_analysis["run_id"]), "name": "北区爆管定位", "description": "爆管定位 · 已失败 · 创建人 alice", "status": "review", "run_type": "burst_location", "run_type_label": "爆管定位", "run_status": "failed", "created_at": NOW, } ] assert snapshot["latest_analysis"]["results"] == [ { "result_type": "burst_location", "created_at": result_created_at, "facts": [], } ] assert snapshot["latest_analysis"]["map_payload"]["features"][0]["properties"] == { "label": "后端结果" } assert snapshot["alerts"] == [] def test_latest_analysis_projects_only_known_result_facts(): projected = project_latest_analysis( analysis_run(run_type="sensor_placement"), [{ "result_type": "sensor_placement", "node_id": "J-1", "created_at": NOW, "payload": { "sensor_number": 2, "minimum_diameter": 300, "sensor_locations": ["J-1", "J-2"], "internal_debug": "must-not-leak", }, }], ) assert projected["results"][0]["facts"] == [ {"label": "关联节点", "value": "J-1"}, {"label": "测点数量", "value": "2"}, {"label": "最小管径参数", "value": "300"}, {"label": "测点位置", "value": "J-1、J-2"}, ] assert "must-not-leak" not in str(projected) def test_sensor_location_projection_uses_latest_result_order_and_known_nodes(): results = [ { "result_type": "sensor_placement", "payload": {"sensor_locations": ["J-old"]}, }, { "result_type": "sensor_placement", "payload": {"sensor_locations": ["J-2", "J-1", "J-2"]}, }, ] node_ids = sensor_location_ids_for_map_projection(results) payload = project_sensor_location_map_payload(node_ids, [ {"node_id": "J-1", "longitude": 121.4, "latitude": 30.8}, {"node_id": "J-2", "longitude": 121.5, "latitude": 30.9}, ]) assert node_ids == ["J-2", "J-1"] assert payload is not None assert [feature["id"] for feature in payload["features"]] == ["J-2", "J-1"] @pytest.mark.anyio async def test_snapshot_projects_persisted_work_orders_and_alerts(): work_order = { "id": "wo-1", "code": "WO-20260917-001", "scheduled_at": NOW.replace(hour=3), "title": "北区阀门现场复核", "summary": "复核阀门开度并回传。", "status": "running", "risk_level": "attention", "updated_at": NOW, "detail": "来自正式工单同步适配器。", "recommendation": "完成后复核压力。", "duration_minutes": 45, "source": "生产工单系统", "dispatcher": "调度中心", "priority": "urgent", "location": "北区阀门 V-1", "assignee": "北区班组", "feedback_requirements": ["回传阀门开度", "复核压力"], "activity": [{"sequence": 1, "occurred_at": NOW, "actor": "调度中心", "kind": "created", "title": "工单创建", "detail": "等待现场复核。"}], } persisted_alert = { "alert_id": "alarm-1", "condition_id": None, "source": { "task_type": "scheduled_task", "task_id": "scada-diagnosis-5m", "run_id": "trigger-1", }, "title": "北区设备通信告警", "description": "由 SCADA 数据诊断任务明确推送。", "occurred_at": NOW.replace(minute=29), "severity": "warning", "status": "active", } snapshot = await build_workbench_runtime_snapshot( work_order_storage_configured=True, work_orders=[work_order], alert_storage_configured=True, persisted_alerts=[persisted_alert], clock=lambda: NOW, ) assert snapshot["availability"] == "ready" assert snapshot["source"] == "none" assert snapshot["schedule_availability"] == "not_configured" assert snapshot["work_order_availability"] == "ready" assert snapshot["alert_availability"] == "ready" assert snapshot["conditions"] == [] assert snapshot["alerts"] == [{ "id": "backend-alert-alarm-1", "condition_id": None, "source": persisted_alert["source"], "title": "北区设备通信告警", "description": "由 SCADA 数据诊断任务明确推送。", "occurred_at": NOW.replace(minute=29), "severity": "warning", }] assert snapshot["work_orders"][0]["code"] == "WO-20260917-001" assert snapshot["work_orders"][0]["activity"][0]["title"] == "工单创建" assert "subtasks" not in snapshot["work_orders"][0]