Aidar a1d4r

Кейс: интеграционный тест на отправку email через taskiq + MailHog

Сценарий: при регистрации пользователя приложение отправляет welcome-письмо. Сама отправка вынесена в taskiq-задачу, чтобы не блокировать HTTP-ответ. Нужно написать интеграционный тест, который убеждается, что после POST /register письмо реально доходит до получателя.

Типичная первая попытка — поднять три контейнера (rabbitmq, mailhog, worker taskiq) и натравить на них pytest. И почти всегда тест в таком виде не взлетает или моргает. Разберём почему и как это правильно собрать.

Sentry middleware for Taskiq

A drop-in Taskiq middleware that reports background tasks to Sentry the same way Sentry's own integrations for celery, arq, rq and huey do.

Taskiq has no official Sentry integration. Wrapping task execution in sentry_sdk.start_span(...) looks right but silently breaks: in sentry-sdk >= 2.x, a span opened without an active transaction is an orphan and gets dropped, so tasks never show up in Performance / Profiling.

	import contextlib

	from collections.abc import AsyncIterator

	import psycopg
	import pytest
	import pytest_alembic

	from filelock import FileLock
	from pytest_alembic.config import Config

	from typing import Any, TypeVar

	import sqlalchemy as sa

	from fastapi.encoders import jsonable_encoder
	from pydantic import BaseModel, TypeAdapter
	from sqlalchemy import Dialect
	from sqlalchemy.dialects.postgresql import JSONB
	from sqlalchemy.sql.type_api import TypeEngine

	# Source: https://gist.github.com/usr-ein/c42d98abca3cb4632ab0c2c6aff8c88a

	################################
	# PYTHON-BASE
	# Sets up all our shared environment variables
	################################
	FROM python:3.11-slim as python-base

	# python
	ENV PYTHONUNBUFFERED=1 \