113 lines
4.5 KiB
Python
113 lines
4.5 KiB
Python
"""
|
|
P014: services/api/orchestrator/main.py had ZERO test coverage before this
|
|
pass, and `POST /orders/{oid}/run` had no idempotency protection at all --
|
|
calling it twice for the same order re-ran the full compliant sourcing
|
|
pipeline a second time, silently overwriting `_JOBS[oid]` with a second,
|
|
independently-produced delivery. This is the actual "duplicate delivery"
|
|
gap P014's recon flagged NOT MET (nothing downstream of order creation was
|
|
idempotent). These tests exercise the real ASGI app end to end (httpx
|
|
ASGITransport, no mocked HTTP layer) and prove: a sequential retry returns
|
|
the cached job without re-running the pipeline, a TRUE CONCURRENT race
|
|
(two requests in flight at once) is also closed by the per-order asyncio
|
|
lock, and an unknown order still 404s exactly as before.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from orchestrator.main import app
|
|
|
|
|
|
def _fake_job_result(calls: list[str], order_id: str) -> dict:
|
|
calls.append(order_id)
|
|
return {
|
|
"stage": "qa",
|
|
"pct": 100,
|
|
"state": "done",
|
|
"metrics": {"rows": 1, "dropped_below_threshold": 0, "per_market": {"NO": 1}, "avg_confidence": 0.95},
|
|
"records": [{"market": "NO", "name": "Test AS", "confidence": 0.95}],
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sequential_retry_returns_cached_job_without_rerunning_pipeline(monkeypatch):
|
|
calls: list[str] = []
|
|
|
|
async def fake_run_job(order_id: str, config: dict) -> dict:
|
|
return _fake_job_result(calls, order_id)
|
|
|
|
monkeypatch.setattr("orchestrator.main.run_job", fake_run_job)
|
|
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
created = await client.post("/orders", json={"flow": "custom", "config": {"markets": ["NO"]}})
|
|
oid = created.json()["order_id"]
|
|
|
|
r1 = await client.post(f"/orders/{oid}/run")
|
|
r2 = await client.post(f"/orders/{oid}/run")
|
|
r3 = await client.post(f"/orders/{oid}/run")
|
|
|
|
assert r1.status_code == 200
|
|
assert r1.json() == r2.json() == r3.json()
|
|
assert calls == [oid] # run_job invoked exactly once across three /run calls
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_true_concurrent_race_still_invokes_run_job_exactly_once(monkeypatch):
|
|
calls: list[str] = []
|
|
|
|
async def fake_run_job(order_id: str, config: dict) -> dict:
|
|
# Widen the race window: both requests reach this await before
|
|
# either has written to _JOBS, which is exactly the race a bare
|
|
# `if oid in _JOBS` check (no lock) would lose.
|
|
await asyncio.sleep(0.05)
|
|
return _fake_job_result(calls, order_id)
|
|
|
|
monkeypatch.setattr("orchestrator.main.run_job", fake_run_job)
|
|
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
created = await client.post("/orders", json={"flow": "custom", "config": {"markets": ["NO"]}})
|
|
oid = created.json()["order_id"]
|
|
|
|
r1, r2 = await asyncio.gather(
|
|
client.post(f"/orders/{oid}/run"),
|
|
client.post(f"/orders/{oid}/run"),
|
|
)
|
|
|
|
assert r1.status_code == 200 and r2.status_code == 200
|
|
assert r1.json() == r2.json()
|
|
assert calls == [oid] # THE regression this test guards: exactly one real pipeline run
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_on_unknown_order_still_404s():
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
res = await client.post("/orders/does-not-exist/run")
|
|
assert res.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_different_orders_each_get_their_own_real_run(monkeypatch):
|
|
calls: list[str] = []
|
|
|
|
async def fake_run_job(order_id: str, config: dict) -> dict:
|
|
return _fake_job_result(calls, order_id)
|
|
|
|
monkeypatch.setattr("orchestrator.main.run_job", fake_run_job)
|
|
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
o1 = (await client.post("/orders", json={"flow": "custom", "config": {"markets": ["NO"]}})).json()["order_id"]
|
|
o2 = (await client.post("/orders", json={"flow": "custom", "config": {"markets": ["GB"]}})).json()["order_id"]
|
|
|
|
await client.post(f"/orders/{o1}/run")
|
|
await client.post(f"/orders/{o2}/run")
|
|
|
|
# Idempotency is per-order, not global -- two distinct orders both really run.
|
|
assert sorted(calls) == sorted([o1, o2])
|