k1nq
61 lines
1.6 KiB
Python
61 lines
1.6 KiB
Python
import asyncio
|
|
from typing import cast
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from faststream.rabbit import RabbitBroker
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
from app.outbox import OutboxPublisher
|
|
|
|
|
|
def make_publisher() -> OutboxPublisher:
|
|
return OutboxPublisher(
|
|
cast(async_sessionmaker[AsyncSession], object()),
|
|
cast(RabbitBroker, object()),
|
|
batch_size=1,
|
|
poll_interval=0.001,
|
|
publish_timeout=1,
|
|
)
|
|
|
|
|
|
async def test_run_stops_after_stop_event(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
publisher = make_publisher()
|
|
stop_event = asyncio.Event()
|
|
|
|
async def publish_batch(_: asyncio.Event) -> int:
|
|
stop_event.set()
|
|
return 0
|
|
|
|
monkeypatch.setattr(publisher, "publish_batch", publish_batch)
|
|
|
|
await asyncio.wait_for(publisher.run(stop_event), timeout=1)
|
|
|
|
assert stop_event.is_set()
|
|
|
|
|
|
async def test_run_recovers_from_iteration_error(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
publisher = make_publisher()
|
|
stop_event = asyncio.Event()
|
|
iteration_finished = asyncio.Event()
|
|
|
|
async def fail_iteration(_: asyncio.Event) -> int:
|
|
iteration_finished.set()
|
|
raise RuntimeError("temporary database error")
|
|
|
|
publish_batch = AsyncMock(side_effect=fail_iteration)
|
|
|
|
async def stop_after_error() -> None:
|
|
await iteration_finished.wait()
|
|
stop_event.set()
|
|
|
|
monkeypatch.setattr(publisher, "publish_batch", publish_batch)
|
|
stopper = asyncio.create_task(stop_after_error())
|
|
|
|
await asyncio.wait_for(publisher.run(stop_event), timeout=1)
|
|
await stopper
|
|
|
|
publish_batch.assert_awaited_once_with(stop_event)
|