Проект 2. Эталонное решение

"""Конфигурация из переменных окружения с проверкой при старте.

Ключевое решение: неверная конфигурация останавливает приложение
немедленно, а не в момент первого запроса. Сервис, стартовавший
с испорченной настройкой, выглядит здоровым и отвечает ошибками —
это худший из возможных исходов.
"""
from __future__ import annotations

import os

from pydantic import Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict

LOG_LEVELS = ("debug", "info", "warning", "error")
ENV_PREFIX = "EVENTAPI_"


class UnknownSettingError(ValueError):
    """Переменная с нашим префиксом, которой не соответствует ни одно поле."""


class Settings(BaseSettings):
    """Все настройки сервиса. Имена соответствуют EVENTAPI_*."""

    model_config = SettingsConfigDict(
        env_prefix=ENV_PREFIX,
        env_file=None,          # файл .env намеренно не читается: см. README
        extra="forbid",         # опечатка в имени переменной — ошибка, не молчание
    )

    host: str = "0.0.0.0"       # noqa: S104 — внутри container'а иначе недоступно
    port: int = Field(default=8000, ge=1, le=65535)
    log_level: str = "info"
    max_events: int = Field(default=10_000, ge=1)
    max_body_bytes: int = Field(default=64 * 1024, ge=1)
    shutdown_grace_seconds: float = Field(default=25.0, ge=0)
    ready_after_seconds: float = Field(default=0.0, ge=0)

    @field_validator("log_level")
    @classmethod
    def _known_level(cls, value: str) -> str:
        normalized = value.strip().lower()
        if normalized not in LOG_LEVELS:
            raise ValueError(
                f"log_level: ожидалось одно из {', '.join(LOG_LEVELS)}, "
                f"получено {value!r}")
        return normalized

    @field_validator("host")
    @classmethod
    def _not_loopback_by_accident(cls, value: str) -> str:
        # 127.0.0.1 внутри container'а делает сервис недоступным снаружи.
        # Это самая частая ошибка проекта, поэтому она не запрещена,
        # а помечена: значение остаётся, предупреждение выводит main.
        return value.strip()


def check_env(env: dict[str, str] | None = None) -> list[str]:
    """Найти переменные EVENTAPI_*, которым не соответствует ни одно поле.

    Нужна отдельно: extra="forbid" отвергает лишние ключи при явной
    передаче, но НЕ ловит опечатку в имени переменной окружения —
    pydantic-settings такую переменную просто не видит. Проверено
    тестом test_unknown_variable_is_rejected.
    """
    env = os.environ if env is None else env
    known = {f"{ENV_PREFIX}{name.upper()}" for name in Settings.model_fields}
    return sorted(
        name for name in env
        if name.startswith(ENV_PREFIX) and name.upper() not in known
    )


def load(env: dict[str, str] | None = None) -> Settings:
    """Собрать настройки и отвергнуть опечатки в именах переменных."""
    unknown = check_env(env)
    if unknown:
        raise UnknownSettingError(
            "неизвестные переменные окружения: " + ", ".join(unknown)
            + f"; известны: {', '.join(sorted(Settings.model_fields))}")
    return Settings()

assert Settings().log_level == "info", "pydantic опечатку не заметил"
with pytest.raises(UnknownSettingError) as exc:
    load()

$ docker run --rm -e EVENTAPI_LOG_LEVE=info eventapi
{"ts": "...", "level": "info", "logger": "uvicorn.error", "message": "Uvicorn running on http://0.0.0.0:8000"}
        ← сервис работает, опечатка проигнорирована

$ docker run --rm -e EVENTAPI_LOG_LEVE=info eventapi; echo "код: $?"
app.settings.UnknownSettingError: неизвестные переменные окружения: EVENTAPI_LOG_LEVE;
известны: host, log_level, max_body_bytes, max_events, port, ready_after_seconds,
shutdown_grace_seconds
код: 1

"""Модели запросов и ответов."""
from __future__ import annotations

from datetime import datetime, timezone
from typing import Annotated, Literal

from pydantic import BaseModel, ConfigDict, Field, field_validator

Method = Literal["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"]


class EventIn(BaseModel):
    """Событие, приходящее от клиента."""

    model_config = ConfigDict(extra="forbid")

    path: Annotated[str, Field(min_length=1, max_length=2048)]
    method: Method
    status: Annotated[int, Field(ge=100, le=599)]
    duration_ms: Annotated[float, Field(ge=0, le=3_600_000)]
    ts: datetime | None = None

    @field_validator("path")
    @classmethod
    def _path_starts_with_slash(cls, value: str) -> str:
        if not value.startswith("/"):
            raise ValueError("path должен начинаться с /")
        return value

    @field_validator("ts")
    @classmethod
    def _aware(cls, value: datetime | None) -> datetime | None:
        if value is None:
            return None
        # Наивное время неоднозначно: две записи из разных зон
        # окажутся неотличимы. Приводим к UTC явно.
        if value.tzinfo is None:
            return value.replace(tzinfo=timezone.utc)
        return value.astimezone(timezone.utc)


class Event(EventIn):
    """Сохранённое событие: время проставлено, идентификатор присвоен."""

    id: int
    ts: datetime


class EventsPage(BaseModel):
    total: int
    returned: int
    items: list[Event]


class StatRow(BaseModel):
    key: str
    count: int
    share: float


class Stats(BaseModel):
    group_by: str
    total: int
    matched: int
    rows: list[StatRow]


class Health(BaseModel):
    status: Literal["alive"]
    version: str


class Ready(BaseModel):
    ready: bool
    checks: dict[str, bool]
    reason: str | None = None

"""Хранилище событий.

Интерфейс отделён от реализации намеренно: в проекте 3 хранилище
в памяти заменяется на PostgreSQL, и заменяться должен один класс,
а не обработчики запросов.
"""
from __future__ import annotations

import asyncio
from collections import Counter, deque
from datetime import datetime, timezone
from typing import Protocol

from .models import Event, EventIn, StatRow, Stats


class Storage(Protocol):
    """Контракт хранилища. Реализация в проекте 3 будет другой."""

    async def add(self, event: EventIn) -> Event: ...

    async def list(self, field: str | None, value: str | None,
                   limit: int, offset: int) -> tuple[int, list[Event]]: ...

    async def stats(self, group_by: str, top: int,
                    field: str | None, value: str | None) -> Stats: ...

    async def count(self) -> int: ...

    async def ping(self) -> bool: ...


GROUPABLE = ("status", "method", "path")


class MemoryStorage:
    """Хранилище в памяти с ограничением размера.

    deque(maxlen=...) вытесняет старые записи автоматически. Без предела
    сервис, принимающий события, рано или поздно исчерпает память —
    и отказ придёт от OOM killer, без записи в лог приложения.
    """

    def __init__(self, max_events: int) -> None:
        self._events: deque[Event] = deque(maxlen=max_events)
        self._next_id = 1
        self._lock = asyncio.Lock()

    async def add(self, event: EventIn) -> Event:
        async with self._lock:
            stored = Event(
                id=self._next_id,
                ts=event.ts or datetime.now(timezone.utc),
                **event.model_dump(exclude={"ts"}),
            )
            self._next_id += 1
            self._events.append(stored)
            return stored

    def _select(self, field: str | None, value: str | None) -> list[Event]:
        items = list(self._events)
        if field is None or value is None:
            return items
        return [e for e in items
                if str(getattr(e, field, None)) == value]

    async def list(self, field: str | None, value: str | None,
                   limit: int, offset: int) -> tuple[int, list[Event]]:
        async with self._lock:
            selected = self._select(field, value)
            page = selected[offset:offset + limit]
            return len(selected), page

    async def stats(self, group_by: str, top: int,
                    field: str | None, value: str | None) -> Stats:
        async with self._lock:
            total = len(self._events)
            selected = self._select(field, value)
            counts: Counter[str] = Counter(
                str(getattr(e, group_by)) for e in selected)
            matched = len(selected)
            # Вторая ось сортировки — по ключу: при равных счётчиках
            # порядок иначе зависит от порядка вставки и невоспроизводим.
            rows = [
                StatRow(key=k, count=n,
                        share=round(n / matched * 100, 2) if matched else 0.0)
                for k, n in sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))[:top]
            ]
            return Stats(group_by=group_by, total=total,
                         matched=matched, rows=rows)

    async def count(self) -> int:
        async with self._lock:
            return len(self._events)

    async def ping(self) -> bool:
        return True

"""Структурированные логи в stdout.

Записи в JSON по одной на строку: их читает сборщик логов, а не человек.
Идентификатор запроса берётся из контекстной переменной, поэтому его
не приходится передавать через все вызовы.
"""
from __future__ import annotations

import json
import logging
import re
import sys
from contextvars import ContextVar
from datetime import datetime, timezone

request_id_var: ContextVar[str] = ContextVar("request_id", default="-")

# Идентификатор приходит из заголовка, то есть от клиента, и попадает
# и в логи, и в заголовок ответа. Без ограничения длины и набора символов
# он становится каналом для порчи журнала: перевод строки внутри значения
# превращает одну запись в две, вторая — под контролем отправителя.
_ID_ALLOWED = re.compile(r"[^A-Za-z0-9._-]")
ID_MAX_LEN = 64


def sanitize_request_id(raw: str | None) -> str | None:
    """Оставить безопасную часть идентификатора либо отвергнуть его."""
    if not raw:
        return None
    cleaned = _ID_ALLOWED.sub("", raw)[:ID_MAX_LEN]
    return cleaned or None

# Поля LogRecord, которые уже вошли в запись или не нужны в выводе.
_STANDARD = {
    "args", "asctime", "created", "exc_info", "exc_text", "filename",
    "funcName", "levelname", "levelno", "lineno", "module", "msecs",
    "message", "msg", "name", "pathname", "process", "processName",
    "relativeCreated", "stack_info", "thread", "threadName", "taskName",
    # uvicorn кладёт в запись копию сообщения с ANSI-кодами раскраски.
    # В структурированном логе это мусор: цвет предназначен терминалу.
    "color_message",
}


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "ts": datetime.fromtimestamp(record.created, timezone.utc)
                          .isoformat(timespec="milliseconds"),
            "level": record.levelname.lower(),
            "logger": record.name,
            "message": record.getMessage(),
            "request_id": request_id_var.get(),
        }
        for key, value in record.__dict__.items():
            if key not in _STANDARD and not key.startswith("_"):
                payload[key] = value
        if record.exc_info:
            payload["exception"] = self.formatException(record.exc_info)
        return json.dumps(payload, ensure_ascii=False, default=str)


def configure(level: str) -> None:
    """Настроить корневой логгер и подчинить ему логгеры uvicorn."""
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JsonFormatter())

    root = logging.getLogger()
    root.handlers.clear()
    root.addHandler(handler)
    root.setLevel(level.upper())

    # Без этого uvicorn пишет своим форматом, и в потоке оказываются
    # строки двух видов — сборщик логов разберёт только одну половину.
    for name in ("uvicorn", "uvicorn.error", "uvicorn.access"):
        logger = logging.getLogger(name)
        logger.handlers.clear()
        logger.propagate = True

INFO:     Started server process [3576249]
INFO:     Waiting for application startup.
{"ts": "...", "level": "info", "logger": "eventapi", ...}

"color_message": "Started server process [\u001b[36m%d\u001b[0m]"

"""HTTP-сервис приёма событий и статистики по ним."""
from __future__ import annotations

import asyncio
import logging
import time
import uuid
from contextlib import asynccontextmanager
from typing import AsyncIterator, Literal

from fastapi import Depends, FastAPI, HTTPException, Query, Request, Response, status
from fastapi.responses import JSONResponse

from . import __version__
from .logging_config import configure, request_id_var, sanitize_request_id
from .models import Event, EventsPage, Health, Ready, Stats
from .models import EventIn
from .settings import Settings, load
from .storage import GROUPABLE, MemoryStorage, Storage

log = logging.getLogger("eventapi")


class AppState:
    """Состояние процесса, влияющее на пробы."""

    def __init__(self) -> None:
        self.storage: Storage | None = None
        self.settings: Settings | None = None
        self.started_at: float = 0.0
        self.shutting_down: bool = False
        self.inflight: int = 0


state = AppState()


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """Инициализация и мягкое завершение.

    lifespan заменяет устаревшие @app.on_event: обе половины лежат
    в одном месте, и завершение выполняется даже при ошибке старта.
    """
    settings = app.state.settings
    state.settings = settings
    state.storage = MemoryStorage(max_events=settings.max_events)
    state.started_at = time.monotonic()
    state.shutting_down = False

    if settings.host in ("127.0.0.1", "localhost", "::1"):
        log.warning("host=%s: снаружи container'а сервис будет недоступен",
                    settings.host, extra={"event": "suspicious_host"})

    log.info("сервис запущен", extra={
        "event": "startup", "version": __version__,
        "max_events": settings.max_events,
    })
    try:
        yield
    finally:
        state.shutting_down = True
        log.info("получен сигнал завершения, дожидаюсь активных запросов",
                 extra={"event": "shutdown_begin", "inflight": state.inflight})

        deadline = time.monotonic() + settings.shutdown_grace_seconds
        while state.inflight > 0 and time.monotonic() < deadline:
            await asyncio.sleep(0.05)

        log.info("завершение", extra={
            "event": "shutdown_done",
            "inflight": state.inflight,
            "drained": state.inflight == 0,
        })


def create_app(settings: Settings | None = None) -> FastAPI:
    # load(), а не Settings(): только load проверяет ИМЕНА переменных.
    # Settings() опечатку в имени не видит — pydantic-settings такую
    # переменную просто не читает, и сервис стартует с умолчанием.
    settings = settings or load()
    # Настройка логирования выполняется ЗДЕСЬ, а не в lifespan: uvicorn
    # печатает первые строки до входа в lifespan, и они уходили бы
    # в другом формате. Проверено запуском — см. test_uvicorn_logs_are_json.
    configure(settings.log_level)
    app = FastAPI(
        title="eventapi",
        version=__version__,
        lifespan=lifespan,
        description="Приём событий и статистика по ним.",
    )
    app.state.settings = settings
    register(app, settings)
    return app


def get_storage() -> Storage:
    if state.storage is None:
        raise HTTPException(status_code=503, detail="хранилище не готово")
    return state.storage


def register(app: FastAPI, settings: Settings) -> None:

    @app.middleware("http")
    async def request_context(request: Request, call_next):
        """Идентификатор запроса, учёт активных запросов, журнал доступа."""
        rid = sanitize_request_id(request.headers.get("x-request-id")) \
            or uuid.uuid4().hex[:12]
        token = request_id_var.set(rid)
        state.inflight += 1
        started = time.perf_counter()
        try:
            response = await call_next(request)
        except Exception:
            log.exception("необработанная ошибка", extra={
                "event": "unhandled", "path": request.url.path})
            response = JSONResponse(
                status_code=500, content={"detail": "внутренняя ошибка"})
        finally:
            state.inflight -= 1
            request_id_var.reset(token)
        elapsed = (time.perf_counter() - started) * 1000
        response.headers["x-request-id"] = rid
        log.info("запрос обработан", extra={
            "event": "access",
            "method": request.method,
            "path": request.url.path,
            "status": response.status_code,
            "duration_ms": round(elapsed, 2),
            "request_id": rid,
        })
        return response

    # ── Пробы ────────────────────────────────────────────────────────
    @app.get("/healthz", response_model=Health, tags=["probes"])
    async def healthz() -> Health:
        """Жив ли процесс. Зависимости НЕ проверяются намеренно:
        отказ зависимости не должен приводить к перезапуску."""
        return Health(status="alive", version=__version__)

    @app.get("/readyz", response_model=Ready, tags=["probes"])
    async def readyz(response: Response) -> Ready:
        """Готов ли обслуживать: зависимости, прогрев, завершение."""
        checks = {
            "storage": state.storage is not None and await state.storage.ping(),
            "warmed_up": (time.monotonic() - state.started_at
                          >= settings.ready_after_seconds),
            "not_shutting_down": not state.shutting_down,
        }
        ready = all(checks.values())
        reason = None if ready else ", ".join(k for k, v in checks.items() if not v)
        if not ready:
            response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
        return Ready(ready=ready, checks=checks, reason=reason)

    @app.get("/startupz", tags=["probes"])
    async def startupz(response: Response) -> dict[str, object]:
        """Завершена ли инициализация. Отделена от readiness, чтобы
        медленный старт не требовал большого initialDelaySeconds."""
        done = state.storage is not None
        if not done:
            response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
        return {"initialized": done,
                "uptime_seconds": round(time.monotonic() - state.started_at, 3)}

    # ── Данные ───────────────────────────────────────────────────────
    @app.post("/events", response_model=Event,
              status_code=status.HTTP_201_CREATED, tags=["events"])
    async def create_event(event: EventIn,
                           storage: Storage = Depends(get_storage)) -> Event:
        if state.shutting_down:
            raise HTTPException(status_code=503, detail="сервис завершается")
        stored = await storage.add(event)
        log.info("событие принято", extra={
            "event": "event_created", "event_id": stored.id,
            "status_code": stored.status})
        return stored

    @app.get("/events", response_model=EventsPage, tags=["events"])
    async def list_events(
        field: Literal["status", "method", "path"] | None = None,
        value: str | None = None,
        limit: int = Query(default=50, ge=1, le=1000),
        offset: int = Query(default=0, ge=0),
        storage: Storage = Depends(get_storage),
    ) -> EventsPage:
        if (field is None) != (value is None):
            raise HTTPException(
                status_code=422,
                detail="field и value задаются только вместе")
        total, items = await storage.list(field, value, limit, offset)
        return EventsPage(total=total, returned=len(items), items=items)

    @app.get("/stats", response_model=Stats, tags=["events"])
    async def get_stats(
        group_by: Literal["status", "method", "path"] = "status",
        top: int = Query(default=10, ge=1, le=100),
        field: Literal["status", "method", "path"] | None = None,
        value: str | None = None,
        storage: Storage = Depends(get_storage),
    ) -> Stats:
        if (field is None) != (value is None):
            raise HTTPException(
                status_code=422,
                detail="field и value задаются только вместе")
        if group_by not in GROUPABLE:
            raise HTTPException(status_code=422, detail="неизвестное поле")
        return await storage.stats(group_by, top, field, value)


app = create_app()

"""eventapi — приём событий и статистика по ним."""

__version__ = "1.0.0"

"""Общие приспособления тестов."""
from __future__ import annotations

import pytest
from fastapi.testclient import TestClient

from app.main import create_app
from app.settings import Settings


@pytest.fixture
def settings() -> Settings:
    return Settings(max_events=100, shutdown_grace_seconds=1.0)


@pytest.fixture
def client(settings: Settings):
    """Клиент с поднятым lifespan: без него состояние не инициализируется."""
    app = create_app(settings)
    with TestClient(app) as c:
        yield c


@pytest.fixture
def sample_events() -> list[dict[str, object]]:
    return [
        {"path": "/api/items", "method": "GET", "status": 200, "duration_ms": 12},
        {"path": "/api/items", "method": "GET", "status": 200, "duration_ms": 8},
        {"path": "/api/missing", "method": "GET", "status": 404, "duration_ms": 3},
        {"path": "/api/items", "method": "POST", "status": 500, "duration_ms": 240},
        {"path": "/health", "method": "GET", "status": 200, "duration_ms": 1},
    ]

"""Конфигурация проверяется при старте, а не при первом запросе."""
from __future__ import annotations

import pytest
from pydantic import ValidationError

from app.main import create_app
from app.settings import Settings, UnknownSettingError, check_env, load


def test_defaults_are_usable_in_container():
    s = Settings()
    assert s.host == "0.0.0.0", "иначе сервис недоступен снаружи container'а"
    assert s.port == 8000


def test_env_is_read(monkeypatch):
    monkeypatch.setenv("EVENTAPI_PORT", "9000")
    monkeypatch.setenv("EVENTAPI_LOG_LEVEL", "DEBUG")
    s = Settings()
    assert s.port == 9000
    assert s.log_level == "debug", "уровень приводится к нижнему регистру"


@pytest.mark.parametrize("value", ["0", "70000", "не число"])
def test_bad_port_fails_at_startup(monkeypatch, value):
    monkeypatch.setenv("EVENTAPI_PORT", value)
    with pytest.raises(ValidationError):
        Settings()


def test_unknown_variable_is_rejected(monkeypatch):
    """Опечатка в имени переменной не должна проходить молча.

    extra="forbid" здесь НЕ помогает: pydantic-settings не видит
    переменную, которой не соответствует поле, и молча её игнорирует.
    Поэтому проверка сделана отдельной функцией.
    """
    monkeypatch.setenv("EVENTAPI_LOG_LEVE", "info")
    assert Settings().log_level == "info", "pydantic опечатку не заметил"
    with pytest.raises(UnknownSettingError) as exc:
        load()
    assert "EVENTAPI_LOG_LEVE" in str(exc.value)


def test_application_itself_rejects_typo(monkeypatch):
    """Проверку должно выполнять приложение, а не только функция load().

    Первая редакция создавала Settings() прямо в create_app: load() был
    написан, покрыт тестом выше — и не вызывался. Тест проходил,
    а образ с опечаткой в имени переменной спокойно стартовал.
    Этот тест входит в приложение той же дверью, что и точка входа.
    """
    monkeypatch.setenv("EVENTAPI_LOG_LEVE", "info")
    with pytest.raises(UnknownSettingError):
        create_app()


def test_check_env_is_quiet_on_clean_environment():
    assert check_env({"EVENTAPI_PORT": "9000", "PATH": "/bin"}) == []


def test_check_env_lists_every_typo():
    unknown = check_env({"EVENTAPI_PROT": "1", "EVENTAPI_LOGLEVEL": "info"})
    assert unknown == ["EVENTAPI_LOGLEVEL", "EVENTAPI_PROT"]


def test_load_returns_settings_when_environment_is_clean(monkeypatch):
    monkeypatch.setenv("EVENTAPI_PORT", "9001")
    assert load().port == 9001


def test_bad_log_level_names_allowed_values(monkeypatch):
    monkeypatch.setenv("EVENTAPI_LOG_LEVEL", "громко")
    with pytest.raises(ValidationError) as exc:
        Settings()
    assert "info" in str(exc.value)


def test_negative_max_events_rejected(monkeypatch):
    monkeypatch.setenv("EVENTAPI_MAX_EVENTS", "0")
    with pytest.raises(ValidationError):
        Settings()

"""Endpoint'ы: валидация, страницы, статистика."""
from __future__ import annotations

import pytest


def post_all(client, events):
    return [client.post("/events", json=e) for e in events]


def test_create_event_returns_201_and_id(client):
    r = client.post("/events", json={
        "path": "/a", "method": "GET", "status": 200, "duration_ms": 1.5})
    assert r.status_code == 201
    body = r.json()
    assert body["id"] == 1
    assert body["ts"], "время проставляется сервером"


def test_ids_are_sequential(client, sample_events):
    ids = [r.json()["id"] for r in post_all(client, sample_events)]
    assert ids == [1, 2, 3, 4, 5]


@pytest.mark.parametrize("payload,field", [
    ({"path": "a", "method": "GET", "status": 200, "duration_ms": 1}, "path"),
    ({"path": "/a", "method": "FETCH", "status": 200, "duration_ms": 1}, "method"),
    ({"path": "/a", "method": "GET", "status": 99, "duration_ms": 1}, "status"),
    ({"path": "/a", "method": "GET", "status": 200, "duration_ms": -1}, "duration_ms"),
    ({"path": "/a", "method": "GET", "status": 200}, "duration_ms"),
])
def test_invalid_payload_is_422(client, payload, field):
    r = client.post("/events", json=payload)
    assert r.status_code == 422
    assert field in r.text


def test_extra_field_is_rejected(client):
    """Лишнее поле — вероятная опечатка, а не расширение."""
    r = client.post("/events", json={
        "path": "/a", "method": "GET", "status": 200,
        "duration_ms": 1, "durationms": 2})
    assert r.status_code == 422


def test_naive_timestamp_becomes_utc(client):
    r = client.post("/events", json={
        "path": "/a", "method": "GET", "status": 200,
        "duration_ms": 1, "ts": "2026-08-04T10:00:00"})
    assert r.status_code == 201
    assert r.json()["ts"].endswith("+00:00") or r.json()["ts"].endswith("Z")


def test_list_returns_all_by_default(client, sample_events):
    post_all(client, sample_events)
    body = client.get("/events").json()
    assert body["total"] == 5
    assert body["returned"] == 5


def test_list_filters_by_field(client, sample_events):
    post_all(client, sample_events)
    body = client.get("/events", params={"field": "method", "value": "GET"}).json()
    assert body["total"] == 4


def test_list_paginates(client, sample_events):
    post_all(client, sample_events)
    body = client.get("/events", params={"limit": 2, "offset": 3}).json()
    assert body["total"] == 5, "total — про весь набор, не про страницу"
    assert body["returned"] == 2


def test_field_without_value_is_422(client):
    r = client.get("/events", params={"field": "method"})
    assert r.status_code == 422


def test_unknown_filter_field_is_422(client):
    r = client.get("/events", params={"field": "нетакого", "value": "1"})
    assert r.status_code == 422


def test_limit_above_maximum_is_422(client):
    assert client.get("/events", params={"limit": 100000}).status_code == 422


def test_stats_group_by_status(client, sample_events):
    post_all(client, sample_events)
    body = client.get("/stats").json()
    assert body["group_by"] == "status"
    assert body["matched"] == 5
    assert body["rows"][0] == {"key": "200", "count": 3, "share": 60.0}


def test_stats_respects_top(client, sample_events):
    post_all(client, sample_events)
    body = client.get("/stats", params={"top": 1}).json()
    assert len(body["rows"]) == 1


def test_stats_with_filter_separates_total_and_matched(client, sample_events):
    post_all(client, sample_events)
    body = client.get("/stats", params={
        "group_by": "path", "field": "method", "value": "GET"}).json()
    assert body["total"] == 5
    assert body["matched"] == 4


def test_stats_order_is_deterministic_on_ties(client):
    """При равных счётчиках порядок задаётся ключом, а не вставкой."""
    for method in ("POST", "GET", "PUT"):
        client.post("/events", json={
            "path": "/x", "method": method, "status": 200, "duration_ms": 1})
    keys = [row["key"] for row in
            client.get("/stats", params={"group_by": "method"}).json()["rows"]]
    assert keys == ["GET", "POST", "PUT"]


def test_storage_evicts_oldest_beyond_limit(client):
    """Предел размера обязателен: иначе отказ придёт от OOM killer."""
    for i in range(150):
        client.post("/events", json={
            "path": f"/p{i}", "method": "GET", "status": 200, "duration_ms": 1})
    body = client.get("/events", params={"limit": 1}).json()
    assert body["total"] == 100, "хранилище ограничено max_events"


def test_request_id_is_echoed(client):
    r = client.get("/healthz", headers={"x-request-id": "check-123"})
    assert r.headers["x-request-id"] == "check-123"


@pytest.mark.parametrize("raw,expected", [
    ("plain-123", "plain-123"),
    ("with space", "withspace"),
    ("a" * 200, "a" * 64),
    ("../../etc/passwd", "....etcpasswd"),
])
def test_request_id_is_sanitized(client, raw, expected):
    """Значение приходит от клиента и попадает в журнал — его чистят."""
    r = client.get("/healthz", headers={"x-request-id": raw})
    assert r.headers["x-request-id"] == expected


def test_request_id_with_newline_cannot_forge_log_line(client):
    """Перевод строки в значении сделал бы одну запись двумя."""
    r = client.get("/healthz", headers={"x-request-id": "ok\tlevel=error"})
    assert "\n" not in r.headers["x-request-id"]
    assert r.headers["x-request-id"] == "oklevelerror"


def test_request_id_of_only_bad_characters_is_replaced(client):
    r = client.get("/healthz", headers={"x-request-id": "!!!@@@"})
    assert len(r.headers["x-request-id"]) == 12, "сгенерирован новый"


def test_request_id_is_generated_when_absent(client):
    r = client.get("/healthz")
    assert len(r.headers["x-request-id"]) == 12

"""Три пробы с разной логикой и их поведение при завершении."""
from __future__ import annotations

import pytest
from fastapi.testclient import TestClient

from app.main import create_app, state
from app.settings import Settings


def test_healthz_is_alive(client):
    r = client.get("/healthz")
    assert r.status_code == 200
    assert r.json()["status"] == "alive"


def test_readyz_is_ready_after_startup(client):
    r = client.get("/readyz")
    assert r.status_code == 200
    assert r.json()["ready"] is True


def test_startupz_reports_initialized(client):
    body = client.get("/startupz").json()
    assert body["initialized"] is True
    assert body["uptime_seconds"] >= 0


def test_healthz_and_readyz_differ_on_shutdown(client):
    """Главное различие: завершение снимает готовность, но не жизнь."""
    state.shutting_down = True
    try:
        assert client.get("/healthz").status_code == 200
        r = client.get("/readyz")
        assert r.status_code == 503
        assert r.json()["reason"] == "not_shutting_down"
    finally:
        state.shutting_down = False


def test_events_are_refused_while_shutting_down(client):
    state.shutting_down = True
    try:
        r = client.post("/events", json={
            "path": "/a", "method": "GET", "status": 200, "duration_ms": 1})
        assert r.status_code == 503
    finally:
        state.shutting_down = False


def test_readyz_waits_for_warmup():
    """ready_after_seconds отделяет «запущен» от «прогрет»."""
    app = create_app(Settings(ready_after_seconds=3600))
    with TestClient(app) as c:
        assert c.get("/healthz").status_code == 200
        r = c.get("/readyz")
        assert r.status_code == 503
        assert r.json()["checks"]["warmed_up"] is False


def test_probes_do_not_depend_on_stored_data(client, sample_events):
    """Пустое хранилище — не признак неготовности."""
    assert client.get("/readyz").json()["ready"] is True
    for e in sample_events:
        client.post("/events", json=e)
    assert client.get("/readyz").json()["ready"] is True

"""Структурированные логи: одна запись — одна строка JSON."""
from __future__ import annotations

import json
import logging

from app.logging_config import JsonFormatter, request_id_var, sanitize_request_id


def record(**kwargs) -> logging.LogRecord:
    extra = kwargs.pop("extra", {})
    rec = logging.LogRecord(
        name=kwargs.get("name", "eventapi"), level=logging.INFO,
        pathname=__file__, lineno=1,
        msg=kwargs.get("msg", "сообщение"), args=(), exc_info=None)
    for k, v in extra.items():
        setattr(rec, k, v)
    return rec


def test_output_is_one_json_object():
    line = JsonFormatter().format(record())
    payload = json.loads(line)
    assert payload["level"] == "info"
    assert payload["message"] == "сообщение"


def test_no_newlines_inside_record():
    """Многострочная запись ломает разбор построчно."""
    line = JsonFormatter().format(record(msg="первая\nвторая"))
    assert "\n" not in line
    assert json.loads(line)["message"] == "первая\nвторая"


def test_extra_fields_are_included():
    line = JsonFormatter().format(record(extra={"event": "access", "status": 200}))
    payload = json.loads(line)
    assert payload["event"] == "access"
    assert payload["status"] == 200


def test_request_id_comes_from_context():
    token = request_id_var.set("abc123")
    try:
        payload = json.loads(JsonFormatter().format(record()))
        assert payload["request_id"] == "abc123"
    finally:
        request_id_var.reset(token)


def test_request_id_defaults_to_placeholder():
    payload = json.loads(JsonFormatter().format(record()))
    assert payload["request_id"] == "-"


def test_non_serializable_value_does_not_break_output():
    """default=str спасает запись от падения на нестандартном типе."""
    line = JsonFormatter().format(record(extra={"obj": object()}))
    assert json.loads(line)["obj"].startswith("<object")


def test_exception_is_captured():
    try:
        raise ValueError("тестовая")
    except ValueError:
        import sys
        rec = record()
        rec.exc_info = sys.exc_info()
        payload = json.loads(JsonFormatter().format(rec))
        assert "ValueError" in payload["exception"]


def test_sanitize_returns_none_for_empty():
    assert sanitize_request_id("") is None
    assert sanitize_request_id(None) is None

"""Мягкое завершение: активные запросы дорабатываются."""
from __future__ import annotations

import asyncio

import pytest
from fastapi.testclient import TestClient

from app.main import create_app, state
from app.settings import Settings


def test_lifespan_marks_shutdown():
    app = create_app(Settings(shutdown_grace_seconds=0.2))
    with TestClient(app) as c:
        assert c.get("/readyz").json()["ready"] is True
        assert state.shutting_down is False
    assert state.shutting_down is True, "выход из lifespan помечает завершение"


def test_inflight_returns_to_zero_after_request(client):
    client.get("/healthz")
    assert state.inflight == 0


@pytest.mark.anyio
async def test_shutdown_waits_for_inflight():
    """Завершение ждёт активных запросов, а не обрывает их."""
    app = create_app(Settings(shutdown_grace_seconds=2.0))
    state.inflight = 0

    async def fake_request():
        state.inflight += 1
        await asyncio.sleep(0.3)
        state.inflight -= 1

    with TestClient(app):
        task = asyncio.get_event_loop().create_task(fake_request())
        await asyncio.sleep(0.05)
        assert state.inflight == 1
        await task
    assert state.inflight == 0


def test_shutdown_gives_up_after_grace_period():
    """Ожидание ограничено: иначе завершение зависает навсегда."""
    app = create_app(Settings(shutdown_grace_seconds=0.1))
    with TestClient(app):
        state.inflight = 1
    try:
        assert state.inflight == 1, "запрос не дождался — но выход состоялся"
    finally:
        state.inflight = 0

"""Проверка на запущенном процессе, а не через TestClient.

TestClient не поднимает uvicorn, поэтому он не покажет двух вещей:
формата строк, которые печатает сам сервер, и реакции на сигнал.
Обе проверяются только запуском.
"""
from __future__ import annotations

import json
import os
import signal
import socket
import subprocess
import sys
import time
import urllib.request
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]


def free_port() -> int:
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


def wait_for(url: str, timeout: float = 10.0) -> bool:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        try:
            with urllib.request.urlopen(url, timeout=0.5) as r:
                if r.status == 200:
                    return True
        except Exception:
            time.sleep(0.1)
    return False


@pytest.fixture
def server(tmp_path):
    port = free_port()
    log = tmp_path / "server.log"
    env = {**os.environ, "EVENTAPI_LOG_LEVEL": "info",
           "PYTHONPATH": str(ROOT), "PYTHONUNBUFFERED": "1"}
    with log.open("w") as fh:
        proc = subprocess.Popen(
            [sys.executable, "-m", "uvicorn", "app.main:app",
             "--host", "127.0.0.1", "--port", str(port)],
            cwd=ROOT, env=env, stdout=fh, stderr=subprocess.STDOUT)
    try:
        assert wait_for(f"http://127.0.0.1:{port}/healthz"), log.read_text()
        yield proc, port, log
    finally:
        if proc.poll() is None:
            proc.send_signal(signal.SIGKILL)
            proc.wait(timeout=5)


def test_every_log_line_is_json(server):
    """Строки uvicorn и приложения должны быть одного формата.

    Настройка логирования выполняется при создании приложения, а не
    в lifespan: до входа в lifespan uvicorn успевает напечатать
    несколько строк, и они уходили бы в другом формате.
    """
    proc, port, log = server
    urllib.request.urlopen(f"http://127.0.0.1:{port}/healthz").read()
    time.sleep(0.3)
    lines = [l for l in log.read_text(encoding="utf-8").splitlines() if l.strip()]
    assert len(lines) >= 5, "лог подозрительно короткий"
    for line in lines:
        json.loads(line)  # упадёт на первой строке не-JSON


def test_no_ansi_escapes_in_log(server):
    proc, port, log = server
    text = log.read_text(encoding="utf-8")
    assert "\x1b[" not in text
    assert "color_message" not in text


def test_sigterm_shuts_down_gracefully(server):
    """Завершение по сигналу, а не по SIGKILL через grace period.

    Признак успеха — ВРЕМЯ и записи в журнале, а не код возврата.
    """
    proc, port, log = server
    started = time.monotonic()
    proc.send_signal(signal.SIGTERM)
    proc.wait(timeout=10)
    elapsed = time.monotonic() - started

    assert elapsed < 3.0, f"завершение заняло {elapsed:.1f} с"

    events = [json.loads(l).get("event")
              for l in log.read_text(encoding="utf-8").splitlines() if l.strip()]
    assert "shutdown_begin" in events, "lifespan не отработал завершение"
    assert "shutdown_done" in events


def test_process_dies_by_signal_and_это_нормально(server):
    """Код возврата НЕ равен нулю — и это правильное поведение.

    uvicorn после мягкого завершения повторно посылает себе тот же
    сигнал (Server.capture_signals -> signal.raise_signal), чтобы
    родительский процесс увидел настоящую причину смерти. Popen
    сообщает такую смерть как -15, а Docker покажет ExitCode 143
    (128 + 15).

    Отсюда практический вывод: 143 после `docker stop` — признак
    ШТАТНОГО завершения, а не отказа. Отличать мягкое завершение
    от жёсткого нужно по времени и по журналу: ровно grace period
    и отсутствие записей о завершении означают SIGKILL.
    """
    proc, port, log = server
    proc.send_signal(signal.SIGTERM)
    proc.wait(timeout=10)
    assert proc.returncode == -signal.SIGTERM, (
        "ожидалась смерть от SIGTERM; 0 означал бы, что сигнал проглочен")


def test_request_id_appears_in_access_log(server):
    proc, port, log = server
    req = urllib.request.Request(f"http://127.0.0.1:{port}/healthz",
                                 headers={"x-request-id": "trace-42"})
    urllib.request.urlopen(req).read()
    time.sleep(0.3)
    records = [json.loads(l) for l in
               log.read_text(encoding="utf-8").splitlines() if l.strip()]
    access = [r for r in records if r.get("event") == "access"]
    assert any(r["request_id"] == "trace-42" for r in access)

64 passed, 1 warning in 14.01s

Name                    Stmts   Miss  Cover   Missing
-----------------------------------------------------
app/__init__.py             1      0   100%
app/logging_config.py      36      0   100%
app/main.py               109      8    93%   52, 99, 115-118, 163, 204, 208
app/models.py              49      1    98%   34
app/settings.py            36      0   100%
app/storage.py             44      2    95%   93-94
-----------------------------------------------------
TOTAL                     275     11    96%

fastapi[standard]==0.141.1
pydantic-settings==2.14.2

-r requirements.txt
pytest==9.1.1
pytest-cov==7.1.0
httpx==0.28.1

[pytest]
testpaths = tests
addopts = -q

# syntax=docker/dockerfile:1

# ── Общая основа ─────────────────────────────────────────────────────
FROM python:3.13-slim AS base
ENV PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1 \
    PIP_DISABLE_PIP_VERSION_CHECK=1 \
    PATH=/opt/venv/bin:$PATH
WORKDIR /app

# ── Зависимости ──────────────────────────────────────────────────────
FROM base AS deps
RUN python -m venv /opt/venv
# Только файл зависимостей: изменение кода этот слой не сбрасывает.
COPY requirements.txt .
RUN --mount=type=cache,target=/root/.cache/pip \
    pip install -r requirements.txt

# ── Тесты: выполняются только при --target test ───────────────────────
FROM deps AS test
COPY requirements-dev.txt .
RUN --mount=type=cache,target=/root/.cache/pip \
    pip install -r requirements-dev.txt
COPY app/ ./app/
COPY tests/ ./tests/
COPY pytest.ini .
RUN python -m pytest --cov=app --cov-report=term-missing --cov-fail-under=85

# ── Итоговый образ ───────────────────────────────────────────────────
FROM base AS runtime
COPY --from=deps /opt/venv /opt/venv
COPY app/ ./app/

USER 10001:10001

EXPOSE 8000
# Проба внутри образа: без curl, средствами Python — они уже есть.
HEALTHCHECK --interval=10s --timeout=3s --start-period=5s --retries=3 \
    CMD ["python", "-c", "import urllib.request,sys; sys.exit(0 if urllib.request.urlopen('http://127.0.0.1:8000/healthz').status==200 else 1)"]

# Exec-форма: PID 1 — сам сервер, SIGTERM доходит до него.
# python -m uvicorn, а НЕ fastapi run: последний печатает баннер и
# перенастраивает логирование uvicorn ПОСЛЕ импорта приложения, отменяя
# настройку из logging_config. Измерено — см. ниже.
ENTRYPOINT ["python", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

.git
.gitignore
.venv
.v
__pycache__
*.pyc
.pytest_cache
.mypy_cache
.ruff_cache
htmlcov
.coverage
README.md
compose.yaml

CopyIgnoredFile: Attempting to Copy file "tests" that is excluded by .dockerignore
ERROR: failed to compute cache key: "/tests": not found

services:
  api:
    build:
      context: .
      target: runtime
    image: eventapi:1.0.0
    ports:
      - "127.0.0.1:8000:8000"
    environment:
      EVENTAPI_LOG_LEVEL: info
      EVENTAPI_MAX_EVENTS: "50000"
      EVENTAPI_SHUTDOWN_GRACE_SECONDS: "25"
    healthcheck:
      test: ["CMD", "python", "-c",
             "import urllib.request,sys; sys.exit(0 if urllib.request.urlopen('http://127.0.0.1:8000/readyz').status==200 else 1)"]
      interval: 10s
      timeout: 3s
      retries: 3
      start_period: 5s
    stop_grace_period: 30s
    read_only: true
    tmpfs:
      - /tmp
    cap_drop: [ALL]
    security_opt:
      - no-new-privileges:true
    deploy:
      resources:
        limits: {cpus: "0.5", memory: 256M}
    logging:
      driver: json-file
      options: {max-size: "10m", max-file: "3"}
    restart: unless-stopped