c1782b9524
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
239 lines
12 KiB
Python
239 lines
12 KiB
Python
"""Параллельная обработка ограничена (spec image-recognition-gateway, задача 5.4; D-3, `SCAN_MAX_INFLIGHT`).
|
||
|
||
Сценарий spec «Пять заданий подряд»: пять запросов при `SCAN_MAX_INFLIGHT = 4` → пять ack'ов сразу, четыре файла во
|
||
входящем каталоге, пятый — только после завершения одного из четырёх. Плюс строгая очерёдность при одном слоте и
|
||
остановка воркера, когда задания ждут слот (T12.9).
|
||
|
||
Детерминизм без «поспать и посмотреть»: пауза между опросами горячей папки подменена (`SteppedSleep`) — задание,
|
||
записавшее файл и опросившее исходящий каталог, «паркуется» и опрашивает снова только по шагу теста. Ожидания —
|
||
по событиям (`asyncio.Condition`) с явным тайм-аутом. Ответ ИС кладёт сам тест (`env.answer`) и ровно для того
|
||
задания, слот которого освобождается. Чтение S3 у фейка не уступает цикл событий, поэтому задание, прошедшее мимо
|
||
семафора, читает S3 в том же проходе цикла, что и его ack, — счётчик чтений S3 различает «ждёт слот» и «слот не нужен»
|
||
без гонок.
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
from collections import Counter
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from tests.worker_helpers import SUCCESS_ANSWER, Env, EnvFactory
|
||
|
||
# Логи воркера перехватываются (`log_lines` из conftest): вывод pytest чистый
|
||
pytestmark = pytest.mark.usefixtures("log_lines")
|
||
|
||
WAIT_TIMEOUT = 3.0
|
||
|
||
|
||
class SteppedSleep:
|
||
"""Подмена `sleep` воркера: пауза между опросами длится, пока тест не сделает шаг (`step`) или не откроет её.
|
||
|
||
Поколение растёт с каждым шагом: `wait_parked(n)` ждёт n заданий, припаркованных уже после последнего шага, —
|
||
так ожидание не срабатывает на заданиях, которые только что отпущены, но ещё не успели проснуться.
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
self._generation = 0
|
||
self._parked: Counter[int] = Counter()
|
||
self._gate = asyncio.Event()
|
||
self._opened = False
|
||
self._changed = asyncio.Condition()
|
||
|
||
@property
|
||
def parked(self) -> int:
|
||
"""Сколько заданий припарковано после последнего шага."""
|
||
return self._parked[self._generation]
|
||
|
||
async def __call__(self, seconds: float) -> None:
|
||
if self._opened:
|
||
await asyncio.sleep(0)
|
||
return
|
||
generation, gate = self._generation, self._gate
|
||
async with self._changed:
|
||
self._parked[generation] += 1
|
||
self._changed.notify_all()
|
||
try:
|
||
await gate.wait()
|
||
finally:
|
||
self._parked[generation] -= 1
|
||
|
||
def step(self) -> None:
|
||
"""Отпустить всех припаркованных: каждое задание опросит исходящий каталог ещё раз."""
|
||
self._gate.set()
|
||
self._gate = asyncio.Event()
|
||
self._generation += 1
|
||
|
||
def open(self) -> None:
|
||
"""Снять паузы совсем: дальше опросы идут без ожидания."""
|
||
self._opened = True
|
||
self._gate.set()
|
||
|
||
async def wait_parked(self, count: int) -> None:
|
||
async with self._changed:
|
||
await asyncio.wait_for(self._changed.wait_for(lambda: self.parked >= count), WAIT_TIMEOUT)
|
||
|
||
|
||
def job_of(name: str) -> str:
|
||
"""`job_id` из имени файла входящего каталога `<job_id>.<ext>`."""
|
||
return name.split(".", 1)[0]
|
||
|
||
|
||
def answer(env: Env, job_id: str) -> None:
|
||
"""Ответ ИС для задания (синтетический успех R-10 → `NOT_IMPLEMENTED`)."""
|
||
env.answer(job_id, json.dumps({**SUCCESS_ANSWER, "file_uid": job_id}).encode())
|
||
|
||
|
||
def index_of(env: Env, event: tuple[str, str]) -> int:
|
||
assert env.events.count(event) == 1, (event, env.events)
|
||
return env.events.index(event)
|
||
|
||
|
||
def source_key(request: dict[str, Any]) -> str:
|
||
return str(request["s3"]["key"])
|
||
|
||
|
||
async def finish_all(env: Env, poll_sleep: SteppedSleep, jobs: list[str]) -> None:
|
||
"""Ответить на все задания, снять паузы, дождаться конца; у каждого задания — ровно один ack и один итог."""
|
||
for job in jobs:
|
||
if not env.local.answer_path(job).exists():
|
||
answer(env, job)
|
||
poll_sleep.open()
|
||
await asyncio.wait_for(env.worker.wait_idle(), WAIT_TIMEOUT)
|
||
for job in jobs:
|
||
assert env.acks(job) == [f"ack:{job}:1"]
|
||
[(msg_id, body)] = env.results(job)
|
||
assert msg_id == f"result:{job}:1"
|
||
assert body["status"] == "error"
|
||
assert body["error"]["code"] == "NOT_IMPLEMENTED"
|
||
assert env.worker.active_jobs == 0
|
||
|
||
|
||
# --- spec «Пять заданий подряд» -----------------------------------------------------------------------------------
|
||
async def test_five_jobs_with_four_slots(make_env: EnvFactory) -> None:
|
||
poll_sleep = SteppedSleep()
|
||
env = await make_env(SCAN_MAX_INFLIGHT=4, SCAN_TIMEOUT_SECONDS=60, sleep=poll_sleep)
|
||
requests = [env.request() for _ in range(5)]
|
||
jobs = [request["job_id"] for request in requests]
|
||
for request in requests:
|
||
await env.deliver(request)
|
||
|
||
# Четыре слота заняты: каждое из четырёх заданий записало файл и ждёт ответа ИС
|
||
await poll_sleep.wait_parked(4)
|
||
# пять ack'ов уже опубликованы — ни один слот ещё не освободился, итогов нет
|
||
assert [env.acks(job) for job in jobs] == [[f"ack:{job}:1"] for job in jobs]
|
||
assert all(env.results(job) == [] for job in jobs)
|
||
# во входящем каталоге ровно четыре файла, пятое задание даже не читало S3
|
||
assert len(env.storage.get_calls) == 4
|
||
assert len(env.hotfolder.puts) == 4
|
||
assert env.inbox() == sorted(env.hotfolder.puts)
|
||
busy = [job_of(name) for name in env.hotfolder.puts]
|
||
[waiting] = [job for job in jobs if job not in busy]
|
||
assert poll_sleep.parked == 4
|
||
assert env.worker.active_jobs == 5
|
||
|
||
# Ещё один круг опроса занятых слотов: пятое задание по-прежнему ждёт
|
||
poll_sleep.step()
|
||
await poll_sleep.wait_parked(4)
|
||
assert len(env.hotfolder.puts) == 4
|
||
assert len(env.storage.get_calls) == 4
|
||
assert waiting not in {job_of(name) for name in env.inbox()}
|
||
|
||
# Ответ ИС для одного задания: его слот освобождается, и только тогда появляется пятый файл
|
||
freed = busy[0]
|
||
answer(env, freed)
|
||
poll_sleep.step()
|
||
await poll_sleep.wait_parked(4) # три прежних задания + пятое, записавшее файл
|
||
assert [job_of(name) for name in env.hotfolder.puts] == [*busy, waiting]
|
||
assert len(env.inbox()) == 5
|
||
assert all(env.results(job) == [] for job in jobs if job != freed)
|
||
freed_result = index_of(env, ("publish", f"result:{freed}:1"))
|
||
waiting_request = requests[jobs.index(waiting)]
|
||
assert max(index_of(env, ("publish", f"ack:{job}:1")) for job in jobs) < freed_result
|
||
assert freed_result < index_of(env, ("s3_get", source_key(waiting_request)))
|
||
assert freed_result < index_of(env, ("put", f"{waiting}.pdf"))
|
||
|
||
# Все пять заданий заканчиваются конвертами
|
||
await finish_all(env, poll_sleep, jobs)
|
||
assert len(env.inbox()) == 5
|
||
|
||
|
||
async def test_single_slot_serialises_jobs(make_env: EnvFactory) -> None:
|
||
poll_sleep = SteppedSleep()
|
||
env = await make_env(SCAN_MAX_INFLIGHT=1, SCAN_TIMEOUT_SECONDS=60, sleep=poll_sleep)
|
||
requests = [env.request() for _ in range(3)]
|
||
jobs = [request["job_id"] for request in requests]
|
||
for request in requests:
|
||
await env.deliver(request)
|
||
|
||
for done in range(3):
|
||
await poll_sleep.wait_parked(1)
|
||
# в работе ровно одно задание; ack'и всех трёх — сразу
|
||
assert len(env.hotfolder.puts) == done + 1
|
||
assert len(env.storage.get_calls) == done + 1
|
||
assert len(env.inbox()) == done + 1
|
||
assert all(env.acks(job) == [f"ack:{job}:1"] for job in jobs)
|
||
current = job_of(env.hotfolder.puts[-1])
|
||
assert env.results(current) == []
|
||
answer(env, current)
|
||
poll_sleep.step()
|
||
await asyncio.wait_for(env.worker.wait_idle(), WAIT_TIMEOUT)
|
||
|
||
# Строгая очерёдность в журнале: следующий файл записан только после итога предыдущего задания
|
||
order = [job_of(name) for name in env.hotfolder.puts]
|
||
assert sorted(order) == sorted(jobs)
|
||
files_and_results = [
|
||
(kind, ref) for kind, ref in env.events if kind == "put" or (kind == "publish" and ref.startswith("result:"))
|
||
]
|
||
expected: list[tuple[str, str]] = []
|
||
for job in order:
|
||
expected += [("put", f"{job}.pdf"), ("publish", f"result:{job}:1")]
|
||
assert files_and_results == expected
|
||
await finish_all(env, poll_sleep, jobs)
|
||
|
||
|
||
# --- остановка при очереди на слот (T12.9) ------------------------------------------------------------------------
|
||
async def test_stop_while_jobs_wait_for_slot(make_env: EnvFactory, log_lines: list[str]) -> None:
|
||
poll_sleep = SteppedSleep()
|
||
env = await make_env(SCAN_MAX_INFLIGHT=2, SCAN_TIMEOUT_SECONDS=60, sleep=poll_sleep)
|
||
tasks_before = asyncio.all_tasks()
|
||
requests = [env.request() for _ in range(5)]
|
||
for request in requests:
|
||
await env.deliver(request)
|
||
await poll_sleep.wait_parked(2)
|
||
assert env.worker.active_jobs == 5
|
||
assert len(env.hotfolder.puts) == 2 # три задания ждут слот
|
||
assert all(env.acks(request["job_id"]) == [f"ack:{request['job_id']}:1"] for request in requests)
|
||
job_tasks = asyncio.all_tasks() - tasks_before
|
||
assert len(job_tasks) == 5
|
||
|
||
# без исключения; `asyncio.timeout`, а не `wait_for`: отдельная задача-обёртка дала бы отменённым заданиям
|
||
# лишние проходы цикла, и проверка «stop() дождался заданий» стала бы слепой
|
||
async with asyncio.timeout(WAIT_TIMEOUT):
|
||
await env.worker.stop()
|
||
|
||
# Все задачи заданий завершены отменой, новых задач не осталось, итоговых конвертов нет
|
||
assert all(task.cancelled() for task in job_tasks)
|
||
assert asyncio.all_tasks() - tasks_before == set()
|
||
assert env.worker.active_jobs == 0
|
||
assert all(env.results(request["job_id"]) == [] for request in requests)
|
||
assert not [line for line in log_lines if line.startswith(("ERROR|", "CRITICAL|"))]
|
||
# После остановки файлы во входящий каталог больше не пишутся и S3 не читается, даже если цикл событий крутится
|
||
for _ in range(20):
|
||
await asyncio.sleep(0)
|
||
assert len(env.hotfolder.puts) == 2
|
||
assert len(env.inbox()) == 2
|
||
assert len(env.storage.get_calls) == 2
|
||
assert poll_sleep.parked == 0
|
||
|
||
# Слоты не утекли: после повторного запуска оба слота свободны
|
||
env.worker.start()
|
||
again = [env.request() for _ in range(3)]
|
||
for request in again:
|
||
await env.deliver(request)
|
||
await poll_sleep.wait_parked(2)
|
||
assert len(env.hotfolder.puts) == 4
|
||
assert len(env.storage.get_calls) == 4
|
||
await finish_all(env, poll_sleep, [request["job_id"] for request in again])
|