k1nq
116 lines
3.6 KiB
Python
116 lines
3.6 KiB
Python
import asyncio
|
|
import json
|
|
from decimal import Decimal
|
|
|
|
import aio_pika
|
|
from faststream.exceptions import RejectMessage
|
|
from faststream.rabbit import Channel, RabbitBroker
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
from app.messaging import (
|
|
DEAD_QUEUE,
|
|
PAYMENTS_EXCHANGE,
|
|
PAYMENTS_QUEUE,
|
|
PAYMENTS_ROUTING_KEY,
|
|
declare_topology,
|
|
)
|
|
from app.models import Currency, OutboxEvent
|
|
from app.outbox import OutboxPublisher
|
|
from app.payments import create_payment
|
|
from app.schemas import PaymentCreate
|
|
|
|
|
|
async def wait_for_message(
|
|
queue: aio_pika.abc.AbstractQueue,
|
|
) -> aio_pika.abc.AbstractIncomingMessage:
|
|
async with asyncio.timeout(5):
|
|
while True:
|
|
message = await queue.get(fail=False)
|
|
if message is not None:
|
|
return message
|
|
await asyncio.sleep(0.05)
|
|
|
|
|
|
async def test_rejected_payment_message_is_dead_lettered(rabbitmq_url: str) -> None:
|
|
broker = RabbitBroker(
|
|
rabbitmq_url,
|
|
default_channel=Channel(publisher_confirms=True, on_return_raises=True),
|
|
)
|
|
|
|
@broker.subscriber(PAYMENTS_QUEUE, PAYMENTS_EXCHANGE)
|
|
async def reject_message(_: dict[str, str]) -> None:
|
|
raise RejectMessage()
|
|
|
|
await broker.start()
|
|
connection = await aio_pika.connect_robust(rabbitmq_url)
|
|
try:
|
|
await declare_topology(broker)
|
|
payload = {"payment_id": "00000000-0000-0000-0000-000000000001"}
|
|
await broker.publish(
|
|
payload,
|
|
queue=PAYMENTS_ROUTING_KEY,
|
|
exchange=PAYMENTS_EXCHANGE,
|
|
persist=True,
|
|
)
|
|
|
|
channel = await connection.channel()
|
|
dead_queue = await channel.declare_queue(DEAD_QUEUE.name, passive=True)
|
|
dead_message = await wait_for_message(dead_queue)
|
|
assert json.loads(dead_message.body) == payload
|
|
await dead_message.ack()
|
|
finally:
|
|
await connection.close()
|
|
await broker.stop()
|
|
|
|
|
|
async def test_outbox_publishes_to_real_payments_queue(
|
|
rabbitmq_url: str,
|
|
database: async_sessionmaker[AsyncSession],
|
|
) -> None:
|
|
broker = RabbitBroker(
|
|
rabbitmq_url,
|
|
default_channel=Channel(publisher_confirms=True, on_return_raises=True),
|
|
)
|
|
await broker.start()
|
|
connection = await aio_pika.connect_robust(rabbitmq_url)
|
|
try:
|
|
await declare_topology(broker)
|
|
channel = await connection.channel()
|
|
queue = await channel.declare_queue(PAYMENTS_QUEUE.name, passive=True)
|
|
await queue.purge()
|
|
|
|
async with database() as session:
|
|
payment = await create_payment(
|
|
session,
|
|
PaymentCreate(
|
|
amount=Decimal("14.50"),
|
|
currency=Currency.RUB,
|
|
description="Real RabbitMQ outbox test",
|
|
webhook_url="https://example.com/hook",
|
|
),
|
|
"real-rabbit-outbox",
|
|
)
|
|
|
|
publisher = OutboxPublisher(
|
|
database,
|
|
broker,
|
|
batch_size=10,
|
|
poll_interval=0.01,
|
|
publish_timeout=5,
|
|
)
|
|
assert await publisher.publish_batch() == 1
|
|
|
|
message = await queue.get(timeout=5, fail=True)
|
|
assert json.loads(message.body) == {"payment_id": str(payment.id)}
|
|
await message.ack()
|
|
async with database() as session:
|
|
event = await session.scalar(
|
|
select(OutboxEvent).where(OutboxEvent.aggregate_id == payment.id)
|
|
)
|
|
assert event is not None
|
|
assert event.published_at is not None
|
|
finally:
|
|
await connection.close()
|
|
await broker.stop()
|