Dmitry
3fc7a954b9
Префикс /api/v1, operationId = "<tag>_<name>", чтобы Dart-клиент получил методы вроде accountsList, а не list_accounts_api_v1_accounts_get. Ошибки — application/problem+json. Auth: access-JWT на час + refresh на 30 дней, который хранится хэшем и ротируется, логин ограничен по частоте в памяти. Деньги в JSON — всегда позиционные строки (api/schemas/common.py): float в проводе потерял бы копейки, которые NUMERIC(24,10) бережёт. Тестовый harness поднимает свой Postgres через pg_ctl (pytest-postgresql), мигрирует его один раз на сессию и усекает таблицы после каждого теста.
98 lines
3.1 KiB
Python
98 lines
3.1 KiB
Python
"""Test harness: a throwaway Postgres (pytest-postgresql + pg_ctl on PATH), migrated with
|
|
Alembic once per session, tables truncated after every test."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import AsyncIterator, Iterator
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
from pytest_postgresql import factories
|
|
from pytest_postgresql.janitor import DatabaseJanitor
|
|
from sqlalchemy import text
|
|
|
|
BACKEND_DIR = Path(__file__).resolve().parent.parent
|
|
|
|
postgresql_proc = factories.postgresql_proc(port=None, unixsocketdir="/tmp")
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def database_url(postgresql_proc) -> Iterator[str]:
|
|
p = postgresql_proc
|
|
with DatabaseJanitor(
|
|
user=p.user,
|
|
host=p.host,
|
|
port=p.port,
|
|
dbname="fintracker_test",
|
|
password=p.password,
|
|
):
|
|
pw = f":{p.password}" if p.password else ""
|
|
url = f"postgresql+asyncpg://{p.user}{pw}@{p.host}:{p.port}/fintracker_test"
|
|
os.environ["DATABASE_URL"] = url
|
|
os.environ["JWT_SECRET"] = "test-secret-not-for-production-0123456789"
|
|
from fintracker.config import get_settings
|
|
|
|
get_settings.cache_clear()
|
|
yield url
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def migrated(database_url: str) -> str:
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
|
|
cfg = Config(str(BACKEND_DIR / "alembic.ini"))
|
|
cfg.set_main_option("script_location", str(BACKEND_DIR / "alembic"))
|
|
command.upgrade(cfg, "head")
|
|
return database_url
|
|
|
|
|
|
@pytest.fixture
|
|
async def app(migrated: str):
|
|
from fintracker.api import app as app_module
|
|
from fintracker.api.routers import auth as auth_router
|
|
from fintracker.db import reset_engine
|
|
|
|
auth_router._login_limiter = None # fresh rate limiter per test
|
|
application = app_module.create_app()
|
|
yield application
|
|
await _truncate_all()
|
|
await reset_engine()
|
|
|
|
|
|
async def _truncate_all() -> None:
|
|
from fintracker.db import get_engine
|
|
from fintracker.db.base import Base
|
|
|
|
tables = ", ".join(f'"{t.name}"' for t in Base.metadata.sorted_tables)
|
|
async with get_engine().begin() as conn:
|
|
await conn.execute(text(f"TRUNCATE {tables} RESTART IDENTITY CASCADE"))
|
|
|
|
|
|
@pytest.fixture
|
|
async def client(app) -> AsyncIterator[AsyncClient]:
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
|
yield c
|
|
|
|
|
|
@pytest.fixture
|
|
async def user(app) -> dict[str, str]:
|
|
from fintracker.api.security import hash_password
|
|
from fintracker.db import get_sessionmaker
|
|
from fintracker.models import AppUser
|
|
|
|
creds = {"email": "ada@example.com", "password": "correct horse battery staple"}
|
|
async with get_sessionmaker()() as session:
|
|
session.add(AppUser(email=creds["email"], password_hash=hash_password(creds["password"])))
|
|
await session.commit()
|
|
return creds
|
|
|
|
|
|
@pytest.fixture
|
|
async def auth_headers(client: AsyncClient, user: dict[str, str]) -> dict[str, str]:
|
|
r = await client.post("/api/v1/auth/login", json=user)
|
|
assert r.status_code == 200, r.text
|
|
return {"Authorization": f"Bearer {r.json()['access_token']}"}
|