129 lines
4.8 KiB
Python
129 lines
4.8 KiB
Python
"""Engine tests with fake processors (no LLM): flow, barrier, retry/dead-letter, producer race."""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
import kanban
|
|
from kanban import Flow, Stage, chain_stages, run_flow
|
|
|
|
TOPIC = "t"
|
|
BOARD = "inventory"
|
|
|
|
|
|
def _advance_proc(db, to_stage):
|
|
async def proc(cards):
|
|
await db.kanban_advance_many(TOPIC, BOARD, [(c["card_id"], to_stage) for c in cards])
|
|
return proc
|
|
|
|
|
|
async def _seed(db, n, stage="s1"):
|
|
for i in range(n):
|
|
await db.kanban_upsert_card(TOPIC, BOARD, f"card-{i}", "title", stage, {"title": f"T{i}"})
|
|
|
|
|
|
async def test_cards_flow_through_stages(testdb):
|
|
db = testdb
|
|
await _seed(db, 7)
|
|
flow = Flow(TOPIC)
|
|
stages = chain_stages([
|
|
Stage(BOARD, "s1", _advance_proc(db, "s2")),
|
|
Stage(BOARD, "s2", _advance_proc(db, "done")),
|
|
])
|
|
await asyncio.wait_for(run_flow(flow, stages), timeout=10)
|
|
assert await db.kanban_count(TOPIC, "done", board=BOARD) == 7
|
|
assert await db.kanban_count(TOPIC, ["s1", "s2"], board=BOARD) == 0
|
|
|
|
|
|
async def test_barrier_waits_for_upstream(testdb):
|
|
db = testdb
|
|
await _seed(db, 6)
|
|
upstream_left: list[int] = []
|
|
|
|
async def slow_s1(cards):
|
|
await asyncio.sleep(0.05) # keep upstream busy so an eager barrier would see queued cards
|
|
await db.kanban_advance_many(TOPIC, BOARD, [(c["card_id"], "gate") for c in cards])
|
|
|
|
async def barrier_proc(cards):
|
|
upstream_left.append(await db.kanban_count(TOPIC, ["s1"], board=BOARD))
|
|
await db.kanban_advance_many(TOPIC, BOARD, [(c["card_id"], "done") for c in cards])
|
|
|
|
flow = Flow(TOPIC)
|
|
stages = chain_stages([
|
|
Stage(BOARD, "s1", slow_s1),
|
|
Stage(BOARD, "gate", barrier_proc, barrier=True),
|
|
])
|
|
await asyncio.wait_for(run_flow(flow, stages), timeout=10)
|
|
assert await db.kanban_count(TOPIC, "done", board=BOARD) == 6
|
|
assert upstream_left and all(n == 0 for n in upstream_left) # barrier never ran with s1 queued
|
|
|
|
|
|
async def test_retry_backoff_then_dead(testdb, monkeypatch):
|
|
db = testdb
|
|
monkeypatch.setattr(kanban, "RETRY_BACKOFF", 0.02)
|
|
await _seed(db, 1)
|
|
attempts = []
|
|
|
|
async def failing(cards):
|
|
attempts.append(cards[0]["retries"])
|
|
raise RuntimeError("kaputt")
|
|
|
|
flow = Flow(TOPIC)
|
|
stages = chain_stages([Stage(BOARD, "s1", failing)])
|
|
await asyncio.wait_for(run_flow(flow, stages), timeout=10)
|
|
card = await db.kanban_get_card(TOPIC, BOARD, "card-0")
|
|
assert card["stage"] == "dead"
|
|
assert card["retries"] == kanban.MAX_CARD_RETRIES
|
|
assert "kaputt" in card["last_error"]
|
|
assert attempts == [0, 1, 2] # backoff between attempts, then dead-letter
|
|
|
|
|
|
async def test_requeue_dead(testdb):
|
|
db = testdb
|
|
await db.kanban_upsert_card(TOPIC, BOARD, "card-0", "title", "s1")
|
|
for _ in range(kanban.MAX_CARD_RETRIES):
|
|
await db.kanban_fail_card(TOPIC, BOARD, "card-0", "x", kanban.MAX_CARD_RETRIES, 0.0)
|
|
assert (await db.kanban_get_card(TOPIC, BOARD, "card-0"))["stage"] == "dead"
|
|
assert await db.kanban_requeue_dead(TOPIC, BOARD, "s1") == 1
|
|
card = await db.kanban_get_card(TOPIC, BOARD, "card-0")
|
|
assert card["stage"] == "s1" and card["retries"] == 0
|
|
|
|
|
|
async def test_producer_attach_in_idle_lull(testdb):
|
|
"""Fix-6 regression: a producer attached while workers sit in the exit grace poll
|
|
must keep the flow alive and its cards must still be processed."""
|
|
db = testdb
|
|
flow = Flow(TOPIC)
|
|
stages = chain_stages([Stage(BOARD, "s1", _advance_proc(db, "done"))])
|
|
|
|
async def producer_a():
|
|
await db.kanban_upsert_card(TOPIC, BOARD, "card-a", "title", "s1")
|
|
flow.wake.set()
|
|
flow.done_producer()
|
|
|
|
async def attacher():
|
|
while await db.kanban_count(TOPIC, "done", board=BOARD) == 0: # wait for card-a done
|
|
await asyncio.sleep(0.01)
|
|
flow.add_producer() # synchronous BEFORE the work — the grace poll must see it
|
|
|
|
async def producer_b():
|
|
await db.kanban_upsert_card(TOPIC, BOARD, "card-b", "title", "s1")
|
|
flow.wake.set()
|
|
flow.done_producer()
|
|
await producer_b()
|
|
|
|
flow.add_producer() # producer_a, counted before run_flow (sync add)
|
|
asyncio.get_event_loop().create_task(attacher())
|
|
await asyncio.wait_for(run_flow(flow, stages, producers=[producer_a()]), timeout=10)
|
|
assert await db.kanban_count(TOPIC, "done", board=BOARD) == 2
|
|
|
|
|
|
async def test_backoff_delays_pull(testdb, monkeypatch):
|
|
db = testdb
|
|
await db.kanban_upsert_card(TOPIC, BOARD, "card-0", "title", "s1")
|
|
await db.kanban_fail_card(TOPIC, BOARD, "card-0", "x", 5, 0.2)
|
|
assert await db.kanban_pull(TOPIC, BOARD, "s1", 10) == [] # in backoff → not pullable
|
|
assert await db.kanban_count(TOPIC, "s1", board=BOARD) == 1 # but still counts as queued
|
|
await asyncio.sleep(0.25)
|
|
assert len(await db.kanban_pull(TOPIC, BOARD, "s1", 10)) == 1
|