78e27806b9
Заливка корпуса была чёрным ящиком: у источника только last_status (idle/running/done/error), без «сколько из скольки», без причины падения и без способа остановить начатое. Теперь каждый запуск создаёт строку parse_runs, куда воркер раз в ~2с пишет стадию, счётчики и журнал событий. Админка: - шкала загрузки у каждого источника (0→50% выборка, 50→100% индексация), раскрытая строка — журнал прогона по шагам с таймингами; - «Запустить всё» / «Остановить всё» и остановка по одному источнику (кооперативная отмена: воркер останавливается сам, не рвя запись в базу); - пакетное добавление источников (тип + список тем), тип pmc в форме; - загрузка PDF/DOCX/TXT прямо в базу сравнения (index.ingest_upload); - страница «Отладка»: воркеры Celery и их текущие таски, очереди RabbitMQ, покрытие корпуса эмбеддингами, зависшие и упавшие прогоны, конфиг бэкендов. Защита от краш-лупа по consumer_timeout RabbitMQ (docs/DR-HA.md §6), без неё массовый запуск 170+ источников гарантированно ронял воркер: - PARSER_TIME_BUDGET_S (1500с) — прогон закругляется сам и помечается partial; - worker_prefetch_multiplier=1 — таймаут считается от ДОСТАВКИ сообщения, и с дефолтным префетчем очередь долгих run_parser убивала канал на задачах, которые ещё не начинались. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
105 lines
3.2 KiB
Python
105 lines
3.2 KiB
Python
"""Юнит-тесты счётчиков прогона парсинга (app.progress) — без БД и Celery."""
|
||
|
||
from app.progress import RunProgress
|
||
|
||
|
||
class FakeClock:
|
||
"""Управляемое время: тесты бюджета не должны ничего ждать по-настоящему."""
|
||
|
||
def __init__(self) -> None:
|
||
self.now = 0.0
|
||
|
||
def __call__(self) -> float:
|
||
return self.now
|
||
|
||
def advance(self, seconds: float) -> None:
|
||
self.now += seconds
|
||
|
||
|
||
def test_counts_split_by_result_status():
|
||
prog = RunProgress(target=10, clock=FakeClock())
|
||
for status in ("indexed", "indexed", "duplicate", "skipped", "boom"):
|
||
prog.count_result(status)
|
||
|
||
assert prog.processed == 5
|
||
assert prog.added == 2
|
||
assert prog.duplicates == 1
|
||
# Мусор от источника не должен смешиваться с реальными ошибками:
|
||
# иначе панель отладки показывает ошибки там, где их нет
|
||
assert prog.skipped == 1
|
||
assert prog.failed == 1
|
||
|
||
|
||
def test_over_budget_only_after_deadline():
|
||
clock = FakeClock()
|
||
prog = RunProgress(target=100, budget_s=1500, clock=clock)
|
||
|
||
assert not prog.over_budget
|
||
clock.advance(1499)
|
||
assert not prog.over_budget
|
||
clock.advance(2)
|
||
assert prog.over_budget
|
||
|
||
|
||
def test_flush_is_throttled_by_interval():
|
||
clock = FakeClock()
|
||
prog = RunProgress(target=100, flush_interval_s=2.0, clock=clock)
|
||
|
||
assert not prog.should_flush()
|
||
clock.advance(2.5)
|
||
assert prog.should_flush()
|
||
|
||
prog.mark_flushed()
|
||
assert not prog.should_flush()
|
||
clock.advance(2.5)
|
||
assert prog.should_flush()
|
||
|
||
|
||
def test_log_keeps_only_tail():
|
||
prog = RunProgress(target=1, log_limit=3, clock=FakeClock())
|
||
for i in range(10):
|
||
prog.log("info", f"строка {i}")
|
||
|
||
assert len(prog.entries) == 3
|
||
assert [e["msg"] for e in prog.entries] == ["строка 7", "строка 8", "строка 9"]
|
||
|
||
|
||
def test_log_entry_shape():
|
||
clock = FakeClock()
|
||
prog = RunProgress(target=1, clock=clock)
|
||
clock.advance(12.34)
|
||
prog.log("warning", "x" * 900)
|
||
|
||
entry = prog.entries[0]
|
||
assert entry["level"] == "warning"
|
||
assert entry["elapsed"] == 12.3
|
||
# Длинные сообщения режем: журнал целиком лежит в одной JSON-колонке
|
||
assert len(entry["msg"]) == 500
|
||
assert entry["ts"]
|
||
|
||
|
||
def test_snapshot_carries_all_counters():
|
||
prog = RunProgress(target=42, clock=FakeClock())
|
||
prog.stage = "index"
|
||
prog.count_fetched(30)
|
||
prog.count_result("indexed")
|
||
prog.log("info", "поехали")
|
||
|
||
snap = prog.snapshot()
|
||
assert snap["stage"] == "index"
|
||
assert snap["target"] == 42
|
||
assert snap["fetched"] == 30
|
||
assert snap["processed"] == 1
|
||
assert snap["added"] == 1
|
||
assert snap["log"][0]["msg"] == "поехали"
|
||
|
||
|
||
def test_snapshot_log_is_detached_copy():
|
||
"""Снимок уходит в БД как есть — последующие записи не должны его менять."""
|
||
prog = RunProgress(target=1, clock=FakeClock())
|
||
prog.log("info", "первая")
|
||
snap = prog.snapshot()
|
||
prog.log("info", "вторая")
|
||
|
||
assert len(snap["log"]) == 1
|