test(stability): stage 7 — end-to-end verification (integration tests + report) (#7)
Caps the stacked stability work (stages 1-6). Verification only — ZERO production code change (git diff against fix/stage-6-healthcheck touches only these two files). - tests/test_stage7_integration.py (8 cross-stage integration tests via TestClient + a mocked TelegramClient, no network): Range at the /media route level (206/206/416 through get_media -> prepare_file_response -> FileResponse); /ping stays prompt and issues zero TG RPC while a slow op is parked; in-flight dedup shares one download and drains _inflight after completion AND after request cancellation (no stuck key / hung waiter); str(channel) access-time hit -> flush -> the bulk UPDATE lands on the seeded row (mutation-verified: transposing the key columns reds it). - docs/stability-verification.md: per-stage DoD -> evidence mapping (test or "operator observation" for prod-only items), the exact manual curl/lsof scenarios for the operator to run post-deploy, the diag-log signals to watch, and the per-stage independent-commit rollback plan. Full integrated suite: 260 passed (252 baseline + 8). The prod deploy + live diag-log observation (plan items 3-4) are the operator's call — this stage does not deploy. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,138 @@
|
||||
# Стадия 7 — сквозная верификация плана стабилизации
|
||||
|
||||
Дата: 2026-07-05. Ветка: `fix/stage-7-verification` (на базе `fix/stage-6-healthcheck`,
|
||||
кумулятивно содержит стадии 1–6). Эта стадия **ничего не меняет в поведении** прод-кода —
|
||||
только автоматизирует проверки и фиксирует, что должен наблюдать оператор после деплоя.
|
||||
|
||||
## Итог автотестов
|
||||
|
||||
Каноничный прогон (изоляция `config` через `sys.modules`): `python -m pytest tests/`.
|
||||
|
||||
```
|
||||
260 passed
|
||||
```
|
||||
|
||||
Раскладка по стадиям (все зелёные):
|
||||
|
||||
| Стадия | Файл тестов | Кол-во | Что покрывает |
|
||||
|--------|-------------|--------|----------------|
|
||||
| 1 — анти-зависания | `tests/test_stage1_hangs.py` | 6 | таймаут RPC-гейта без утечки пермита; воркер переживает Exception/FloodWait, `task_done` сбалансирован; отмена в spacing-ожидании не теряет пермит |
|
||||
| 2 — статика/большие видео | `tests/test_stage2_static.py` | 15 | атомарный `_download_atomic` (publish-on-rename, чистка `.part.` при таймауте/zero-size/гонке); дедуп конкурентных скачиваний; FloodWait→429; touch mtime у `temp_*`; свипер чистит `.part.`+legacy `.tmp.`; баланс HTTP-семафора |
|
||||
| 3 — FileResponse | `tests/test_stage3_fileresponse.py` | 19 | матрица Range (0-499/500-/-500/за EOF→416/мусор→400/мульти-range→206 multipart); сохранены mtime-touch, delete_after-BackgroundTask, MIME-кэш; чистый ASGI-логгер |
|
||||
| 4 — гигиена event loop | `tests/test_stage4_eventloop.py` | 19 | ленивый `raw_message`; вынос side-effect IO из `process_message` (bulk upsert media id); рендер-пайплайн ушёл в поток без create_task/get_running_loop; XSS вычищен во всех 4 выводах ровно одним проходом |
|
||||
| 5 — батчинг SQLite | `tests/test_stage5_sqlite.py` | 8 | кэш-хит пишет в аккумулятор, а не в SQLite; flush→DB; snapshot-then-clear не теряет поздние апдейты; re-queue при сбое; str(channel)-ключи |
|
||||
| 6 — healthcheck | `tests/test_stage6_healthcheck.py` | 11 | `/ping` 200/503 по connected+age; ноль TG RPC; watchdog отключён→чистая проверка коннекта; `watchdog_last_ok_age()` |
|
||||
| 7 — интеграция | `tests/test_stage7_integration.py` | 8 | сквозные сценарии (см. ниже) |
|
||||
| — регрессия парсера | `tests/test_postparser_*.py` | 174 | заголовки/флаги/автор (существовавшие до плана) |
|
||||
|
||||
## Новые интеграционные тесты стадии 7 (DoD → сквозное доказательство)
|
||||
|
||||
`tests/test_stage7_integration.py` драйвит **реальные точки входа** (`get_media`, `ping`,
|
||||
flush), чтобы поймать регрессии, которые проявляются только при взаимодействии стадий:
|
||||
|
||||
1. **Range на уровне маршрута `/media`** — `test_media_route_range_prefix_0_99`,
|
||||
`_range_suffix_last_100`, `_range_unsatisfiable_416`. Прогоняют `get_media` (кэш-хит) →
|
||||
`prepare_file_response` → `FileResponse` через реальный ASGI: `bytes=0-99`→206 с точным
|
||||
Content-Range/длиной и байтами, `bytes=-100`→206 (суффикс), `bytes=999999999-`→416 (`*/size`).
|
||||
Ловит: если кэш-хит перестанет доходить до FileResponse (ре-буферизация тела, ручные
|
||||
заголовки) или сломается связка digest-гейта — Range перестанет работать. (Стадия 3
|
||||
пинит это на `prepare_file_response` напрямую; здесь — сшивка стадий 2+3.)
|
||||
2. **`/ping` быстр и без RPC при висящей операции** — `test_ping_prompt_and_rpc_free_while_slow_op_pending`
|
||||
и `_reports_degraded_promptly_while_slow_op_pending`. Пока фейковая медленная корутина
|
||||
(модель зависшего hot-path) висит на `Event`, `ping()` возвращает 200/503 корректно и
|
||||
**до** завершения медленной операции, при нуле вызовов `get_me`/`safe_get_messages`.
|
||||
Ловит: рекаплинг `/ping` к TG RPC или к любому awaitable, который может застопориться.
|
||||
3. **Дедуп + очистка при disconnect через реальный `get_media`** —
|
||||
`test_get_media_concurrent_shares_one_download_and_drains_registry` (2 конкурентных
|
||||
запроса → одна скачка, `_inflight` пуст) и `_request_cancel_does_not_stick_registry_or_hang_sibling`
|
||||
(отмена запроса-«клиента» не оставляет застрявший ключ и не вешает соседа). Ловит:
|
||||
возврат скачивания в корутину запроса (отмена убила бы download) или потерю `finally`-pop.
|
||||
4. **str(channel)-ключ access-time end-to-end** — `test_media_cache_hit_flush_updates_str_keyed_db_row`.
|
||||
Кэш-хит `/media` записывает str-ключ, flush обновляет ту же строку в реальной SQLite
|
||||
(hit→flush→DB). Сшивает две половины: ключ, который пишет hot-path, и WHERE, по которому
|
||||
апдейтит flush. Ловит любое расхождение ключа аккумулятора и WHERE bulk-UPDATE (напр.
|
||||
перестановку колонок в SQL — проверено мутацией: тест краснеет, `added` остаётся stale).
|
||||
(int/str-хазард самого канала живёт в `download_media_file` и закрыт стадией 5 —
|
||||
`test_cache_hit_keys_channel_as_str`; здесь покрыта связка get_media+flush.)
|
||||
|
||||
## Соответствие DoD стадий → доказательство
|
||||
|
||||
- **DoD 1** «ни один путь не ждёт TG без таймаута; воркер не умирает молча» →
|
||||
`test_stage1_hangs.py` (гейт-таймаут, живучесть воркера). Живая проверка supervision
|
||||
под нагрузкой — **наблюдение оператора** (лог `supervised_task_crashed/…_exited`).
|
||||
- **DoD 2** «клиенту никогда не отдаётся неполный файл; флуд→429; обрезки не живут >часа» →
|
||||
`test_stage2_static.py` целиком + интеграционный дедуп-тест стадии 7.
|
||||
- **DoD 3** «Range-тесты зелёные; поведение эндпоинта эквивалентно (± RFC-допущения)» →
|
||||
`test_stage3_fileresponse.py` + Range на уровне `/media` (стадия 7). «Нет потока-на-чанк»
|
||||
— **наблюдение оператора** (нагрузочный запрос большого файла, число io-потоков).
|
||||
- **DoD 4** «генерация 100-сообщ. фида не блокирует луп (параллельный /ping <100 мс); XSS
|
||||
зелёный; media id сохраняются» → `test_stage4_eventloop.py` (рендер в потоке, XSS,
|
||||
bulk upsert) + `/ping`-decoupling стадии 7. Живой замер «<100 мс на проде» —
|
||||
**наблюдение оператора**.
|
||||
- **DoD 5** «на кэш-хит /media ноль обращений к SQLite; фоновая запись раз в минуту» →
|
||||
`test_stage5_sqlite.py` + str-ключ end-to-end стадии 7.
|
||||
- **DoD 6** «во время зависшего TG RPC /ping отвечает мгновенно (503); контейнер не
|
||||
рестартится от медленного фида» → `test_stage6_healthcheck.py` + `/ping`-under-slow-op
|
||||
стадии 7. Отсутствие autoheal-рестартов на холодном кэше — **наблюдение оператора**.
|
||||
|
||||
## Ручные curl-сценарии для оператора (пост-деплой, локально в контейнере)
|
||||
|
||||
Подставить реальный `{channel}/{post_id}/{fid}/{digest}` (валидная подпись обязательна).
|
||||
|
||||
```bash
|
||||
# 1. Range-тройка (ожидания: 206 / 206 / 416)
|
||||
curl -s -D- -o /dev/null -H "Range: bytes=0-99" "http://127.0.0.1:80/media/{ch}/{pid}/{fid}/{digest}"
|
||||
curl -s -D- -o /dev/null -H "Range: bytes=-100" "http://127.0.0.1:80/media/{ch}/{pid}/{fid}/{digest}"
|
||||
curl -s -D- -o /dev/null -H "Range: bytes=999999999-" "http://127.0.0.1:80/media/{ch}/{pid}/{fid}/{digest}"
|
||||
|
||||
# 2. Параллельные запросы одного БОЛЬШОГО видео (>100 MB), пока не в кэше:
|
||||
# оба должны получить ПОЛНЫЙ файл (одинаковый размер), без частичной отдачи.
|
||||
URL="http://127.0.0.1:80/media/{ch}/{pid}/{bigfid}/{digest}"
|
||||
curl -s -o /tmp/a "$URL" & curl -s -o /tmp/b "$URL" & wait
|
||||
ls -l /tmp/a /tmp/b # размеры совпадают и равны полному файлу; на диске нет *.part.*
|
||||
|
||||
# 3. Обрыв на середине стрима — нет утечки тасков/фд:
|
||||
# считать fd до/после и убедиться, что не растут монотонно.
|
||||
lsof -p $(pgrep -f api_server) | wc -l # baseline
|
||||
timeout 2 curl -s -o /dev/null "$URL"; sleep 5 # оборвать скачку на середине (несколько раз)
|
||||
lsof -p $(pgrep -f api_server) | wc -l # не выросло относительно baseline
|
||||
|
||||
# 4. Генерация большого фида + параллельный /ping (<100 мс во время генерации):
|
||||
curl -s -o /dev/null "http://127.0.0.1:80/rss/{big_channel}" &
|
||||
for i in $(seq 1 20); do curl -s -o /dev/null -w "%{time_total}\n" "http://127.0.0.1:80/ping"; done
|
||||
wait # все замеры /ping должны быть заметно < 0.100 s
|
||||
```
|
||||
|
||||
> Сценарии 2 и 3 (реальная скачка большого видео + реальный подсчёт fd через `lsof`) в
|
||||
> headless-тестах **намеренно не подделаны** — им нужен живой сервер, реальные загрузки и
|
||||
> реальные файловые дескрипторы. Дедуп-инвариант и disconnect-очистка доказаны на уровне
|
||||
> реестра/`get_media` (тесты стадии 7 №3), но «нет утечки fd на проде» проверяет оператор.
|
||||
|
||||
## Diag-логи для наблюдения после деплоя
|
||||
|
||||
Ожидаемая динамика (сравнить с до-деплойным baseline):
|
||||
|
||||
- `diag_semaphore_wait` — ожидание HTTP-семафора должно **упасть** (реже/меньше секунд).
|
||||
- `diag_download_timing` — время скачивания стабилизируется; нет длинных «зависаний».
|
||||
- `diag_sanitize_slow` — почти **исчезнуть** (один sanitize на выходную границу, стадия 4).
|
||||
- `watchdog: heartbeat` — **продолжаются** штатно (живость TG-сессии).
|
||||
- На момент FloodWait — **нет всплеска 404** на `/media`; вместо этого `media_flood_wait`
|
||||
и ответы **429** с `Retry-After` (стадия 2.3).
|
||||
- Признаки supervision: любые `supervised_task_crashed` / `supervised_task_exited` на
|
||||
CRITICAL — сигнал разобраться, но задача при этом перезапускается (не тихая смерть).
|
||||
|
||||
## План отката (rollback)
|
||||
|
||||
Каждая стадия — отдельный, независимо ревертируемый коммит на своей ветке
|
||||
(`fix/stage-1-hangs` … `fix/stage-6-healthcheck`); порядок стадий = порядок деплоя.
|
||||
|
||||
- Стадии 1 и 2 — низкорисковые, деплоятся первыми; при регрессе откатываются по одной
|
||||
(`git revert <commit>` соответствующей стадии) без затрагивания остальных.
|
||||
- Стадия 3 (FileResponse) независима — реверт возвращает прежний ручной стриминг.
|
||||
- Стадия 4 требует, чтобы 4.2 откатывалась не позже 4.3 (иначе рендер-в-потоке остался бы
|
||||
без безопасного пути сохранения media id) — откатывать стадию 4 целиком.
|
||||
- Стадии 5 и 6 независимы, откатываются по отдельности.
|
||||
- Стадия 7 — только тесты и этот документ: реверт ничего не меняет в проде.
|
||||
|
||||
Прод-деплой, живое наблюдение diag-логов и обновление прод-compose (healthcheck→`/ping`,
|
||||
вне репозитория) — **зона ответственности оператора** (пункты 3–4 плана стадии 7).
|
||||
@@ -0,0 +1,349 @@
|
||||
# flake8: noqa
|
||||
# pylint: disable=protected-access, missing-function-docstring, missing-class-docstring
|
||||
# pylint: disable=redefined-outer-name, logging-fstring-interpolation, line-too-long
|
||||
# pylance: disable=reportMissingImports, reportMissingModuleSource
|
||||
"""
|
||||
Stage 7 — cross-stage END-TO-END integration tests.
|
||||
|
||||
Unlike the per-stage suites (which pin one seam in isolation), these drive the real
|
||||
public entry points (`get_media`, `ping`, the access-time flush) so a regression that
|
||||
only shows up when the stages interact goes red. Every scenario here maps to one of the
|
||||
plan's "Стадия 7 — сквозные ручные сценарии"; the ones that genuinely need a running
|
||||
server + real downloads + lsof (fd-leak counting) are NOT faked here — they stay in
|
||||
docs/stability-verification.md for the operator's prod observation.
|
||||
|
||||
Automated here:
|
||||
- Range semantics at the /media ROUTE level (get_media -> prepare_file_response ->
|
||||
FileResponse driven through real ASGI): bytes=0-99 -> 206, bytes=-100 -> 206,
|
||||
bytes=999999999- -> 416. (Stage 3 pins these on prepare_file_response directly; this
|
||||
adds the end-to-end assertion that get_media's cache-hit branch reaches FileResponse
|
||||
with a live Range still honored — i.e. stages 2+3 wired together.)
|
||||
- /ping (stage 6) stays fast + correct while a deliberately-slow op is in flight, issuing
|
||||
ZERO Telegram RPC — proving the healthcheck is decoupled from the hot/blocked paths.
|
||||
- In-flight dedup + disconnect cleanup (stages 1/2) through the REAL get_media entry
|
||||
point: concurrent requests share one download and the _inflight registry drains; a
|
||||
cancelled request (client disconnect) leaves neither a stuck key nor a hung sibling.
|
||||
- str(channel) access-time key consistency (stage 5) end-to-end: a /media cache hit for
|
||||
an int-ish channel records the str-keyed timestamp, and the flush UPDATE matches the
|
||||
str-keyed DB row (hit -> flush -> DB), the exact affinity gotcha the plan warns about.
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import sqlite3
|
||||
import asyncio
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
# Add project root to sys.path and mock the config module (same pattern as the other tests).
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
||||
sys.modules['config'] = __import__('tests.mock_config', fromlist=['get_settings'])
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
import api_server
|
||||
from pyrogram import errors
|
||||
from file_io import init_db_sync
|
||||
|
||||
|
||||
# 2048 deterministic bytes so Range slices are byte-checkable.
|
||||
BODY = bytes(range(256)) * 8
|
||||
SIZE = len(BODY)
|
||||
|
||||
|
||||
def _media_app():
|
||||
"""A bare app that mounts the REAL get_media (and ping) with NO lifespan, so
|
||||
client.start() never runs — a pure cache hit never touches Telegram, and FileResponse
|
||||
computes Range/206/416 at send time, which only happens when driven through ASGI."""
|
||||
app = FastAPI()
|
||||
app.add_api_route("/media/{channel}/{post_id}/{file_unique_id}/{digest}", api_server.get_media, methods=["GET"])
|
||||
app.add_api_route("/media/{channel}/{post_id}/{file_unique_id}", api_server.get_media, methods=["GET"])
|
||||
return app
|
||||
|
||||
|
||||
def _seed_cache(tmp_path, channel, post_id, fid, body=BODY):
|
||||
cache_dir = tmp_path / "data" / "cache" / str(channel) / str(post_id)
|
||||
cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
(cache_dir / fid).write_bytes(body)
|
||||
return cache_dir / fid
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Range semantics at the /media ROUTE level (stages 2 + 3 wired together).
|
||||
# Plan scenario: `curl -H "Range: bytes=0-99" / "bytes=-100" / "bytes=999999999-"`.
|
||||
# Regression caught: any change that makes get_media's cache-hit branch stop reaching
|
||||
# FileResponse (e.g. re-buffering the body, hand-rolling headers, dropping the media_key
|
||||
# path) or that breaks the digest gate wiring — the Range would stop being honored.
|
||||
# --------------------------------------------------------------------------- #
|
||||
def test_media_route_range_prefix_0_99(monkeypatch, tmp_path):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
_seed_cache(tmp_path, "chan", 3, "fidR")
|
||||
monkeypatch.setattr(api_server, "verify_media_digest", lambda url, digest: True)
|
||||
# Keep the MIME path DB-free; the FileResponse/Range machinery is what we exercise.
|
||||
monkeypatch.setattr(api_server, "get_mime_type_sync", lambda *a, **k: None)
|
||||
monkeypatch.setattr(api_server, "set_mime_type_sync", lambda *a, **k: None)
|
||||
|
||||
c = TestClient(_media_app())
|
||||
r = c.get("/media/chan/3/fidR/anydigest", headers={"Range": "bytes=0-99"})
|
||||
assert r.status_code == 206
|
||||
assert r.headers["content-range"] == f"bytes 0-99/{SIZE}"
|
||||
assert r.headers["content-length"] == "100"
|
||||
assert r.content == BODY[:100]
|
||||
assert r.headers["accept-ranges"] == "bytes"
|
||||
|
||||
|
||||
def test_media_route_range_suffix_last_100(monkeypatch, tmp_path):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
_seed_cache(tmp_path, "chan", 3, "fidS")
|
||||
monkeypatch.setattr(api_server, "verify_media_digest", lambda url, digest: True)
|
||||
monkeypatch.setattr(api_server, "get_mime_type_sync", lambda *a, **k: None)
|
||||
monkeypatch.setattr(api_server, "set_mime_type_sync", lambda *a, **k: None)
|
||||
|
||||
c = TestClient(_media_app())
|
||||
r = c.get("/media/chan/3/fidS/anydigest", headers={"Range": "bytes=-100"})
|
||||
assert r.status_code == 206
|
||||
assert r.headers["content-range"] == f"bytes {SIZE - 100}-{SIZE - 1}/{SIZE}"
|
||||
assert r.headers["content-length"] == "100"
|
||||
assert r.content == BODY[-100:]
|
||||
|
||||
|
||||
def test_media_route_range_unsatisfiable_416(monkeypatch, tmp_path):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
_seed_cache(tmp_path, "chan", 3, "fidU")
|
||||
monkeypatch.setattr(api_server, "verify_media_digest", lambda url, digest: True)
|
||||
monkeypatch.setattr(api_server, "get_mime_type_sync", lambda *a, **k: None)
|
||||
monkeypatch.setattr(api_server, "set_mime_type_sync", lambda *a, **k: None)
|
||||
|
||||
c = TestClient(_media_app())
|
||||
r = c.get("/media/chan/3/fidU/anydigest", headers={"Range": "bytes=999999999-"})
|
||||
assert r.status_code == 416
|
||||
# Starlette's 416 Content-Range is `*/size` (documented stage-3 RFC-7233 delta).
|
||||
assert r.headers["content-range"] == f"*/{SIZE}"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# /ping (stage 6) stays fast + correct while a slow op is in flight, zero TG RPC.
|
||||
# Plan scenario: "Генерация фида на 100+ сообщений + параллельный /ping -> ping < 100 мс".
|
||||
# We model the concurrent slow op as a coroutine parked on an Event that is NEVER set
|
||||
# during the ping, and assert ping resolves while it is still pending AND touches no RPC.
|
||||
# Regression caught: re-coupling /ping to a TG RPC (get_me / safe_get_messages) or to any
|
||||
# awaitable that a blocked hot path could stall — the ping would no longer return promptly
|
||||
# or the spy counts would go non-zero.
|
||||
# --------------------------------------------------------------------------- #
|
||||
class _FakeKurigram:
|
||||
def __init__(self, is_connected=True):
|
||||
self.is_connected = is_connected
|
||||
self.get_me_calls = 0
|
||||
|
||||
async def get_me(self):
|
||||
self.get_me_calls += 1
|
||||
raise AssertionError("/ping must never call get_me()")
|
||||
|
||||
|
||||
class _FakeTelegramClient:
|
||||
def __init__(self, age, is_connected=True):
|
||||
self._age = age
|
||||
self.client = _FakeKurigram(is_connected=is_connected)
|
||||
self.safe_get_messages_calls = 0
|
||||
|
||||
def watchdog_last_ok_age(self):
|
||||
return self._age
|
||||
|
||||
async def safe_get_messages(self, *a, **k):
|
||||
self.safe_get_messages_calls += 1
|
||||
raise AssertionError("/ping must never call safe_get_messages()")
|
||||
|
||||
|
||||
async def test_ping_prompt_and_rpc_free_while_slow_op_pending(monkeypatch):
|
||||
threshold = api_server.Config["tg_ping_unhealthy_after"]
|
||||
fake = _FakeTelegramClient(age=threshold - 10, is_connected=True)
|
||||
monkeypatch.setattr(api_server, "client", fake)
|
||||
|
||||
# A concurrent slow operation (a stand-in for a hung feed/RPC hot path) parked on an
|
||||
# Event we deliberately never set for the duration of the ping.
|
||||
gate = asyncio.Event()
|
||||
|
||||
async def slow_op():
|
||||
await gate.wait()
|
||||
|
||||
slow = asyncio.create_task(slow_op())
|
||||
await asyncio.sleep(0) # let slow_op start and park on the gate
|
||||
|
||||
# The real proof of decoupling: ping() returns under a tight deadline while the slow op
|
||||
# is parked, AND issues zero TG RPC. wait_for reds if ping ever blocks; the spies (which
|
||||
# raise if touched) red if ping recouples to any RPC. `not slow.done()` is only a sanity
|
||||
# check that ping did not somehow drive the parked op — the gate keeps it pending anyway.
|
||||
resp = await asyncio.wait_for(api_server.ping(), timeout=1.0)
|
||||
|
||||
assert resp.status_code == 200 # correct health while connected + fresh
|
||||
assert not slow.done() # sanity: slow op still parked, ping did not await it
|
||||
assert fake.client.get_me_calls == 0 # decoupled: zero TG RPC
|
||||
assert fake.safe_get_messages_calls == 0
|
||||
|
||||
gate.set()
|
||||
await slow
|
||||
|
||||
|
||||
async def test_ping_reports_degraded_promptly_while_slow_op_pending(monkeypatch):
|
||||
threshold = api_server.Config["tg_ping_unhealthy_after"]
|
||||
fake = _FakeTelegramClient(age=threshold + 100, is_connected=True) # stale probe
|
||||
monkeypatch.setattr(api_server, "client", fake)
|
||||
|
||||
gate = asyncio.Event()
|
||||
|
||||
async def slow_op():
|
||||
await gate.wait()
|
||||
|
||||
slow = asyncio.create_task(slow_op())
|
||||
await asyncio.sleep(0)
|
||||
|
||||
resp = await asyncio.wait_for(api_server.ping(), timeout=1.0)
|
||||
|
||||
assert resp.status_code == 503 # stale watchdog probe -> degraded, still instant
|
||||
assert not slow.done() # sanity: slow op still parked, ping did not await it
|
||||
assert fake.client.get_me_calls == 0
|
||||
assert fake.safe_get_messages_calls == 0
|
||||
|
||||
gate.set()
|
||||
await slow
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# In-flight dedup + disconnect cleanup (stages 1/2) through the REAL get_media entry.
|
||||
# Plan scenario: "Параллельные запросы одного большого видео -> нет частичной отдачи" and
|
||||
# "Отключение клиента на середине стрима -> нет утечки тасков/фд".
|
||||
# The pure-unit stage-2 tests pin _download_deduped directly; these drive get_media so the
|
||||
# HTTP semaphore + dedup registry + serve path are proven wired together.
|
||||
# Regression caught: moving the download back into the request coroutine (so a disconnect
|
||||
# cancels it), or dropping the finally-pop, would leave a stuck _inflight key / hung sibling.
|
||||
# --------------------------------------------------------------------------- #
|
||||
async def test_get_media_concurrent_shares_one_download_and_drains_registry(monkeypatch, tmp_path):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
api_server._inflight.clear()
|
||||
monkeypatch.setattr(api_server, "verify_media_digest", lambda url, digest: True)
|
||||
# Serve step is not under test here; keep it to a sentinel so we assert on dedup + registry.
|
||||
sentinel = object()
|
||||
|
||||
async def fake_prepare(file_path, request=None, delete_after=False, media_key=None):
|
||||
return sentinel
|
||||
|
||||
monkeypatch.setattr(api_server, "prepare_file_response", fake_prepare)
|
||||
|
||||
calls = []
|
||||
|
||||
async def slow_dl(channel, post_id, fid):
|
||||
calls.append(fid)
|
||||
await asyncio.sleep(0.05) # real overlap window for the two requests
|
||||
return (f"/final/{fid}", False)
|
||||
|
||||
monkeypatch.setattr(api_server, "download_media_file", slow_dl)
|
||||
|
||||
req = SimpleNamespace(headers={})
|
||||
t1 = asyncio.create_task(api_server.get_media("chan", 9, "vfid", request=req, digest="x"))
|
||||
t2 = asyncio.create_task(api_server.get_media("chan", 9, "vfid", request=req, digest="x"))
|
||||
r1, r2 = await asyncio.gather(t1, t2)
|
||||
|
||||
assert r1 is sentinel and r2 is sentinel
|
||||
assert calls == ["vfid"] # exactly ONE real download, shared by both requests
|
||||
assert not api_server._inflight # registry drained — no forever-busy key
|
||||
|
||||
|
||||
async def test_get_media_request_cancel_does_not_stick_registry_or_hang_sibling(monkeypatch, tmp_path):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
api_server._inflight.clear()
|
||||
monkeypatch.setattr(api_server, "verify_media_digest", lambda url, digest: True)
|
||||
sentinel = object()
|
||||
|
||||
async def fake_prepare(file_path, request=None, delete_after=False, media_key=None):
|
||||
return sentinel
|
||||
|
||||
monkeypatch.setattr(api_server, "prepare_file_response", fake_prepare)
|
||||
|
||||
gate = asyncio.Event()
|
||||
calls = []
|
||||
|
||||
async def held_dl(channel, post_id, fid):
|
||||
calls.append(fid)
|
||||
await gate.wait() # hold the shared download open until we release it
|
||||
return (f"/final/{fid}", False)
|
||||
|
||||
monkeypatch.setattr(api_server, "download_media_file", held_dl)
|
||||
|
||||
req = SimpleNamespace(headers={})
|
||||
t1 = asyncio.create_task(api_server.get_media("chan", 9, "cfid", request=req, digest="x"))
|
||||
await asyncio.sleep(0.02) # t1 registers the future + starts the detached download
|
||||
t2 = asyncio.create_task(api_server.get_media("chan", 9, "cfid", request=req, digest="x"))
|
||||
await asyncio.sleep(0.02)
|
||||
|
||||
# First client disconnects: its request coroutine is cancelled mid-wait.
|
||||
t1.cancel()
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await t1
|
||||
|
||||
# The detached download is unaffected; releasing it resolves the surviving request.
|
||||
gate.set()
|
||||
r2 = await asyncio.wait_for(t2, timeout=2.0)
|
||||
assert r2 is sentinel
|
||||
assert calls == ["cfid"] # download ran exactly once (not restarted)
|
||||
assert not api_server._inflight # registry drained despite the disconnect
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# str(channel) access-time key consistency (stage 5) END-TO-END: hit -> flush -> DB.
|
||||
# Plan gotcha #9: "Ключи SQLite: channel всегда str(...)"; if the key the hot path RECORDS
|
||||
# and the key the flush UPDATEs by ever diverge, the timestamp silently stops updating and
|
||||
# the file eventually falls out of the cache. Stage 5 pins hit and flush separately; this
|
||||
# stitches them: the /media cache-hit records a (str-channel) key, and the flush's bulk
|
||||
# UPDATE must land on that exact DB row. (get_media's route channel is already a str, so the
|
||||
# str() there is a defensive no-op; the live int/str hazard is in download_media_file, pinned
|
||||
# by stage-5's test_cache_hit_keys_channel_as_str — this test covers the get_media+flush leg.)
|
||||
# Regression caught: any accumulator-key vs UPDATE-WHERE inconsistency (e.g. a transposed
|
||||
# key-column order in the bulk SQL, verified to turn this test red) leaves `added` stale.
|
||||
# --------------------------------------------------------------------------- #
|
||||
async def test_media_cache_hit_flush_updates_str_keyed_db_row(monkeypatch, tmp_path):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
api_server._access_updates = {}
|
||||
|
||||
channel, post_id, fid = "12345", 7, "fidINT" # int-ish channel, passed as the route str
|
||||
_seed_cache(tmp_path, channel, post_id, fid)
|
||||
|
||||
db = str(tmp_path / "access.db")
|
||||
init_db_sync(db)
|
||||
# Seed the row keyed by the STRING channel with an old access time.
|
||||
conn = sqlite3.connect(db)
|
||||
conn.execute(
|
||||
"INSERT INTO media_file_ids (channel, post_id, file_unique_id, added) VALUES (?,?,?,?)",
|
||||
(channel, post_id, fid, 1.0),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
monkeypatch.setattr(api_server, "DB_PATH", db)
|
||||
|
||||
monkeypatch.setattr(api_server, "verify_media_digest", lambda url, digest: True)
|
||||
sentinel = object()
|
||||
|
||||
async def fake_prepare(file_path, request=None, media_key=None):
|
||||
return sentinel
|
||||
|
||||
monkeypatch.setattr(api_server, "prepare_file_response", fake_prepare)
|
||||
|
||||
before = time.time()
|
||||
resp = await api_server.get_media(channel, post_id, fid, request=SimpleNamespace(headers={}), digest="x")
|
||||
assert resp is sentinel
|
||||
|
||||
# Hot path recorded the str-keyed access time (never the raw int form).
|
||||
assert (channel, post_id, fid) in api_server._access_updates
|
||||
|
||||
# Flush the accumulator; the bulk UPDATE must match the str-keyed row and refresh `added`.
|
||||
await api_server._flush_access_updates()
|
||||
assert api_server._access_updates == {}
|
||||
|
||||
conn = sqlite3.connect(db)
|
||||
added = conn.execute(
|
||||
"SELECT added FROM media_file_ids WHERE channel = ? AND post_id = ? AND file_unique_id = ?",
|
||||
(channel, post_id, fid),
|
||||
).fetchone()[0]
|
||||
conn.close()
|
||||
assert added >= before # refreshed from the stale 1.0 to ~now via the str-keyed WHERE
|
||||
Reference in New Issue
Block a user