185 lines
7.7 KiB
Python
185 lines
7.7 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
from uuid import uuid4
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.infra.integrations.operational_actions import (
|
|
HttpOperationalActionDispatcher, OperationalActionSettings,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def anyio_backend():
|
|
return "asyncio"
|
|
|
|
|
|
def inputs(kind="create_work_order"):
|
|
project_id, task_id = uuid4(), uuid4()
|
|
now = datetime.now(timezone.utc)
|
|
payload = {"title": "现场采样", "description": "核验投诉水质", "work_order_type": "sampling"}
|
|
if kind == "issue_command":
|
|
payload = {"command_type": "set_valve", "target": {"kind": "valve", "id": "V-01"},
|
|
"parameters": {"opening": 20}, "impact_summary": "影响V-01阀门控制范围",
|
|
"preconditions": ["现场人员确认"],
|
|
"expires_at": (now + timedelta(minutes=10)).isoformat()}
|
|
return {
|
|
"context": SimpleNamespace(project_id=project_id),
|
|
"task": SimpleNamespace(task_id=task_id, project_id=project_id),
|
|
"action": SimpleNamespace(task_id=task_id, action_id=uuid4(), version=1, kind=kind,
|
|
execution_request_id=uuid4(),
|
|
approval_status="approved", payload=payload,
|
|
decided_by="operator", decided_at=now),
|
|
"credential": "Bearer TEST_ONLY",
|
|
}
|
|
|
|
|
|
def settings(**kwargs):
|
|
return OperationalActionSettings(_env_file=None, peripheral_url="http://sample.test",
|
|
command_gateway_url="http://commands.test", **kwargs)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_work_order_confirms_creation_not_field_completion():
|
|
data = inputs()
|
|
calls = []
|
|
def handler(request):
|
|
calls.append(request)
|
|
return httpx.Response(201, json={"work_order_id": str(uuid4()),
|
|
"project_id": str(data["context"].project_id), "created_at": "2026-09-23T08:00:00Z",
|
|
"origin": "synthetic", "status": "open"})
|
|
adapter = HttpOperationalActionDispatcher(settings(), transport=httpx.MockTransport(handler))
|
|
result = await adapter.dispatch(**data)
|
|
assert result.status == "succeeded"
|
|
assert result.receipt.external_status == "open"
|
|
assert "现场工作进度" in result.detail
|
|
assert calls[0].url.path == "/api/sample/v1/work-orders"
|
|
assert calls[0].headers["X-Project-Id"] == str(data["context"].project_id)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_work_order_receipt_accepts_awaiting_reply_status():
|
|
data = inputs()
|
|
def handler(request):
|
|
return httpx.Response(201, json={
|
|
"work_order_id": str(uuid4()),
|
|
"project_id": str(data["context"].project_id),
|
|
"created_at": "2026-09-23T08:00:00Z",
|
|
"origin": "synthetic",
|
|
"status": "awaiting_reply",
|
|
})
|
|
adapter = HttpOperationalActionDispatcher(
|
|
settings(), transport=httpx.MockTransport(handler)
|
|
)
|
|
|
|
result = await adapter.dispatch(**data)
|
|
|
|
assert result.status == "succeeded"
|
|
assert result.receipt.external_status == "awaiting_reply"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_action_work_order_is_not_reclassified_as_a_new_inbound_event():
|
|
data = inputs()
|
|
data["action"].payload["work_order_type"] = "water_quality_complaint"
|
|
submitted = {}
|
|
def handler(request):
|
|
submitted.update(__import__("json").loads(request.content))
|
|
return httpx.Response(201, json={
|
|
"work_order_id": str(uuid4()),
|
|
"project_id": str(data["context"].project_id),
|
|
"created_at": "2026-09-23T08:00:00Z",
|
|
"origin": "synthetic",
|
|
"status": "open",
|
|
})
|
|
|
|
result = await HttpOperationalActionDispatcher(
|
|
settings(), transport=httpx.MockTransport(handler)
|
|
).dispatch(**data)
|
|
|
|
assert result.status == "succeeded"
|
|
assert submitted["work_order_type"] == "general"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_lost_create_response_recovers_same_request_without_second_write():
|
|
data, methods = inputs(), []
|
|
def handler(request):
|
|
methods.append(request.method)
|
|
if request.method == "POST":
|
|
raise httpx.ReadTimeout("lost", request=request)
|
|
return httpx.Response(200, json={"request_id": request.url.path.split("/")[-1],
|
|
"result": {"work_order_id": str(uuid4()), "project_id": str(data["context"].project_id),
|
|
"created_at": "2026-09-23T08:00:00Z", "origin": "synthetic", "status": "open"}})
|
|
result = await HttpOperationalActionDispatcher(settings(), transport=httpx.MockTransport(handler)).dispatch(**data)
|
|
assert result.status == "succeeded"
|
|
assert methods == ["POST", "GET"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_unconfirmed_receipt_remains_unknown_not_failed_or_succeeded():
|
|
data = inputs()
|
|
def handler(request):
|
|
return httpx.Response(200, json={"work_order_id": str(uuid4()), "project_id": str(uuid4()),
|
|
"created_at": "2026-09-23T08:00:00Z", "origin": "synthetic", "status": "open"})
|
|
result = await HttpOperationalActionDispatcher(settings(), transport=httpx.MockTransport(handler)).dispatch(**data)
|
|
assert result.status == "unknown"
|
|
assert result.receipt is None
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_unapproved_and_expired_actions_never_reach_external_system():
|
|
def handler(request):
|
|
raise AssertionError("no external call expected")
|
|
adapter = HttpOperationalActionDispatcher(settings(), transport=httpx.MockTransport(handler))
|
|
data = inputs()
|
|
data["action"].approval_status = "pending"
|
|
assert (await adapter.dispatch(**data)).status == "failed"
|
|
data = inputs("issue_command")
|
|
data["action"].payload["expires_at"] = "2020-01-01T00:00:00Z"
|
|
assert (await adapter.dispatch(**data)).status == "failed"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_command_acceptance_is_not_execution_completion():
|
|
data = inputs("issue_command")
|
|
def handler(request):
|
|
import json
|
|
body = json.loads(request.content)
|
|
assert body["command"]["target"]["id"] == "V-01"
|
|
return httpx.Response(202, json={"request_id": body["request_id"],
|
|
"project_id": body["project_id"], "receipt_id": "gateway-1", "status": "accepted",
|
|
"submitted_at": "2026-09-23T08:00:00Z", "completed_at": None, "detail": "已接收"})
|
|
result = await HttpOperationalActionDispatcher(settings(), transport=httpx.MockTransport(handler)).dispatch(**data)
|
|
assert result.status == "submitted"
|
|
assert result.receipt.completed_at is None
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_missing_configuration_does_not_invent_receipt():
|
|
adapter = HttpOperationalActionDispatcher(OperationalActionSettings(
|
|
_env_file=None, peripheral_url="", command_gateway_url=""))
|
|
result = await adapter.dispatch(**inputs())
|
|
assert result.status == "unsupported"
|
|
assert result.receipt is None
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_reconciliation_is_read_only_and_keeps_execution_identity_across_versions():
|
|
data, paths = inputs("issue_command"), []
|
|
def handler(request):
|
|
assert request.method == "GET"
|
|
paths.append(request.url.path)
|
|
return httpx.Response(200, json={
|
|
"request_id": str(data["action"].execution_request_id),
|
|
"project_id": str(data["context"].project_id), "receipt_id": "confirmed-command",
|
|
"status": "completed", "submitted_at": "2026-09-23T08:00:00Z",
|
|
"completed_at": "2026-09-23T08:01:00Z", "detail": "执行回执已确认",
|
|
})
|
|
adapter = HttpOperationalActionDispatcher(settings(), transport=httpx.MockTransport(handler))
|
|
assert (await adapter.reconcile(**data)).status == "succeeded"
|
|
data["action"].version += 1
|
|
assert (await adapter.reconcile(**data)).status == "succeeded"
|
|
assert paths[0] == paths[1]
|