diff --git a/.env.example b/.env.example index 48f7907..f80fa4e 100644 --- a/.env.example +++ b/.env.example @@ -6,6 +6,8 @@ # 基础运行 PROJECT_NAME="TJWater Server" ENVIRONMENT="production" +# 任务领取范围；留空时沿用 ENVIRONMENT。历史任务需先执行 034 迁移。 +OPERATIONAL_TASK_EXECUTION_SCOPE="" API_V1_STR="/api/v1" NETWORK_NAME="tjwater" diff --git a/app/api/v1/endpoints/operational_tasks.py b/app/api/v1/endpoints/operational_tasks.py index 9b3c701..13c7718 100644 --- a/app/api/v1/endpoints/operational_tasks.py +++ b/app/api/v1/endpoints/operational_tasks.py @@ -9,6 +9,7 @@ from fastapi import APIRouter, Depends, Header, HTTPException, Query, status from psycopg import AsyncConnection from app.auth.metadata_dependencies import get_metadata_repository +from app.core.config import settings from app.auth.permissions import ( OPERATIONAL_TASK_EXECUTE, OPERATIONAL_TASK_EXTERNAL_SYNC, @@ -122,6 +123,7 @@ def _service(ctx: ProjectContext, conn: AsyncConnection) -> OperationalTaskServi OperationalTaskRepository(conn, ctx.project_id), ctx, _task_permissions(ctx), + execution_scope=settings.operational_task_execution_scope, ) diff --git a/app/api/v1/endpoints/workbench_events.py b/app/api/v1/endpoints/workbench_events.py index 71df44d..0fda6f4 100644 --- a/app/api/v1/endpoints/workbench_events.py +++ b/app/api/v1/endpoints/workbench_events.py @@ -36,6 +36,7 @@ def _service(ctx: ProjectContext, conn: AsyncConnection) -> OperationalTaskServi OperationalTaskRepository(conn, ctx.project_id), ctx, permissions_for_context(ctx), + execution_scope=settings.operational_task_execution_scope, ) diff --git a/app/core/config.py b/app/core/config.py index bbed419..ec131ea 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -8,6 +8,12 @@ from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): PROJECT_NAME: str = "TJWater Server" ENVIRONMENT: str = "production" + OPERATIONAL_TASK_EXECUTION_SCOPE: str = "" + + @property + def operational_task_execution_scope(self) -> str: + return self.OPERATIONAL_TASK_EXECUTION_SCOPE.strip() or self.ENVIRONMENT + API_V1_STR: str = "/api/v1" NETWORK_NAME: str = "default_network" # 敏感配置加密密钥 (Fernet) diff --git a/app/domain/schemas/operational_task.py b/app/domain/schemas/operational_task.py index 8125b8e..60ad8ae 100644 --- a/app/domain/schemas/operational_task.py +++ b/app/domain/schemas/operational_task.py @@ -574,6 +574,7 @@ class OperationalTaskRun(StrictModel): class OperationalTaskSummary(StrictModel): + execution_scope: NonEmptyStr project_id: UUID task_id: UUID task_type: TaskType diff --git a/app/infra/db/postgresql/operational_tasks.py b/app/infra/db/postgresql/operational_tasks.py index b6c0545..407cb57 100644 --- a/app/infra/db/postgresql/operational_tasks.py +++ b/app/infra/db/postgresql/operational_tasks.py @@ -264,6 +264,7 @@ class OperationalTaskRepository: async def list_runs( self, *, status: str | None, limit: int, cursor_created_at=None, cursor_run_id=None, task_type: str | None = None, task_created_from: datetime | None = None, + execution_scope: str = "production", ) -> list[dict]: result = await self.connection.execute( """SELECT run_id,task_id,generation,execution_request_id,status,revision, @@ -282,18 +283,20 @@ class OperationalTaskRepository: AND t.task_id=operational_task_runs.task_id AND (%s::text IS NULL OR t.task_type=%s::text) AND (%s::timestamptz IS NULL OR t.created_at>=%s::timestamptz) + AND t.document->>'execution_scope'=%s ) ORDER BY created_at,run_id LIMIT %s""", ( self.project_id, status, status, cursor_created_at, cursor_created_at, cursor_run_id, - task_type, task_type, task_created_from, task_created_from, limit, + task_type, task_type, task_created_from, task_created_from, execution_scope, limit, ), ) return [dict(row) for row in await result.fetchall()] async def claim_expired_runs( self, limit: int, *, task_type: str | None = None, task_created_from: datetime | None = None, + execution_scope: str = "production", ) -> list[dict]: cursor = await self.connection.execute( """SELECT run_id,task_id,generation,execution_request_id,status,revision, @@ -310,9 +313,10 @@ class OperationalTaskRepository: AND t.task_id=operational_task_runs.task_id AND (%s::text IS NULL OR t.task_type=%s::text) AND (%s::timestamptz IS NULL OR t.created_at>=%s::timestamptz) + AND t.document->>'execution_scope'=%s ) ORDER BY lease_expires_at,run_id FOR UPDATE SKIP LOCKED LIMIT %s""", - (self.project_id, task_type, task_type, task_created_from, task_created_from, limit), + (self.project_id, task_type, task_type, task_created_from, task_created_from, execution_scope, limit), ) return [dict(row) for row in await cursor.fetchall()] @@ -711,15 +715,16 @@ class OperationalTaskRepository: row = await cursor.fetchone() return dict(row) if row else None - async def recovery_candidates(self, limit: int, created_from=None) -> list[dict]: + async def recovery_candidates(self, limit: int, created_from=None, execution_scope: str = "production") -> list[dict]: cursor = await self.connection.execute( """SELECT document FROM workbench.operational_tasks WHERE project_id=%s AND task_type='event_investigation' AND status NOT IN ('cancelled','failed') AND document#>>'{handling_context,phase}'='waiting_scada' + AND document->>'execution_scope'=%s AND (%s::timestamptz IS NULL OR created_at>=%s) ORDER BY COALESCE((document#>>'{handling_context,last_checked_at}')::timestamptz,created_at),task_id LIMIT %s""", - (self.project_id, created_from, created_from, limit), + (self.project_id, execution_scope, created_from, created_from, limit), ) return [row["document"] for row in await cursor.fetchall()] diff --git a/app/mcp/query_service.py b/app/mcp/query_service.py index 151fbcc..192c2c4 100644 --- a/app/mcp/query_service.py +++ b/app/mcp/query_service.py @@ -12,6 +12,7 @@ from datetime import datetime, timezone from uuid import uuid4 from fastapi import HTTPException from fastapi.encoders import jsonable_encoder +from app.core.config import settings from app.auth.permissions import permissions_for_context from app.infra.db.postgresql import read_queries as reads from app.infra.db.postgresql.spatial_queries import nearby_elements @@ -52,6 +53,7 @@ def task_service(scope: QueryScope) -> OperationalTaskService: OperationalTaskRepository(scope.connection, scope.project.project_id), scope.project, set(permissions_for_context(scope.project)), + execution_scope=settings.operational_task_execution_scope, ) diff --git a/app/services/operational_recovery.py b/app/services/operational_recovery.py index a658b72..1a07c5b 100644 --- a/app/services/operational_recovery.py +++ b/app/services/operational_recovery.py @@ -131,9 +131,11 @@ async def reconcile_recoveries(service: OperationalTaskService, timescale: Async from app.infra.db.timescaledb.repositories.scada import ScadaRepository async with service.repo.transaction(): - candidates = await service.repo.recovery_candidates(limit, created_from) + candidates = await service.repo.recovery_candidates(limit, created_from, service.execution_scope) resumed = 0 for task in candidates: + if task.get("execution_scope") != service.execution_scope: + continue context = task["handling_context"] returned_at = datetime.fromisoformat(context["returned_at"]) incident_at = datetime.fromisoformat(task["trigger"]["occurred_at"]) diff --git a/app/services/operational_tasks.py b/app/services/operational_tasks.py index 6b0590b..ea77c55 100644 --- a/app/services/operational_tasks.py +++ b/app/services/operational_tasks.py @@ -88,11 +88,13 @@ def _task_json(model: OperationalTaskSummary) -> dict: class OperationalTaskService: - def __init__(self, repository, context, permissions: Iterable[str], *, clock=None): + def __init__(self, repository, context, permissions: Iterable[str], *, clock=None, + execution_scope: str = "production"): self.repo = repository self.context = context self.permissions = frozenset(permissions) self.clock = clock or (lambda: datetime.now(timezone.utc)) + self.execution_scope = execution_scope def _now(self) -> datetime: now = self.clock() @@ -195,6 +197,7 @@ class OperationalTaskService: now = self._now() task_id = uuid4() task = OperationalTaskSummary( + execution_scope=self.execution_scope, project_id=self.context.project_id, task_id=task_id, task_type=command.task_type, @@ -489,6 +492,7 @@ class OperationalTaskService: task_type: str | None = None, task_created_from: datetime | None = None, ) -> tuple[list[OperationalTaskRun], str | None]: rows = await self.repo.list_runs( + execution_scope=self.execution_scope, status=status, limit=limit + 1, cursor_created_at=cursor_created_at, @@ -510,6 +514,9 @@ class OperationalTaskService: scope = f"run:{run_id}:claim" async with self.repo.transaction(): await self.repo.require_execution_storage() + task = await self._task(task_id) + require(task.get("execution_scope") == self.execution_scope, + "EXECUTION_SCOPE_MISMATCH", "当前环境不能领取其他执行范围的任务", status=403) replay = await self._request_replay(scope, command.request_id, body) if replay: return OperationalTaskRun.model_validate(replay["run"]) @@ -759,6 +766,7 @@ class OperationalTaskService: async with self.repo.transaction(): for run in await self.repo.claim_expired_runs( limit, task_type=task_type, task_created_from=task_created_from, + execution_scope=self.execution_scope, ): # Lease recovery already holds a row lock. Try rather than wait # for the shared run lock to avoid reversing the lock order of @@ -2516,6 +2524,7 @@ class OperationalTaskService: due_at: datetime, now: datetime, ) -> OperationalTaskSummary: return OperationalTaskSummary( + execution_scope=self.execution_scope, project_id=self.context.project_id, task_id=task_id, task_type="scheduled_flow", workflow=schedule.schedule_key, lifecycle_status="open", status="queued", priority=schedule.priority, revision=1, diff --git a/app/services/scada_replay.py b/app/services/scada_replay.py index b6bcc57..4567b29 100644 --- a/app/services/scada_replay.py +++ b/app/services/scada_replay.py @@ -6,6 +6,7 @@ import json from uuid import NAMESPACE_URL, uuid4, uuid5 from fastapi import HTTPException +from app.core.config import settings from app.auth.permissions import permissions_for_context from app.domain.schemas.project_settings import ScadaMonitoringSettings @@ -439,7 +440,8 @@ class ScadaReplayService: alerts = TaskAlertService(TaskAlertRepository(self.repo.connection, self.context.project_id), self.context, internal_producer_id=producer) tasks = OperationalTaskService(OperationalTaskRepository(self.repo.connection, self.context.project_id), - self.context, permissions_for_context(self.context)) + self.context, permissions_for_context(self.context), + execution_scope=settings.operational_task_execution_scope) for event in changes: # Period diagnosis owns measurement notifications during manual steps. # Retain rule events as evidence; quality alerts remain independent. @@ -503,6 +505,7 @@ class ScadaReplayService: tasks = OperationalTaskService( OperationalTaskRepository(self.repo.connection, self.context.project_id), self.context, permissions_for_context(self.context), + execution_scope=settings.operational_task_execution_scope, ) await tasks.advance_replay_investigation(UUID(incident["task_id"]), { "replay_id": document["replay_id"], "knowledge_cutoff": monitoring["knowledge_cutoff"], @@ -520,6 +523,7 @@ class ScadaReplayService: OperationalTaskRepository(self.repo.connection, self.context.project_id), self.context, permissions_for_context(self.context), + execution_scope=settings.operational_task_execution_scope, ) external = incident["external_id"] pressure = monitoring["pressure"] diff --git a/docs/operational-task-workers.md b/docs/operational-task-workers.md index 305c79b..7410bb8 100644 --- a/docs/operational-task-workers.md +++ b/docs/operational-task-workers.md @@ -190,3 +190,27 @@ OPERATIONAL_TASKS_STORAGE_TEST=1 conda run -n server python -m pytest tests/inte ``` 第二条需要当前配置的 PostgreSQL 可连接及测试 DDL 权限。测试在强制回滚事务中创建表结构和隔离项目数据，结束后不保留这些修改；不代替正式迁移。外部工单调用仍使用隔离 HTTP transport，不会操作实际工单系统。该测试覆盖持久化报告、反馈重新排队、旧会话拒绝、人工确认与丢响应回执核对，而非真实模型推理质量。 + +### 共用数据库时的执行范围 + +任务创建时，Server 将 `OPERATIONAL_TASK_EXECUTION_SCOPE`（留空则使用 +`ENVIRONMENT`）写入 `workbench.operational_tasks.document.execution_scope`。 +本地配置 `development`，线上配置 `production`；浏览器不提交此字段， +重试保留原范围。`execution_scope` 为必填字段，运行时代码不为缺失值兜底。 +现有任务通过 `resources/sql/034_operational_task_execution_scope.sql` 一次性补为 +`production`；已有明确范围的任务保持原值。此回填是历史任务的迁移策略， +不代表通过旧记录推断出了创建环境。 + +迁移需要与新版本切换协调执行：暂停任务写入，执行每个业务库的 SQL， +启动新版本并恢复任务写入。旧版本的严格模型不接受新增 JSON 字段， +新版本也拒绝未迁移数据；不能在旧版本仍服务任务请求时单独回填。 +SQL 同时增加必填范围约束，不设置数据库默认值。回滚版本前需同步处理该约束 +与 JSON 字段，不得直接切回旧镜像。 + +Worker 应连接所属环境的 `OPERATIONAL_WORKER_SERVER_URL` 和 +`OPERATIONAL_WORKER_AGENT_URL`。Server 在运行列表、领取、租约回收与 +SCADA recovery 候选查询上限制执行范围；直接领取其他范围返回 +`403 EXECUTION_SCOPE_MISMATCH`。这属于配置隔离，不能替代服务凭据授权： +不要让本地 Worker 通过线上 Server 领取任务。普通任务与历史记录仍共用存储。 + +修改范围配置需要重启相应 Server。暂停的本地 Worker 不会因代码更新恢复。 diff --git a/resources/sql/034_operational_task_execution_scope.sql b/resources/sql/034_operational_task_execution_scope.sql new file mode 100644 index 0000000..d0ff4d7 --- /dev/null +++ b/resources/sql/034_operational_task_execution_scope.sql @@ -0,0 +1,26 @@ +-- Run against each project business database during the coordinated Server cutover. +-- Historical tasks are explicitly assigned to production; retain existing scopes. +BEGIN; +SET LOCAL lock_timeout = '3s'; +SET LOCAL statement_timeout = '30s'; + +UPDATE workbench.operational_tasks +SET document = jsonb_set(document, '{execution_scope}', '"production"'::jsonb) +WHERE NOT (document ? 'execution_scope'); + +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conrelid = 'workbench.operational_tasks'::regclass + AND conname = 'operational_tasks_execution_scope_required' + ) THEN + ALTER TABLE workbench.operational_tasks + ADD CONSTRAINT operational_tasks_execution_scope_required CHECK ( + (jsonb_typeof(document->'execution_scope') = 'string' + AND length(btrim(document->>'execution_scope')) > 0) IS TRUE + ); + END IF; +END $$; + +COMMIT; diff --git a/tests/unit/test_operational_tasks.py b/tests/unit/test_operational_tasks.py index 398784b..ef3be47 100644 --- a/tests/unit/test_operational_tasks.py +++ b/tests/unit/test_operational_tasks.py @@ -235,6 +235,7 @@ async def test_targeted_notification_cannot_be_acknowledged_without_permission() def _task_document(context, now, *, lease=None): task_id = uuid4() return OperationalTaskSummary( + execution_scope="production", project_id=context.project_id, task_id=task_id, task_type="event_investigation", @@ -436,16 +437,39 @@ def _active_run(task, now): @pytest.mark.anyio -async def test_run_claim_persists_agent_instance_ownership(): +@pytest.mark.parametrize("task_scope,worker_scope,allowed", [ + (None, "production", False), + (None, "development", False), + ("production", "production", True), + ("production", "development", False), + ("development", "production", False), + ("development", "development", True), +]) +async def test_run_claim_persists_agent_instance_ownership(task_scope, worker_scope, allowed): now = datetime(2026, 9, 23, 8, tzinfo=timezone.utc) context = _context() task = _task_document(context, now, lease=None) task.update(status="queued", agent_session_id=None, agent_lease_expires_at=None) + if task_scope is None: + task.pop("execution_scope", None) + else: + task["execution_scope"] = task_scope run = _active_run(task, now) run.update(status="queued", revision=1, agent_session_id=None, lease_expires_at=None, heartbeat_at=None, started_at=None) repository = RunTaskRepository(task, run) - service = OperationalTaskService(repository, context, set(), clock=lambda: now) + service = OperationalTaskService(repository, context, set(), clock=lambda: now, + execution_scope=worker_scope) + + if not allowed: + with pytest.raises(DomainError) as failure: + await service.claim_run(task["task_id"], run["run_id"], AgentRunCommand( + request_id=run["execution_request_id"], expected_revision=1, + source_session_id="session-1", agent_instance_id="agent-a")) + assert failure.value.code == "EXECUTION_SCOPE_MISMATCH" + assert repository.run["status"] == "queued" + assert repository.task["status"] == "queued" + return claimed = await service.claim_run( task["task_id"], run["run_id"], @@ -2371,3 +2395,35 @@ async def test_recovery_resolves_alerts_with_original_facts_and_stable_event_id( assert command.event_id == uuid5(NAMESPACE_URL, f"recovery:{task_id}:recovery:evidence-1:incident-1") assert command.source.model_dump(mode="json") == alert["source"] assert command.title == alert["title"] and command.description == alert["description"] + + +@pytest.mark.anyio +@pytest.mark.parametrize("scope", ["production", "development"]) +async def test_task_creation_persists_server_execution_scope(monkeypatch, scope): + from app.domain.schemas.operational_task import OperationalTaskCreate + + now = datetime(2026, 11, 8, tzinfo=timezone.utc) + context = _context() + document = _task_document(context, now) + repository = TaskRepository(document) + repository.insert_task = AsyncMock() + repository.insert_run = AsyncMock() + service = OperationalTaskService(repository, context, set(), clock=lambda: now, + execution_scope=scope) + monkeypatch.setattr(service, "_task_event", AsyncMock()) + monkeypatch.setattr(service, "_run_event", AsyncMock()) + monkeypatch.setattr(service, "detail", AsyncMock()) + command = OperationalTaskCreate(request_id=uuid4(), **{ + key: document[key] for key in ("task_type", "workflow", "title", "summary", "source", "trigger") + }) + await service.create(command) + assert repository.insert_task.await_args.args[0]["execution_scope"] == scope + + +def test_task_scope_is_required_in_persisted_contract(): + task = _task_document(_context(), datetime(2026, 11, 8, tzinfo=timezone.utc)) + task.pop("execution_scope") + with pytest.raises(ValidationError) as error: + OperationalTaskSummary.model_validate(task) + assert any(item["loc"] == ("execution_scope",) and item["type"] == "missing" + for item in error.value.errors()) diff --git a/tests/unit/test_scada_replay_service.py b/tests/unit/test_scada_replay_service.py index 296fa3b..995d135 100644 --- a/tests/unit/test_scada_replay_service.py +++ b/tests/unit/test_scada_replay_service.py @@ -579,7 +579,7 @@ def test_suspected_monitoring_creates_burst_report_with_frozen_replay_context(mo monkeypatch.setattr("app.infra.db.postgresql.operational_tasks.OperationalTaskRepository", lambda *_: object()) monkeypatch.setattr( "app.services.operational_tasks.OperationalTaskService", - lambda *_: SimpleNamespace(create=create_task), + lambda *_, **_kwargs: SimpleNamespace(create=create_task), ) monitoring = { "status": "suspected_burst", @@ -618,7 +618,7 @@ def test_monitoring_summary_counts_signals_without_copying_reading_details(monke _, 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 *_: SimpleNamespace(create=create_task)) + 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"} @@ -641,7 +641,7 @@ def test_continuing_burst_reuses_task_across_persisted_periods(monkeypatch): 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 *_: SimpleNamespace(create=create_task, advance_replay_investigation=advance_task)) + lambda *_, **_kwargs: SimpleNamespace(create=create_task, advance_replay_investigation=advance_task)) document = {"replay_id": str(uuid4())} async def run(): @@ -683,7 +683,7 @@ def test_incident_recovers_once_then_allows_new_accident(monkeypatch): 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 *_: SimpleNamespace(create=create_task)) + lambda *_, **_kwargs: SimpleNamespace(create=create_task)) document = {"replay_id": str(uuid4())} async def run(): @@ -784,7 +784,7 @@ def test_combined_monitoring_keeps_rule_evidence_and_quality_notifications(monke 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 *_: SimpleNamespace(create=AsyncMock())) + 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} @@ -812,7 +812,7 @@ def test_failed_incident_publication_rolls_back_state_and_retries_same_task_requ 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 *_: SimpleNamespace(create=create_task)) + 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} @@ -843,7 +843,7 @@ def test_any_single_measurement_anomaly_enters_diagnosis_below_burst_threshold(m _, 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 *_: SimpleNamespace(create=create_task)) + 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