Тестирование¶
102 тест-файла, ~1132 теста. Фреймворк: pytest + pytest-asyncio (auto mode). БД: SQLite in-memory (unit) / PostgreSQL 15 (integration). Минимальное покрытие: 40%.
Быстрый старт¶
# Все unit-тесты (SQLite, ~52s)
pytest tests/ -m "not integration and not slow" -v --tb=short
# С покрытием
pytest tests/ --cov --cov-report=term-missing --cov-fail-under=40
# Один файл
pytest tests/test_prompts.py -v
# Один класс / метод
pytest tests/test_ai_engine.py::TestResponseGuard -v
# Integration (требует PostgreSQL + Redis)
pytest tests/ -m integration -v
Конфигурация pytest¶
[tool.pytest.ini_options]
asyncio_mode = "auto" # Автодетект async-тестов
asyncio_default_fixture_loop_scope = "function" # Новый event loop на тест
testpaths = ["tests"]
timeout = 30 # Дефолтный таймаут
Покрытие¶
[tool.coverage.run]
source = ["app", "core", "telegram", "db", "api", "config"]
omit = ["*/migrations/*", "*/alembic/*", "telegram/emulator_*",
"telegram/auto_registrar*", "telegram/goip_*", ...]
Маркеры¶
| Маркер | Описание | CI job |
|---|---|---|
unit |
Без внешних сервисов | Unit Tests (SQLite) |
integration |
Требуют PostgreSQL + Redis | Integration Tests (PG) |
slow |
Длительные тесты | Исключены из unit |
ai_simulation |
Claude CLI end-to-end | Ручной запуск |
timeout |
Переопределение таймаута | @pytest.mark.timeout(N) |
Основные фикстуры¶
db_session --- синхронный SQLite + AsyncSessionAdapter¶
Основная DB-фикстура: синхронный sqlite:// engine с StaticPool, все таблицы создаются автоматически. AsyncSessionAdapter сохраняет await-совместимый API.
@pytest_asyncio.fixture
async def db_session():
engine = create_engine("sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool)
event.listen(engine, "connect", _register_sqlite_functions)
Base.metadata.create_all(engine)
session = sessionmaker(engine, expire_on_commit=False)()
adapter = AsyncSessionAdapter(session)
try:
yield adapter
finally:
await adapter.close()
Base.metadata.drop_all(engine)
engine.dispose()
Другие фикстуры¶
| Фикстура | Scope | Назначение |
|---|---|---|
db_session |
function | SQLite + AsyncSessionAdapter |
repo |
function | Repository(db_session) |
memory_dir |
function | Временная директория для JSON-памяти |
client_factory |
function | Создание тестового Client |
account_factory |
function | Создание тестового TgAccount |
_dispose_module_engine |
session | Предотвращает hang asyncpg pool |
SQLite <--> PostgreSQL совместимость¶
Автоматические хуки в conftest.py:
JSONBкомпилируется какJSONдля SQLitegreatest(a, b)регистрируется какmax(a, b)StaticPoolдержит один shared in-memory databaseAsyncSessionAdapter--- async-friendly вызовы поверх syncSession
Структура тестов по модулям¶
| Модуль | Файлы | ~Тестов | Описание |
|---|---|---|---|
| AI Engine | test_ai_engine.py |
61 | Парсинг, SDK/CLI, response guard |
| Промпты | test_prompts.py |
127 | Контракты секций, anti-injection, forbidden phrases |
| Decision Engine | test_decision_engine.py |
66 | Рабочие часы, лимиты, graduated logic |
| API | test_api_*.py (7 файлов) |
71 | FastAPI endpoints |
| Telegram | test_message_sender.py и др. |
22 | Сплиттинг, задержки, нормализация |
| Tools | test_tool_executor.py |
11 | Tool parsing, JSON валидация |
| Workflows | test_workflow_*.py (4 файла) |
26 | API v2, модели, resolver |
| Resilience | 3 файла | 13 | Config, retry, Redis auth |
| RAG | 5 файлов | ~20 | Smoke/contract тесты |
Паттерны тестирования¶
Async-тесты¶
# asyncio_mode="auto" --- async def автоматически детектится
class TestSomething:
async def test_example(self, db_session):
repo = Repository(db_session)
result = await repo.get_something()
assert result is not None
Мокирование AI¶
class TestAIEngine:
async def test_process_message(self, engine):
mock_response = '{"response": "text", "actions": []}'
with patch.object(engine, "_query_claude",
new_callable=AsyncMock,
return_value=(mock_response, None)):
result = await engine.process_message(client_id=1, ...)
assert result.parse_success is True
Freeze time¶
from freezegun import freeze_time
@freeze_time("2026-01-05 08:00:00", tz_offset=0) # Mon Kyiv 10:00
def test_working_hours(self):
assert DecisionEngine().is_working_hours() is True
CI/CD Pipeline¶
graph TD
SQ[1. Strict Quality Gates<br/>ruff, mypy, compile] --> RG[3. Resilience Guards<br/>redis, db, prompts]
SQ --> UT[4. Unit Tests<br/>SQLite, ~52s]
SQ --> IT[5. Integration Tests<br/>PostgreSQL, ~1m12s]
SQ --> DB[6. Docker Build<br/>Dockerfile + Worker]
SR[2. Soft Quality Report<br/>telegram, admin_bot] -.->|не блокирует| SQ
RG --> PASS[CI Pass]
UT --> PASS
IT --> PASS
DB --> PASS
6 CI Jobs¶
| # | Job | Блокирует | Время | Проверяет |
|---|---|---|---|---|
| 1 | Strict Quality Gates | Да | ~30s | ruff, mypy, compileall (core/db/api) |
| 2 | Soft Quality Report | Нет | ~20s | ruff+mypy для telegram/admin_bot/workers |
| 3 | Resilience Guards | Да | ~1-2m | docker compose, resilience, prompt contracts |
| 4 | Unit Tests (SQLite) | Да | ~52s | 440+ тестов, coverage >= 40% |
| 5 | Integration Tests (PG) | Да | ~1m12s | alembic upgrade head + pytest -m integration |
| 6 | Docker Build | Да | ~59s | Сборка main + worker images |
Quality Baseline
.github/quality-baseline.json задает максимум ruff (26) и mypy (65) ошибок. CI падает при превышении. Baseline может только снижаться.
Lint¶
# Ruff --- все ошибки
ruff check .
# Ruff --- только fatal
ruff check . --select E9,F63,F7,F82
# Ruff autofix
ruff check . --fix --unsafe-fixes
# Mypy (strict scope)
mypy core db api --ignore-missing-imports
# Format check
ruff format --check .
Dev-зависимости¶
| Пакет | Версия | Назначение |
|---|---|---|
| pytest-asyncio | >= 0.24 | Async test support |
| pytest-mock | >= 3.14 | MagicMock, AsyncMock |
| pytest-timeout | >= 2.3.1 | Таймауты тестов |
| freezegun | >= 1.4.0 | Фриз времени |
| aiosqlite | >= 0.20 | SQLite async (file-backed API fixtures) |
| ruff | >= 0.8.0 | Lint |
| mypy | >= 1.13.0 | Type checking |
Добавление нового теста¶
- Создать
tests/test_<module>.py - Использовать
db_sessionдля unit-тестов с БД - Использовать
memory_dirдля тестов с памятью - Мокировать
_query_claudeдля AI-тестов (без реальных API-вызовов) - Пометить
@pytest.mark.integrationесли нужен PostgreSQL - Таймаут <= 30s (или явно
@pytest.mark.timeout(N))