233 lines
12 KiB
Python
233 lines
12 KiB
Python
"""Opt-in v2 concurrent writer regression; every event is rolled back."""
|
|
import asyncio
|
|
import os
|
|
from datetime import datetime, timezone
|
|
from uuid import uuid4
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
import psycopg
|
|
import pytest
|
|
from psycopg.conninfo import make_conninfo
|
|
from psycopg.rows import dict_row
|
|
from psycopg.pq import TransactionStatus
|
|
|
|
from app.core.config import get_pgconn_string
|
|
from app.auth.permissions import OPERATIONAL_ACTION_WORK_ORDER_APPROVE, OPERATIONAL_ACTION_WORK_ORDER_EXECUTE
|
|
from app.auth.project_dependencies import ProjectContext
|
|
from app.domain.schemas.operational_task import (
|
|
ActionExecuteCommand, ActionReconcileCommand, ActionReceipt, HumanApprovalAction,
|
|
WorkOrderConfirmDispatchCommand,
|
|
OperationalTaskCreate,
|
|
)
|
|
from app.infra.db.postgresql.operational_tasks import OperationalTaskRepository
|
|
from app.infra.db.postgresql.task_alerts import TaskAlertRepository
|
|
from app.infra.db.postgresql.workbench_events import lock_event_transaction
|
|
from app.infra.integrations.operational_actions import ActionDispatchOutcome
|
|
from app.services.operational_tasks import OperationalTaskService
|
|
|
|
|
|
pytestmark = pytest.mark.skipif(
|
|
os.getenv("WORKBENCH_EVENT_ORDER_TEST") != "1", reason="explicit v2 concurrency opt-in required",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def anyio_backend():
|
|
return "asyncio"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_later_producer_cannot_allocate_a_cursor_before_first_transaction_finishes():
|
|
dsn = make_conninfo(get_pgconn_string(), dbname="tjwater_v2")
|
|
first_ready, release_first, second_ready = asyncio.Event(), asyncio.Event(), asyncio.Event()
|
|
project_id = uuid4()
|
|
async with await psycopg.AsyncConnection.connect(dsn, autocommit=True, row_factory=dict_row) as first:
|
|
async with await psycopg.AsyncConnection.connect(dsn, autocommit=True, row_factory=dict_row) as second:
|
|
async with await psycopg.AsyncConnection.connect(dsn, autocommit=True, row_factory=dict_row) as observer:
|
|
assert first.info.dbname == second.info.dbname == observer.info.dbname == "tjwater_v2"
|
|
|
|
async def write_first():
|
|
async with first.transaction(force_rollback=True):
|
|
repo = OperationalTaskRepository(first, project_id)
|
|
async with repo.transaction():
|
|
await repo.append_event(
|
|
topic="operational.task.changed", aggregate_type="operational_task",
|
|
aggregate_id=str(uuid4()), aggregate_revision=1,
|
|
occurred_at=datetime.now(timezone.utc), metadata={},
|
|
)
|
|
first_ready.set()
|
|
await release_first.wait()
|
|
|
|
async def write_second():
|
|
async with second.transaction(force_rollback=True):
|
|
async with TaskAlertRepository(second, project_id).transaction():
|
|
second_ready.set()
|
|
|
|
first_task = asyncio.create_task(write_first())
|
|
second_task = None
|
|
try:
|
|
await asyncio.wait_for(first_ready.wait(), 3)
|
|
second_task = asyncio.create_task(write_second())
|
|
async with asyncio.timeout(3):
|
|
while True:
|
|
row = await (await observer.execute(
|
|
"SELECT wait_event_type FROM pg_stat_activity WHERE pid=%s",
|
|
(second.info.backend_pid,),
|
|
)).fetchone()
|
|
if row and row["wait_event_type"] == "Lock":
|
|
break
|
|
assert not second_ready.is_set(), "later producer bypassed the event ordering lock"
|
|
await asyncio.sleep(0.01)
|
|
assert not second_ready.is_set()
|
|
release_first.set()
|
|
await asyncio.wait_for(asyncio.gather(first_task, second_task), 3)
|
|
assert second_ready.is_set()
|
|
row = await (await observer.execute(
|
|
"SELECT count(*) AS n FROM workbench.domain_event_log WHERE project_id=%s",
|
|
(project_id,),
|
|
)).fetchone()
|
|
assert row["n"] == 0
|
|
finally:
|
|
release_first.set()
|
|
pending = [task for task in (first_task, second_task) if task is not None]
|
|
for task in pending:
|
|
if not task.done():
|
|
task.cancel()
|
|
await asyncio.gather(*pending, return_exceptions=True)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize("operation", ["confirm", "submit", "reconcile"])
|
|
async def test_default_connection_ends_transactions_before_external_dispatch(monkeypatch, operation):
|
|
"""Real psycopg reads/locks and service orchestration, without fixture writes."""
|
|
dsn = make_conninfo(get_pgconn_string(), dbname="tjwater_v2")
|
|
task_id, action_id, order_id = uuid4(), uuid4(), uuid4()
|
|
now = datetime.now(timezone.utc)
|
|
context = ProjectContext(project_id=uuid4(), project_code="event-order-test",
|
|
user_id=uuid4(), project_role="member")
|
|
action = HumanApprovalAction(
|
|
action_id=action_id, task_id=task_id, version=1, kind="create_work_order",
|
|
reason="AUTOTEST transaction boundary", evidence_refs=[],
|
|
payload={"title": "AUTOTEST", "description": "not persisted", "work_order_type": "general"},
|
|
required_approval_permission=OPERATIONAL_ACTION_WORK_ORDER_APPROVE,
|
|
required_execution_permission=OPERATIONAL_ACTION_WORK_ORDER_EXECUTE,
|
|
approval_status="pending" if operation == "confirm" else "approved",
|
|
execution_status="submitted" if operation == "reconcile" else "not_submitted",
|
|
proposed_by="AUTOTEST", proposed_at=now,
|
|
)
|
|
async with await psycopg.AsyncConnection.connect(dsn, row_factory=dict_row) as conn:
|
|
async with await psycopg.AsyncConnection.connect(dsn, autocommit=True) as external:
|
|
assert conn.info.dbname == external.info.dbname == "tjwater_v2"
|
|
repo = OperationalTaskRepository(conn, context.project_id)
|
|
service = OperationalTaskService(repo, context, {
|
|
OPERATIONAL_ACTION_WORK_ORDER_APPROVE, OPERATIONAL_ACTION_WORK_ORDER_EXECUTE,
|
|
})
|
|
|
|
async def get_action(*_args):
|
|
await conn.execute("SELECT 1") # Same implicit BEGIN as a product read.
|
|
return action.model_dump(mode="json")
|
|
|
|
async def update_action(_task_id, document):
|
|
nonlocal action
|
|
await conn.execute("SELECT 1")
|
|
action = HumanApprovalAction.model_validate(document)
|
|
|
|
async def get_task(*_args):
|
|
await conn.execute("SELECT 1")
|
|
return {"title": "AUTOTEST", "lifecycle_status": "open", "source": {}}
|
|
|
|
async def get_request(*_args):
|
|
await conn.execute("SELECT 1")
|
|
return None
|
|
|
|
async def detail(*_args):
|
|
await conn.execute("SELECT 1")
|
|
return SimpleNamespace(title="AUTOTEST", task_id=task_id, project_id=context.project_id)
|
|
|
|
monkeypatch.setattr(repo, "get_action", get_action)
|
|
monkeypatch.setattr(repo, "get_task", get_task)
|
|
monkeypatch.setattr(repo, "get_request", get_request)
|
|
monkeypatch.setattr(repo, "update_action", update_action)
|
|
monkeypatch.setattr(repo, "save_request", AsyncMock())
|
|
monkeypatch.setattr(service, "detail", detail)
|
|
monkeypatch.setattr(service, "_action_event", AsyncMock())
|
|
monkeypatch.setattr(service, "_resolve_notifications", AsyncMock())
|
|
monkeypatch.setattr(service, "_notify", AsyncMock())
|
|
|
|
async def dispatch(**_kwargs):
|
|
assert conn.info.transaction_status == TransactionStatus.IDLE, \
|
|
"implicit outer transaction survives into the external call"
|
|
async with external.transaction():
|
|
await external.execute("SET LOCAL lock_timeout='1s'")
|
|
await lock_event_transaction(external)
|
|
return ActionDispatchOutcome("submitted", "AUTOTEST receipt", ActionReceipt(
|
|
adapter="peripheral_sample", receipt_id=str(order_id), submitted_at=now,
|
|
external_status="open", detail="not persisted",
|
|
))
|
|
|
|
dispatcher = SimpleNamespace(dispatch=dispatch, reconcile=dispatch)
|
|
command = dict(request_id=uuid4(), expected_version=1)
|
|
if operation == "confirm":
|
|
result = await service.confirm_work_order_dispatch(
|
|
task_id, action_id, WorkOrderConfirmDispatchCommand(**command), dispatcher, None,
|
|
)
|
|
elif operation == "submit":
|
|
result = await service.submit_action(
|
|
task_id, action_id, ActionExecuteCommand(**command), dispatcher, None,
|
|
)
|
|
else:
|
|
result = await service.reconcile_action(
|
|
task_id, action_id, ActionReconcileCommand(**command), dispatcher, None,
|
|
)
|
|
assert result.execution_status == "submitted"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_work_order_filter_matches_source_and_receipts_without_cross_project_rows():
|
|
dsn = make_conninfo(get_pgconn_string(), dbname="tjwater_v2")
|
|
project_id, order_id = uuid4(), uuid4()
|
|
context = ProjectContext(project_id=project_id, project_code="event-filter-test",
|
|
user_id=uuid4(), project_role="member")
|
|
now = datetime.now(timezone.utc)
|
|
async with await psycopg.AsyncConnection.connect(dsn, row_factory=dict_row) as conn:
|
|
assert conn.info.dbname == "tjwater_v2"
|
|
async with conn.transaction(force_rollback=True):
|
|
repo = OperationalTaskRepository(conn, project_id)
|
|
service = OperationalTaskService(repo, context, set())
|
|
|
|
async def create(source, *, task_type="external_work_order"):
|
|
return await service.create(OperationalTaskCreate(
|
|
request_id=uuid4(), task_type=task_type, workflow="external-work-order",
|
|
title="AUTOTEST filtered work order", summary="isolated rollback", source=source,
|
|
trigger={"kind": "manual", "occurred_at": now, "detail": "filter regression"},
|
|
))
|
|
|
|
projection = await create({"kind": "external_system", "external_id": f"peripheral_sample:{order_id}"})
|
|
investigation = await create({"kind": "external_system", "external_id": str(order_id)},
|
|
task_type="event_investigation")
|
|
linked = await create({"kind": "operator"}, task_type="event_investigation")
|
|
await create({"kind": "external_system", "external_id": f"peripheral_sample:{uuid4()}"})
|
|
action = HumanApprovalAction(
|
|
action_id=uuid4(), task_id=linked.task_id, version=1, kind="create_work_order",
|
|
reason="AUTOTEST receipt", evidence_refs=[],
|
|
payload={"title": "AUTOTEST", "description": "rollback", "work_order_type": "general"},
|
|
required_approval_permission=OPERATIONAL_ACTION_WORK_ORDER_APPROVE,
|
|
required_execution_permission=OPERATIONAL_ACTION_WORK_ORDER_EXECUTE,
|
|
approval_status="approved", execution_status="submitted", proposed_by="AUTOTEST", proposed_at=now,
|
|
receipt=ActionReceipt(adapter="peripheral_sample", receipt_id=str(order_id), submitted_at=now,
|
|
external_status="open", detail="rollback"),
|
|
)
|
|
await repo.insert_action(str(linked.task_id), action.model_dump(mode="json"))
|
|
items, cursor = await service.list(task_type=None, status=None, limit=100, offset=0,
|
|
peripheral_work_order_id=order_id)
|
|
assert {item.task_id for item in items} == {projection.task_id, investigation.task_id, linked.task_id}
|
|
assert cursor is None
|
|
foreign = OperationalTaskRepository(conn, uuid4())
|
|
assert await foreign.list_tasks(task_type=None, status=None, limit=100, offset=0,
|
|
peripheral_work_order_id=str(order_id)) == []
|
|
row = await (await conn.execute(
|
|
"SELECT count(*) AS n FROM workbench.operational_tasks WHERE project_id=%s", (project_id,),
|
|
)).fetchone()
|
|
assert row["n"] == 0
|