import asyncio from contextlib import asynccontextmanager from copy import deepcopy from datetime import datetime, timedelta, timezone from types import SimpleNamespace from unittest.mock import AsyncMock from uuid import uuid4 import pytest from fastapi import HTTPException from app.domain.schemas.scada_replay import ScadaReplayCreate, ScadaReplayCommand from app.domain.schemas.project_settings import ScadaMonitoringSettings from app.infra.db.postgresql.scada import ScadaInfoRepository from app.infra.db.timescaledb.repositories.scada import ScadaRepository from app.services.scada_replay import BaselineBudgetExceeded, ScadaReplayService, snapshot START = datetime(2026, 10, 8, tzinfo=timezone.utc) class Connection: def __init__(self, owner=None): self.statements, self.depth, self.owner = [], 0, owner async def execute(self, statement): self.statements.append(statement) @asynccontextmanager async def transaction(self): prior = (deepcopy(self.owner.document), deepcopy(self.owner.batches)) if self.owner else None self.depth += 1 try: yield except BaseException: if prior: self.owner.document, self.owner.batches = prior raise finally: self.depth -= 1 class Repository: def __init__(self): self.project_id = str(uuid4()) self.connection = Connection(self) self.document = None self.batches = {} async def available(self): return True @asynccontextmanager async def control(self): yield @asynccontextmanager async def processing(self): yield True async def by_request(self, key): return deepcopy(self.document) if self.document and self.document["config"]["request_id"] == str(key) else None async def active_other(self, key): return False async def insert(self, document): self.document = deepcopy(document) async def save(self, document): self.document = deepcopy(document) async def get(self, _): return deepcopy(self.document) async def current(self): return deepcopy(self.document) async def pending_batch(self, _): return next((deepcopy(batch) for batch in self.batches.values() if batch["sequence"] > 0 and batch["stage"] != "detected"), None) async def save_batch(self, _, batch): self.batches[batch["sequence"]] = deepcopy(batch) async def clear_unpublished_baseline(self, _): self.batches = {key: value for key, value in self.batches.items() if not (key < 0 and value.get("kind") == "baseline" and value["stage"] == "clean_prepared")} async def publish_baseline(self, _): for key, value in self.batches.items(): if key < 0 and value.get("kind") == "baseline" and value["stage"] == "clean_prepared": value["stage"] = "detected" async def monitoring_outputs(self, _replay_id, device_ids, start, end, before_sequence): return [row for batch in self.batches.values() if batch["sequence"] < before_sequence and batch["stage"] == "detected" for row in batch.get("outputs", []) if row["device_id"] in device_ids and start <= datetime.fromisoformat(row["time"]) <= end] def setup(monkeypatch): repo = Repository() context = SimpleNamespace( project_id=repo.project_id, user_id=uuid4(), ) service = ScadaReplayService(repo, context, Connection(), clock=lambda: START + timedelta(seconds=2)) monkeypatch.setattr(ScadaInfoRepository, "get_scadas", AsyncMock(return_value=[{ "device_id": "P1", "device_type": "pressure", "measurement_unit": "m", "transmission_frequency": "0:15:00" }])) async def raw_page(_connection, _devices, _start, _end, _through, _limit, after=None, **_): if _end <= START: return [] return ([{"time": START, "device_id": "P1", "monitored_value": 20., "source_hash": "h"}] if after is None else []) query = AsyncMock(side_effect=raw_page) cleaning = AsyncMock() monkeypatch.setattr(ScadaRepository, "get_raw_page", query) monkeypatch.setattr(ScadaRepository, "apply_replay_cleaning", cleaning) monkeypatch.setattr(service, "_create_investigations", AsyncMock()) monkeypatch.setattr(service, "_publish_monitoring_alert", AsyncMock()) monkeypatch.setattr("app.services.scada_replay.permissions_for_context", lambda _: {"task.write"}) monkeypatch.setattr("app.services.operational_tasks.OperationalTaskService.create", AsyncMock(return_value=SimpleNamespace(task_id=uuid4()))) create = ScadaReplayCreate(request_id=uuid4(), start_time=START, end_time=START + timedelta(hours=1), device_ids=["P1"], rules=[{"rule_id": "p", "device_ids": ["P1"], "kind": "bounds", "upper": 10, "unit": "m"}]) return repo, service, query, cleaning, create def test_baseline_freezes_all_pages_silently_without_cleaning_the_source(monkeypatch): repo, service, query, cleaning, create = setup(monkeypatch) before = START - timedelta(minutes=15) query.side_effect = None query.side_effect = [ [{"time": before - timedelta(minutes=15), "device_id": "P1", "monitored_value": 100., "source_hash": "a"}], [{"time": before, "device_id": "P1", "monitored_value": 200., "source_hash": "b"}], [], ] async def run(): await service.create(create, monitoring_interval_seconds=900) document = await repo.get(None) await service._freeze_baseline(document, create) assert datetime.fromisoformat(repo.document["virtual_now"]) == create.start_time assert repo.document["batch_sequence"] == repo.document["processed_count"] == 0 assert repo.document["anomaly_count"] == repo.document["monitoring_count"] == 0 assert repo.document["rule_states"] == repo.document["latest"] == {} assert repo.document["baseline"]["row_count"] == 2 assert repo.document["histories"]["P1"] == [100., 200.] assert set(repo.batches) == {-1, -2} assert all(batch["kind"] == "baseline" and batch["stage"] == "detected" and batch["events"] == [] for batch in repo.batches.values()) assert repo.batches[-2]["outputs"][0]["cleaned_value"] == 200. asyncio.run(run()) assert service.timescale.statements == ["SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY"] assert query.await_args_list[0].args[2:4] == (START - timedelta(days=15), START) assert query.await_args_list[1].args[6] == (before - timedelta(minutes=15), "P1") cleaning.assert_not_awaited() service._create_investigations.assert_not_awaited() def test_snapshot_returns_frozen_monitoring_rules_and_actual_calibration(monkeypatch): repo, service, _, _, create = setup(monkeypatch) settings = ScadaMonitoringSettings(pressure_device_ids=["P1"]) async def run(): await service.create(create, monitoring_interval_seconds=900, monitoring_config=settings) repo.document["monitoring_calibration"] = { "status": "ready", "burst_detection": {"threshold": -0.0843}, "pressure": {"P1": {"threshold_percent": 6.5}}, } # Project settings may change later; the replay still reports its frozen configuration. settings.weights.pressure = 0.3 settings.weights.burst_detection = 0.6 result = snapshot(repo.document) assert result.monitoring_config.weights.pressure == 0.4 assert result.monitoring_config.weights.burst_detection == 0.5 assert result.monitoring_calibration["burst_detection"]["threshold"] == -0.0843 repo.document["monitoring_config"].pop("decision_policy") assert snapshot(repo.document).monitoring_config.decision_policy == "all_signals" asyncio.run(run()) def test_baseline_stages_each_page_before_reading_more_without_publishing_or_holding_business_transaction(monkeypatch): repo, service, query, _, create = setup(monkeypatch) calls = 0 async def raw_page(*args, **kwargs): nonlocal calls assert service.timescale.depth == 1 assert repo.connection.depth == 0 if calls: assert repo.batches[-calls]["stage"] == "clean_prepared" assert not repo.document.get("warmed") and "baseline" not in repo.document assert await repo.pending_batch(None) is None assert await repo.monitoring_outputs(None, ["P1"], START - timedelta(days=15), START, 1) == [] calls += 1 if calls > 2: return [] return [{"time": START - timedelta(minutes=45 - calls * 15), "device_id": "P1", "monitored_value": 30., "source_hash": str(calls)}] query.side_effect = raw_page async def run(): await service.create(create, monitoring_interval_seconds=900) await service._freeze_baseline(await repo.get(None), create) assert repo.document["warmed"] and repo.document["baseline"]["row_count"] == 2 assert all(batch["stage"] == "detected" for batch in repo.batches.values()) asyncio.run(run()) def test_failed_baseline_read_never_publishes_partial_history(monkeypatch): repo, service, query, cleaning, create = setup(monkeypatch) query.side_effect = [ [{"time": START - timedelta(minutes=15), "device_id": "P1", "monitored_value": 30., "source_hash": "a"}], RuntimeError("source unavailable"), ] async def run(): await service.create(create, monitoring_interval_seconds=900) with pytest.raises(RuntimeError, match="source unavailable"): await service._freeze_baseline(await repo.get(None), create) assert "baseline" not in repo.document and not repo.document.get("warmed") assert repo.batches[-1]["stage"] == "clean_prepared" assert await repo.pending_batch(None) is None assert await repo.monitoring_outputs(None, ["P1"], START - timedelta(days=15), START, 1) == [] asyncio.run(run()) cleaning.assert_not_awaited() def test_baseline_retry_discards_unpublished_pages_and_refreezes_one_source_snapshot(monkeypatch): repo, service, query, _, create = setup(monkeypatch) def row(minutes, value): return {"time": START - timedelta(minutes=minutes), "device_id": "P1", "monitored_value": value, "source_hash": str(value)} query.side_effect = [[row(30, 100.)], [row(15, 200.)], RuntimeError("source unavailable")] async def run(): first = await service.create(create, monitoring_interval_seconds=900) await service.command(first.replay_id, ScadaReplayCommand( request_id=uuid4(), expected_revision=1, action="step")) with pytest.raises(RuntimeError, match="source unavailable"): await service.process_pending_step(first.replay_id) assert repo.document["status"] == "failed" and set(repo.batches) == {-1, -2} assert all(batch["stage"] == "clean_prepared" for batch in repo.batches.values()) query.side_effect = [[row(15, 400.)], [], []] await service.command(first.replay_id, ScadaReplayCommand( request_id=uuid4(), expected_revision=2, action="retry")) result = await service.process_pending_step(first.replay_id) assert not result.step_pending and result.status == "paused" assert repo.document["baseline"]["row_count"] == 1 assert repo.document["histories"]["P1"] == [400.] assert -2 not in repo.batches and repo.batches[-1]["stage"] == "detected" assert repo.batches[-1]["outputs"][0]["cleaned_value"] == 400. assert result.anomaly_count == result.processed_count == 0 asyncio.run(run()) @pytest.mark.parametrize("budget", ["rows", "bytes"]) def test_baseline_budget_stops_before_saving_excess_rows_and_exposes_actionable_error(monkeypatch, budget): repo, service, query, cleaning, create = setup(monkeypatch) if budget == "rows": monkeypatch.setattr("app.services.scada_replay.BASELINE_MAX_ROWS", 2) else: monkeypatch.setattr("app.services.scada_replay.BASELINE_MAX_BYTES", 1) query.side_effect = [ [{"time": START - timedelta(minutes=minute), "device_id": "P1", "monitored_value": 30., "source_hash": str(minute)} for minute in [45, 30]], [{"time": START - timedelta(minutes=15), "device_id": "P1", "monitored_value": 30., "source_hash": "15"}], ] async def run(): first = await service.create(create, monitoring_interval_seconds=900) await service.command(first.replay_id, ScadaReplayCommand( request_id=uuid4(), expected_revision=1, action="step")) with pytest.raises(BaselineBudgetExceeded): await service.process_pending_step(first.replay_id) result = snapshot(repo.document) assert result.status == "failed" and not result.step_pending assert "BASELINE_BUDGET_EXCEEDED" in result.error and "历史基线" in result.error and "重试" in result.error assert "baseline" not in repo.document and not repo.document.get("warmed") assert result.processed_count == result.anomaly_count == result.monitoring_count == 0 assert result.virtual_now == START assert set(repo.batches) == ({-1} if budget == "rows" else set()) assert all(batch["stage"] == "clean_prepared" for batch in repo.batches.values()) asyncio.run(run()) cleaning.assert_not_awaited() service._create_investigations.assert_not_awaited() assert query.await_count == (2 if budget == "rows" else 1) if budget == "rows": assert query.await_args_list[1].args[5] == 1 def test_failed_baseline_publication_rolls_back_visibility_and_initialized_marker(monkeypatch): repo, service, query, _, create = setup(monkeypatch) query.side_effect = [[{"time": START - timedelta(minutes=15), "device_id": "P1", "monitored_value": 30., "source_hash": "a"}], []] async def run(): await service.create(create, monitoring_interval_seconds=900) monkeypatch.setattr(repo, "save", AsyncMock(side_effect=RuntimeError("publish unavailable"))) with pytest.raises(RuntimeError, match="publish unavailable"): await service._freeze_baseline(await repo.get(None), create) assert repo.batches[-1]["stage"] == "clean_prepared" assert "baseline" not in repo.document and not repo.document.get("warmed") assert await repo.monitoring_outputs(None, ["P1"], START - timedelta(days=15), START, 1) == [] asyncio.run(run()) def test_baseline_already_published_is_not_refrozen_or_replaced(monkeypatch): repo, service, query, _, create = setup(monkeypatch) async def run(): await service.create(create, monitoring_interval_seconds=900) repo.document.update(warmed=True, baseline={"row_count": 1}) repo.batches[-1] = {"sequence": -1, "stage": "detected", "kind": "baseline", "outputs": []} before = deepcopy((repo.document, repo.batches)) await service._freeze_baseline(await repo.get(None), create) assert (repo.document, repo.batches) == before asyncio.run(run()) query.assert_not_awaited() def test_first_step_uses_frozen_fifteen_days_without_counting_historical_events(monkeypatch): repo, service, query, cleaning, _ = setup(monkeypatch) pressure_ids, flow_id = [f"P{i}" for i in range(5)], "F1" device_ids = pressure_ids + [flow_id] config = ScadaReplayCreate( request_id=uuid4(), start_time=START, end_time=START + timedelta(hours=1), device_ids=device_ids, rules=[{"rule_id": "pressure", "kind": "relative_change", "device_ids": pressure_ids, "direction": "decrease", "threshold": 20, "unit": "m"}, {"rule_id": "flow", "kind": "relative_change", "device_ids": [flow_id], "direction": "increase", "threshold": 20, "unit": "m3/h"}], ) monkeypatch.setattr(ScadaInfoRepository, "get_scadas", AsyncMock(return_value=[{ "device_id": device, "device_type": "pressure" if device in pressure_ids else "pipe_flow", "measurement_unit": "m" if device in pressure_ids else "m3/h", "transmission_frequency": "0:15:00", } for device in device_ids])) source = [{"time": START + timedelta(minutes=15 * i), "device_id": device, "monitored_value": (70. if device in pressure_ids else 150.) if i == 1 else 100., "source_hash": f"{i}-{device}"} for i in range(-15 * 96, 2) for device in sorted(device_ids)] async def raw_page(_connection, _devices, start, end, through, limit, after=None, **options): rows = [row for row in source if start <= row["time"] < end and row["time"] <= through and (after is None or (row["time"], row["device_id"]) > after)] if options.get("frame") and rows: rows = [row for row in rows if row["time"] == rows[0]["time"]] return rows[:limit] query.side_effect = raw_page def detection(**kwargs): target = kwargs["observed_pressure_data"].iloc[-1].min() return {"rows": [{"Score": 0.02}] * 14 + [{"Score": -0.2 if target < 90 else 0.02}], "summary": {}} monkeypatch.setattr("app.services.scada_replay_monitoring.run_burst_detection", detection) monkeypatch.setattr(service, "_create_monitoring_task", AsyncMock()) async def run(): first = await service.create(config, monitoring_interval_seconds=900, monitoring_config=ScadaMonitoringSettings( pressure_device_ids=pressure_ids, outlet_flow_device_ids=[flow_id])) await service.command(first.replay_id, ScadaReplayCommand( request_id=uuid4(), expected_revision=1, action="step")) result = await service.process_pending_step(first.replay_id) assert result.last_monitoring.status == "suspected_burst" assert result.last_monitoring.weighted["weights"]["outlet_flow"] == 0.1 assert repo.document["monitoring_calibration"]["status"] == "ready" assert result.processed_count == 12 and result.monitoring_count == 1 assert repo.document["baseline"]["row_count"] == 8640 assert result.virtual_now == START + timedelta(minutes=15) assert all(not batch["events"] for key, batch in repo.batches.items() if key < 0) asyncio.run(run()) assert cleaning.await_count == 2 assert all(row["time"] >= START.isoformat() for call in cleaning.await_args_list for row in call.args[1]) def test_step_command_is_revisioned_idempotent_and_does_not_process_until_requested(monkeypatch): repo, service, query, cleaning, create = setup(monkeypatch) async def run(): first = await service.create(create, monitoring_interval_seconds=900) restored = await service.create(create, monitoring_interval_seconds=300) assert restored.replay_id == first.replay_id assert restored.monitoring_interval_seconds == 900 command = ScadaReplayCommand(request_id=uuid4(), expected_revision=1, action="step") played = await service.command(first.replay_id, command) assert played.status == "paused" and played.step_pending and played.revision == 2 assert (await service.command(first.replay_id, command)).revision == 2 with pytest.raises(HTTPException) as conflict: await service.command(first.replay_id, command.model_copy(update={"request_id": uuid4()})) assert conflict.value.status_code == 409 asyncio.run(run()) query.assert_not_awaited(); cleaning.assert_not_awaited() @pytest.mark.parametrize("minutes", [0, 10]) def test_monitoring_stops_exactly_at_selected_time_and_reuses_same_period(monkeypatch, minutes): repo, service, query, cleaning, create = setup(monkeypatch) target = START + timedelta(minutes=minutes) async def run(): initial = await service.create(create, monitoring_interval_seconds=900) pending = await service.command(initial.replay_id, ScadaReplayCommand( request_id=uuid4(), expected_revision=1, action="monitor", at_time=target)) assert pending.step_pending assert datetime.fromisoformat(repo.document["step_target"]) == target result = await service.process_pending_step(initial.replay_id) assert result.virtual_now == target assert result.last_monitoring.period_end == target assert result.last_monitoring.knowledge_cutoff == target assert result.monitoring_count == 1 reads, writes, batches = query.await_count, cleaning.await_count, len(repo.batches) repeated = await service.command(initial.replay_id, ScadaReplayCommand( request_id=uuid4(), expected_revision=result.revision, action="monitor", at_time=target)) assert not repeated.step_pending assert repeated.last_monitoring == result.last_monitoring assert repeated.monitoring_count == 1 assert (query.await_count, cleaning.await_count, len(repo.batches)) == (reads, writes, batches) await service.command(initial.replay_id, ScadaReplayCommand( request_id=uuid4(), expected_revision=repeated.revision, action="step")) assert datetime.fromisoformat(repo.document["step_target"]) == START + timedelta(minutes=15) asyncio.run(run()) assert all(call.args[4] <= target for call in query.await_args_list) @pytest.mark.parametrize("minutes,status_code", [(-5, 422), (65, 422), (15, 409)]) def test_selected_monitoring_rejects_out_of_range_or_backward_targets_without_processing(monkeypatch, minutes, status_code): repo, service, query, cleaning, create = setup(monkeypatch) async def run(): initial = await service.create(create, monitoring_interval_seconds=900) repo.document["virtual_now"] = (START + timedelta(minutes=30)).isoformat() before = deepcopy(repo.document) with pytest.raises(HTTPException) as failure: await service.command(initial.replay_id, ScadaReplayCommand( request_id=uuid4(), expected_revision=1, action="monitor", at_time=START + timedelta(minutes=minutes))) assert failure.value.status_code == status_code assert repo.document == before asyncio.run(run()) query.assert_not_awaited() cleaning.assert_not_awaited() def test_selected_monitoring_cannot_replace_a_pending_period(monkeypatch): repo, service, query, cleaning, create = setup(monkeypatch) async def run(): initial = await service.create(create, monitoring_interval_seconds=900) await service.command(initial.replay_id, ScadaReplayCommand(request_id=uuid4(), expected_revision=1, action="step")) before = deepcopy(repo.document) with pytest.raises(HTTPException, match="当前时间步正在处理"): await service.command(initial.replay_id, ScadaReplayCommand( request_id=uuid4(), expected_revision=2, action="monitor", at_time=START)) assert repo.document == before asyncio.run(run()) query.assert_not_awaited() cleaning.assert_not_awaited() @pytest.mark.parametrize("action,at_time", [("monitor", None), ("monitor", "2026-07-16T08:00:00"), ("step", "2026-07-16T00:00:00Z")]) def test_selected_monitoring_requires_an_explicit_zoned_time_only_for_monitor(action, at_time): with pytest.raises(ValueError): ScadaReplayCommand(request_id=uuid4(), expected_revision=1, action=action, at_time=at_time) def test_one_raw_read_then_correction_then_detection_without_reread(monkeypatch): repo, service, query, cleaning, create = setup(monkeypatch) async def run(): first = await service.create(create, monitoring_interval_seconds=900) repo.document["warmed"] = True await service.command(first.replay_id, ScadaReplayCommand(request_id=uuid4(), expected_revision=1, action="step")) result = await service.process_pending_step(first.replay_id) assert result.processed_count == 1 assert result.anomaly_count == 1 assert repo.batches[1]["stage"] == "detected" asyncio.run(run()) assert query.await_count == 2 assert cleaning.await_count == 2 assert cleaning.await_args_list[0].args[1][0]["cleaned_value"] == 20 def test_detection_failure_resumes_cleaned_batch_without_duplicate_reads_or_writes(monkeypatch): repo, service, query, cleaning, create = setup(monkeypatch) publish = AsyncMock(side_effect=[RuntimeError("Agent storage unavailable"), None, None]) monkeypatch.setattr(service, "_create_investigations", publish) async def run(): first = await service.create(create, monitoring_interval_seconds=900) repo.document["warmed"] = True await service.command(first.replay_id, ScadaReplayCommand(request_id=uuid4(), expected_revision=1, action="step")) with pytest.raises(RuntimeError): await service.process_pending_step(first.replay_id) assert repo.document["status"] == "failed" assert repo.batches[1]["stage"] == "cleaned" assert repo.document["processed_count"] == 0 await service.command(first.replay_id, ScadaReplayCommand(request_id=uuid4(), expected_revision=2, action="retry")) result = await service.process_pending_step(first.replay_id) assert result.processed_count == 1 and result.anomaly_count == 1 asyncio.run(run()) assert query.await_count == 2 assert cleaning.await_count == 2 def test_cleaning_failure_never_runs_detection_and_preserves_prepared_result(monkeypatch): repo, service, query, cleaning, create = setup(monkeypatch) cleaning.side_effect = RuntimeError("Timescale unavailable") async def run(): first = await service.create(create, monitoring_interval_seconds=900) repo.document["warmed"] = True await service.command(first.replay_id, ScadaReplayCommand(request_id=uuid4(), expected_revision=1, action="step")) with pytest.raises(RuntimeError): await service.process_pending_step(first.replay_id) assert repo.batches[1]["stage"] == "clean_prepared" assert repo.document["processed_count"] == 0 asyncio.run(run()) service._create_investigations.assert_not_awaited() def test_creation_requires_explicit_project_monitoring_interval(monkeypatch): repo, service, query, cleaning, create = setup(monkeypatch) async def run(): with pytest.raises(HTTPException) as missing: await service.create(create, monitoring_interval_seconds=None) assert missing.value.status_code == 422 assert repo.document is None asyncio.run(run()) query.assert_not_awaited() cleaning.assert_not_awaited() def test_step_processes_every_observation_within_one_monitoring_interval(monkeypatch): repo, service, query, cleaning, create = setup(monkeypatch) rows = [{"time": START + timedelta(minutes=minute), "device_id": "P1", "monitored_value": 20., "source_hash": f"h{minute}"} for minute in (0, 5, 10, 15, 20)] async def raw_page(_connection, _devices, _start, _end, through, _limit, after=None, **_): return next(([row] for row in rows if row["time"] <= through and (after is None or (row["time"], row["device_id"]) > after)), []) query.side_effect = raw_page async def run(): first = await service.create(create, monitoring_interval_seconds=900) repo.document["warmed"] = True stepping = await service.command(first.replay_id, ScadaReplayCommand( request_id=uuid4(), expected_revision=1, action="step" )) assert stepping.status == "paused" and stepping.step_pending with pytest.raises(HTTPException): await service.command(first.replay_id, ScadaReplayCommand( request_id=uuid4(), expected_revision=2, action="step" )) result = await service.process_pending_step(first.replay_id) assert result.processed_count == 4 assert result.virtual_now == START + timedelta(minutes=15) assert result.status == "paused" assert not result.step_pending assert result.processed_until == START + timedelta(minutes=15) assert result.monitoring_count == 1 assert result.last_monitoring.status == "insufficient_data" assert "step_target" not in repo.document assert all(call.args[4] == START + timedelta(minutes=15) for call in query.await_args_list) asyncio.run(run()) assert cleaning.await_count == 4 def test_partial_final_step_checks_empty_interval_and_completes_without_fabricated_rows(monkeypatch): repo, service, query, cleaning, create = setup(monkeypatch) create = create.model_copy(update={"end_time": START + timedelta(minutes=22)}) query.side_effect = None query.return_value = [] async def run(): first = await service.create(create, monitoring_interval_seconds=900) repo.document.update(warmed=True, virtual_now=(START + timedelta(minutes=15)).isoformat()) await service.command(first.replay_id, ScadaReplayCommand( request_id=uuid4(), expected_revision=1, action="step" )) result = await service.process_pending_step(first.replay_id) assert result.virtual_now == create.end_time assert result.processed_until == create.end_time assert result.status == "completed" and not result.step_pending assert result.processed_count == 0 asyncio.run(run()) assert query.await_args.args[4] == create.end_time @pytest.mark.parametrize("weighted", [False, True]) def test_suspected_monitoring_creates_burst_report_with_frozen_replay_context(monkeypatch, weighted): repo, service, *_ = setup(monkeypatch) create_task = AsyncMock(return_value=SimpleNamespace(task_id=uuid4())) monkeypatch.setattr("app.services.scada_replay.permissions_for_context", lambda _: {"task.write"}) monkeypatch.setattr("app.infra.db.postgresql.operational_tasks.OperationalTaskRepository", lambda *_: object()) monkeypatch.setattr( "app.services.operational_tasks.OperationalTaskService", lambda *_, **_kwargs: SimpleNamespace(create=create_task), ) monitoring = { "status": "suspected_burst", "period_end": (START + timedelta(minutes=15)).isoformat(), "knowledge_cutoff": (START + timedelta(minutes=15)).isoformat(), "rule_version": "burst-monitor-v1", "pressure": {"hit_device_ids": ["P1", "P2"]}, "outlet_flow": {"hit_device_ids": ["F1"]}, "burst_detection": {"target_score": -0.05}, "task_id": None, } if weighted: monitoring["burst_detection"]["score_threshold"] = -0.0843 monitoring["weighted"] = { "score": 0.9, "threshold": 0.8, "weights": {"burst_detection": 0.5, "pressure": 0.4, "outlet_flow": 0.1}, } document = {"replay_id": str(uuid4())} asyncio.run(service._create_monitoring_task(document, monitoring, {"sequence": 8})) command = create_task.await_args.args[0] assert command.workflow == "scada_anomaly" assert command.source.replay_context.batch_sequence == 8 assert command.source.replay_context.knowledge_cutoff == START + timedelta(minutes=15) assert monitoring["task_id"] == str(create_task.return_value.task_id) assert document["monitoring_incident"]["alert"]["condition_id"] == f"operational-task:{monitoring['task_id']}" assert document["monitoring_incident"]["alert"]["description"] == "已启动调查，详细诊断见调查报告。" if weighted: assert "-0.0843" in command.summary assert "综合评分 0.90" in command.summary assert "流量 10%" in command.summary def test_monitoring_summary_counts_signals_without_copying_reading_details(monkeypatch): _, service, *_ = setup(monkeypatch) create_task = AsyncMock(return_value=SimpleNamespace(task_id=uuid4())) monkeypatch.setattr("app.infra.db.postgresql.operational_tasks.OperationalTaskRepository", lambda *_: object()) monkeypatch.setattr("app.services.operational_tasks.OperationalTaskService", lambda *_, **_kwargs: SimpleNamespace(create=create_task)) monitoring = incident_monitoring(0) monitoring["anomaly_signals"] = [ {"device_ids": [f"P{i}"], "detail": f"P{i}=28.14402136077437 m，超出冻结正常同时段范围 [28.3253, 28.8685] m"} for i in range(100) ] asyncio.run(service._create_monitoring_task({"replay_id": str(uuid4())}, monitoring, {"sequence": 1})) summary = create_task.await_args.args[0].summary assert len(summary) <= 300 assert "100" in summary assert "28.14402136077437" not in summary assert monitoring["anomaly_signals"][0]["detail"].startswith("P0=28.14402136077437") def test_continuing_burst_reuses_task_across_persisted_periods(monkeypatch): repo, service, *_ = setup(monkeypatch) create_task = AsyncMock(return_value=SimpleNamespace(task_id=uuid4())) advance_task = AsyncMock() monkeypatch.setattr("app.services.scada_replay.permissions_for_context", lambda _: {"task.write"}) monkeypatch.setattr("app.infra.db.postgresql.operational_tasks.OperationalTaskRepository", lambda *_: object()) monkeypatch.setattr("app.services.operational_tasks.OperationalTaskService", lambda *_, **_kwargs: SimpleNamespace(create=create_task, advance_replay_investigation=advance_task)) document = {"replay_id": str(uuid4())} async def run(): nonlocal document for step in range(3): monitoring = incident_monitoring(step, status="attention" if step == 0 else "suspected_burst") await service._create_monitoring_task(document, monitoring, {"sequence": step + 1}) document = deepcopy(document) # Simulate loading persisted state after restart. assert monitoring["task_id"] == str(create_task.return_value.task_id) assert create_task.await_count == 1 service._publish_monitoring_alert.assert_awaited_once() assert document["monitoring_incident"]["pressure_anchors"]["P1"]["baseline"] == 100. advance_task.assert_awaited_once() assert advance_task.await_args.args[0] == create_task.return_value.task_id assert advance_task.await_args.args[1]["knowledge_cutoff"] == (START + timedelta(minutes=30)).isoformat() asyncio.run(run()) def incident_monitoring(step, *, status="suspected_burst", value=60.): return { "monitoring_id": str(uuid4()), "status": status, "period_start": (START + timedelta(minutes=15 * step)).isoformat(), "period_end": (START + timedelta(minutes=15 * (step + 1))).isoformat(), "knowledge_cutoff": (START + timedelta(minutes=15 * (step + 1))).isoformat(), "rule_version": "burst-monitor-v2", "detail": "监测结论", "task_id": None, "pressure": {"status": "ready", "hit": status == "suspected_burst", "hit_device_ids": ["P1"], "devices": [{"device_id": "P1", "baseline": 100., "value": value, "threshold_percent": 10., "hit": status == "suspected_burst", "observed_at": (START + timedelta(minutes=15 * (step + 1))).isoformat()}]}, "outlet_flow": {"status": "ready", "hit": False, "hit_device_ids": []}, "burst_detection": {"status": "ready", "hit": status == "suspected_burst", "target_score": -0.2}, } def test_incident_recovers_once_then_allows_new_accident(monkeypatch): _, service, *_ = setup(monkeypatch) create_task = AsyncMock(side_effect=[SimpleNamespace(task_id=uuid4()), SimpleNamespace(task_id=uuid4())]) monkeypatch.setattr("app.services.scada_replay.permissions_for_context", lambda _: {"task.write"}) monkeypatch.setattr("app.infra.db.postgresql.operational_tasks.OperationalTaskRepository", lambda *_: object()) monkeypatch.setattr("app.services.operational_tasks.OperationalTaskService", lambda *_, **_kwargs: SimpleNamespace(create=create_task)) document = {"replay_id": str(uuid4())} async def run(): first = incident_monitoring(0) await service._create_monitoring_task(document, first, {"sequence": 1}) for step in (1, 2): result = incident_monitoring(step, status="normal", value=100.) await service._create_monitoring_task(document, result, {"sequence": step + 1}) assert result["incident"]["action"] == "resolved" assert result["task_id"] == first["task_id"] recurrence = incident_monitoring(3) await service._create_monitoring_task(document, recurrence, {"sequence": 4}) assert recurrence["incident"]["incident_id"] != first["incident"]["incident_id"] assert recurrence["task_id"] != first["task_id"] assert create_task.await_count == 2 assert [call.args[2] for call in service._publish_monitoring_alert.await_args_list] == ["raised", "resolved", "raised"] commands = [call.args[0] for call in create_task.await_args_list] assert commands[0].source.replay_context.knowledge_cutoff == START + timedelta(minutes=15) asyncio.run(run()) @pytest.mark.parametrize("interruption", ["insufficient_data", "attention", "normal_low_pressure", "missing_sensor", "stale_reading", "time_gap"]) def test_incident_needs_two_fresh_complete_pressure_recoveries(interruption): from app.services.scada_monitoring_incidents import update_incident document = {"replay_id": str(uuid4())} incident, _ = update_incident(document, incident_monitoring(0), 1) incident["task_id"] = str(uuid4()) update_incident(document, incident_monitoring(1, status="normal", value=100.), 2) interrupted = incident_monitoring(2, status="normal", value=100.) if interruption in {"insufficient_data", "attention"}: interrupted["status"] = interruption elif interruption == "normal_low_pressure": interrupted["pressure"]["devices"][0]["value"] = 60. elif interruption == "missing_sensor": interrupted["pressure"]["devices"] = [] elif interruption == "stale_reading": interrupted["pressure"]["devices"][0]["observed_at"] = incident["last_observed_at"] else: interrupted["period_start"] = START.isoformat() incident, action = update_incident(document, interrupted, 3) assert action == "ongoing" assert incident["recovery_count"] == (1 if interruption == "time_gap" else 0) if interruption != "time_gap": _, action = update_incident(document, incident_monitoring(3, status="normal", value=100.), 4) assert action == "ongoing" def test_same_period_retry_does_not_count_as_second_recovery(): from app.services.scada_monitoring_incidents import update_incident document = {"replay_id": str(uuid4())} update_incident(document, incident_monitoring(0), 1) recovery = incident_monitoring(1, status="normal", value=100.) update_incident(document, recovery, 2) incident, action = update_incident(deepcopy(document), recovery, 2) assert action == "ongoing" and incident["recovery_count"] == 1 def test_older_replay_adopts_existing_task_without_creating_another(monkeypatch): _, service, *_ = setup(monkeypatch) previous = incident_monitoring(0) previous["task_id"] = str(uuid4()) document = {"replay_id": str(uuid4()), "last_monitoring": previous} current = incident_monitoring(1) asyncio.run(service._create_monitoring_task(document, current, {"sequence": 2})) assert current["task_id"] == previous["task_id"] assert current["incident"]["action"] == "ongoing" service._publish_monitoring_alert.assert_not_awaited() def test_combined_monitoring_alert_uses_same_facts_for_recovery(monkeypatch): repo, service, *_ = setup(monkeypatch) push = AsyncMock() monkeypatch.setattr("app.infra.db.postgresql.task_alerts.TaskAlertRepository", lambda *_: object()) monkeypatch.setattr("app.services.task_alerts.TaskAlertService", lambda *_args, **_kwargs: SimpleNamespace(push=push)) incident = {"incident_id": str(uuid4()), "alert": { "alert_id": "test-accident", "producer_id": "scada-replay:demo", "occurred_at": START.isoformat(), "condition_id": None, "source": {"kind": "scada_replay", "replay_id": "demo", "rule_id": "burst_monitor", "batch_sequence": 1}, "title": "SCADA 监测疑似爆管", "description": "同一事故", "severity": "warning", }} async def run(): for change in ("raised", "resolved"): await ScadaReplayService._publish_monitoring_alert(service, {"replay_id": "demo"}, incident, change) asyncio.run(run()) first, second = [call.args[0].model_dump(mode="json") for call in push.await_args_list] assert first.pop("event_id") != second.pop("event_id") assert first.pop("event_type") == "raised" assert second.pop("event_type") == "resolved" assert first == second def test_combined_monitoring_keeps_rule_evidence_and_quality_notifications(monkeypatch): _, service, *_ = setup(monkeypatch) push = AsyncMock() monkeypatch.setattr("app.services.scada_replay.permissions_for_context", lambda _: {"task.write"}) monkeypatch.setattr("app.infra.db.postgresql.task_alerts.TaskAlertRepository", lambda *_: object()) monkeypatch.setattr("app.services.task_alerts.TaskAlertService", lambda *_args, **_kwargs: SimpleNamespace(push=push)) monkeypatch.setattr("app.infra.db.postgresql.operational_tasks.OperationalTaskRepository", lambda *_: object()) monkeypatch.setattr("app.services.operational_tasks.OperationalTaskService", lambda *_, **_kwargs: SimpleNamespace(create=AsyncMock())) def event(rule, category=None): return {"event_id": str(uuid4()), "rule_id": rule, "change": "raised", "occurred_at": START.isoformat(), "detail": "检测信号", "category": category, "task_id": None} pressure, water, quality = event("p"), event("water"), event("quality-P1", "data_quality") document = {"replay_id": str(uuid4()), "monitoring_config": {"decision_policy": "weighted"}, "config": {"rules": [{"rule_id": "p", "kind": "relative_change"}, {"rule_id": "water", "kind": "bounds"}]}, "rule_states": {item["rule_id"]: {"event": deepcopy(item)} for item in (pressure, water, quality)}} async def run(): await ScadaReplayService._create_investigations(service, document, [pressure, water, quality], {"sequence": 1}, create_tasks=False) assert push.await_count == 1 # Measurement signals share diagnosis; telemetry quality remains independent. assert document["rule_states"]["p"]["event"]["notification_suppressed"] assert document["rule_states"]["water"]["event"]["notification_suppressed"] resolved = {**document["rule_states"]["p"]["event"], "change": "resolved"} await ScadaReplayService._create_investigations(service, document, [resolved], {"sequence": 2}, create_tasks=False) assert push.await_count == 1 resolved.pop("notification_suppressed") # An old alert raised before this change still resolves. await ScadaReplayService._create_investigations(service, document, [resolved], {"sequence": 2}, create_tasks=False) assert push.await_count == 2 asyncio.run(run()) def test_failed_incident_publication_rolls_back_state_and_retries_same_task_request(monkeypatch): repo, service, _, _, create = setup(monkeypatch) create_task = AsyncMock(return_value=SimpleNamespace(task_id=uuid4())) monkeypatch.setattr("app.services.scada_replay.permissions_for_context", lambda _: {"task.write"}) monkeypatch.setattr("app.infra.db.postgresql.operational_tasks.OperationalTaskRepository", lambda *_: object()) monkeypatch.setattr("app.services.operational_tasks.OperationalTaskService", lambda *_, **_kwargs: SimpleNamespace(create=create_task)) def evaluate(_doc, _rows, *, period_start, period_end, batch_sequence): return {**incident_monitoring(0), "period_start": period_start.isoformat(), "period_end": period_end.isoformat(), "knowledge_cutoff": period_end.isoformat(), "batch_sequence": batch_sequence} monkeypatch.setattr("app.services.scada_replay.evaluate_monitoring_period", evaluate) service._publish_monitoring_alert.side_effect = [RuntimeError("alert unavailable"), None] async def run(): first = await service.create(create, monitoring_interval_seconds=900) repo.document["warmed"] = True await service.command(first.replay_id, ScadaReplayCommand(request_id=uuid4(), expected_revision=1, action="step")) with pytest.raises(RuntimeError, match="alert unavailable"): await service.process_pending_step(first.replay_id) assert repo.document["status"] == "failed" assert "monitoring_incident" not in repo.document assert repo.document["monitoring_count"] == 0 await service.command(first.replay_id, ScadaReplayCommand(request_id=uuid4(), expected_revision=2, action="retry")) result = await service.process_pending_step(first.replay_id) assert result.last_monitoring.incident.action == "raised" assert result.monitoring_count == 1 first_command, retried_command = [call.args[0] for call in create_task.await_args_list] assert first_command == retried_command assert result.last_monitoring.task_id == create_task.return_value.task_id asyncio.run(run()) @pytest.mark.parametrize("indicator", ["pressure", "outlet_flow", "burst_detection", "quality", "partial_data"]) def test_any_single_measurement_anomaly_enters_diagnosis_below_burst_threshold(monkeypatch, indicator): _, service, *_ = setup(monkeypatch) create_task = AsyncMock(return_value=SimpleNamespace(task_id=uuid4())) monkeypatch.setattr("app.infra.db.postgresql.operational_tasks.OperationalTaskRepository", lambda *_: object()) monkeypatch.setattr("app.services.operational_tasks.OperationalTaskService", lambda *_, **_kwargs: SimpleNamespace(create=create_task)) monitoring = incident_monitoring(0, status="attention") for key in ("pressure", "outlet_flow", "burst_detection"): monitoring[key]["hit"] = key == indicator monitoring["weighted"] = {"score": 0.1, "threshold": 0.8, "weights": {"burst_detection": 0.5, "pressure": 0.4, "outlet_flow": 0.1}} if indicator == "quality": monitoring["status"] = "insufficient_data" monitoring["anomaly_signals"] = [{"event_id": str(uuid4()), "rule_id": "quality_bounds", "device_ids": ["Q1"], "detail": "水质测点超出已配置上限"}] monitoring["burst_detection"] = {"status": "not_run", "hit": False} if indicator == "partial_data": monitoring["status"] = "insufficient_data" monitoring["pressure"]["hit"] = True document = {"replay_id": str(uuid4())} asyncio.run(service._create_monitoring_task(document, monitoring, {"sequence": 1})) command = create_task.await_args.args[0] assert command.workflow == "scada_anomaly" assert command.source.event_kind == "scada_anomaly" assert "任一指标异常即启动调查" in command.summary assert "爆管嫌疑门槛 0.80" in command.summary assert monitoring["task_id"] == str(create_task.return_value.task_id) assert monitoring["incident"]["action"] == "raised" def test_flow_only_incident_does_not_recover_while_absolute_flow_remains_high(): from app.services.scada_monitoring_incidents import update_incident document = {"replay_id": str(uuid4())} first = incident_monitoring(0, status="attention", value=100.) first["outlet_flow"] = {"status": "ready", "hit": True, "devices": [ {"device_id": "F1", "hit": True, "baseline": 100., "threshold_percent": 10., "value": 150.} ]} incident, _ = update_incident(document, first, 1) incident["task_id"] = str(uuid4()) for step in (1, 2): normal = incident_monitoring(step, status="normal", value=100.) normal["outlet_flow"] = {"status": "ready", "hit": False, "devices": [ {"device_id": "F1", "value": 150., "observed_at": normal["period_end"]} ]} _, action = update_incident(document, normal, step + 1) assert action == "ongoing" for step in (3, 4): normal = incident_monitoring(step, status="normal", value=100.) normal["outlet_flow"] = {"status": "ready", "hit": False, "devices": [ {"device_id": "F1", "value": 100., "observed_at": normal["period_end"]} ]} _, action = update_incident(document, normal, step + 1) assert action == "resolved" def test_transient_measurement_anomaly_still_enters_diagnosis_at_period_end(monkeypatch): repo, service, query, _, create = setup(monkeypatch) readings = [(START, 20.), (START + timedelta(minutes=5), 5.), (START + timedelta(minutes=10), 5.)] async def raw_page(_conn, _devices, _start, _end, through, _limit, after=None, **_): return [{"time": at, "device_id": "P1", "monitored_value": value, "source_hash": str(index)} for index, (at, value) in enumerate(readings) if at <= through and (after is None or at > after[0])][:1] query.side_effect = raw_page def evaluate(_doc, _rows, *, period_start, period_end, batch_sequence): return {**incident_monitoring(0, status="normal", value=5.), "period_start": period_start.isoformat(), "period_end": period_end.isoformat(), "knowledge_cutoff": period_end.isoformat(), "batch_sequence": batch_sequence} monkeypatch.setattr("app.services.scada_replay.evaluate_monitoring_period", evaluate) async def run(): initial = await service.create(create, monitoring_interval_seconds=900) repo.document["warmed"] = True await service.command(initial.replay_id, ScadaReplayCommand(request_id=uuid4(), expected_revision=1, action="step")) result = await service.process_pending_step(initial.replay_id) assert result.active_events == [] assert result.last_monitoring.status == "attention" assert result.last_monitoring.anomaly_signals[0]["rule_id"] == "p" assert result.last_monitoring.task_id is not None assert result.last_monitoring.incident.action == "raised" assert "step_anomaly_signals" not in repo.document asyncio.run(run()) @pytest.mark.parametrize("action,status", [("step", "paused"), ("retry", "failed")]) def test_warmed_weighted_replay_with_old_calibration_requires_rebuild(monkeypatch, action, status): repo, service, _, _, create = setup(monkeypatch) async def run(): initial = await service.create(create, monitoring_interval_seconds=900, monitoring_config=ScadaMonitoringSettings()) repo.document.update(warmed=True, status=status, monitoring_calibration={"version": "scada-monitor-calibration@1"}) with pytest.raises(HTTPException) as error: await service.command(initial.replay_id, ScadaReplayCommand( request_id=uuid4(), expected_revision=1, action=action)) assert error.value.status_code == 409 assert "新建" in error.value.detail asyncio.run(run()) def test_old_active_percentage_event_does_not_enter_new_period_diagnosis(monkeypatch): from app.services.scada_normal_profile import CALIBRATION_VERSION, DIAGNOSTIC_POLICY repo, service, _, _, _ = setup(monkeypatch) config = ScadaReplayCreate(request_id=uuid4(), start_time=START, end_time=START + timedelta(hours=1), device_ids=["P1"], rules=[{"rule_id": "p", "device_ids": ["P1"], "kind": "relative_change", "direction": "both", "threshold": 30, "unit": "m"}]) monkeypatch.setattr(service, "_create_monitoring_task", AsyncMock()) def evaluate(_doc, _rows, *, period_start, period_end, batch_sequence): return {**incident_monitoring(0, status="normal", value=20.), "period_start": period_start.isoformat(), "period_end": period_end.isoformat(), "knowledge_cutoff": period_end.isoformat(), "batch_sequence": batch_sequence} monkeypatch.setattr("app.services.scada_replay.evaluate_monitoring_period", evaluate) async def run(): initial = await service.create(config, monitoring_interval_seconds=900, monitoring_config=ScadaMonitoringSettings()) old = {"event_id": str(uuid4()), "rule_id": "p", "device_ids": ["P1"], "status": "active", "occurred_at": START.isoformat(), "last_observed_at": START.isoformat(), "detail": "旧30%事件", "task_id": None} repo.document.update(warmed=True, rule_states={"p": {"event": old, "hits": 1, "recoveries": 0}}, monitoring_calibration={"version": CALIBRATION_VERSION, "diagnostic_policy": DIAGNOSTIC_POLICY, "knowledge_cutoff": START.isoformat(), "diagnostic": {"P1": { "phase_origin": START.isoformat(), "interval_seconds": 900, "lower": [10.], "upper": [30.]}}}) await service.command(initial.replay_id, ScadaReplayCommand( request_id=uuid4(), expected_revision=1, action="step")) result = await service.process_pending_step(initial.replay_id) assert result.last_monitoring.status == "normal" assert result.last_monitoring.anomaly_signals == [] assert repo.document["rule_states"]["p"]["event"] == old asyncio.run(run())