140 lines
4.8 KiB
Python
140 lines
4.8 KiB
Python
"""Регресс-тесты под пункты прожарки. Без сети: только чистые функции и локальная база.
|
||
|
||
Запуск: python -m pytest tests -q
|
||
"""
|
||
import gc
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
sys.path.insert(0, str(ROOT))
|
||
os.environ.setdefault("PYTHONIOENCODING", "utf-8")
|
||
|
||
|
||
# ---------- 1. store.add возвращает пару, /run больше не падает ----------
|
||
def test_store_add_returns_pair():
|
||
from omni import store
|
||
from omni.model import Item
|
||
|
||
it = Item(provider="t", kind="post", source="pytest-src", title="pytest %d" % time.time_ns(),
|
||
url="https://example.invalid/%d" % time.time_ns())
|
||
res = store.add([it])
|
||
assert isinstance(res, tuple) and len(res) == 2, "store.add обязан возвращать (new, dup)"
|
||
new, dup = res
|
||
assert isinstance(new, int) and isinstance(dup, int)
|
||
|
||
|
||
def test_bot_run_unpacks_pair():
|
||
import inspect
|
||
|
||
from omni import bot
|
||
src = inspect.getsource(bot.cmd_run)
|
||
assert "new += store.add(items)" not in src, "старый баг: += по кортежу"
|
||
assert "n_new" in src
|
||
|
||
|
||
# ---------- 2. параметры API ----------
|
||
@pytest.mark.parametrize("raw,default,lo,hi,expect", [
|
||
("abc", 40, 1, 200, 40),
|
||
("-1", 40, 1, 200, 1),
|
||
("99999", 40, 1, 200, 200),
|
||
("25", 40, 1, 200, 25),
|
||
(None, 40, 1, 200, 40),
|
||
("", 40, 1, 200, 40),
|
||
])
|
||
def test_intv(raw, default, lo, hi, expect):
|
||
from omni.serve import _intv
|
||
q = {} if raw is None else {"limit": [raw]}
|
||
assert _intv(q, "limit", default, lo, hi) == expect
|
||
|
||
|
||
def test_no_raw_int_parsing_in_serve():
|
||
src = (ROOT / "omni" / "serve.py").read_text(encoding="utf-8")
|
||
assert "min(int((q.get" not in src, "сырой int() по query-параметру вернулся"
|
||
|
||
|
||
# ---------- 3. соединения закрываются ----------
|
||
def test_no_sqlite_conn_leak():
|
||
from omni import store
|
||
|
||
def live():
|
||
gc.collect()
|
||
return sum(1 for o in gc.get_objects() if isinstance(o, sqlite3.Connection))
|
||
|
||
for _ in range(5):
|
||
store.top(limit=1)
|
||
base = live()
|
||
for _ in range(40):
|
||
store.top(limit=1)
|
||
assert live() - base <= 2, "соединения копятся: db() не закрывает"
|
||
|
||
|
||
# ---------- 4. троттлинг ----------
|
||
def test_throttle_serialises():
|
||
from omni import net
|
||
old = net._cfg.get("per_host_delay")
|
||
net._cfg["per_host_delay"] = 0.05
|
||
net._last.clear()
|
||
try:
|
||
t0 = time.time()
|
||
for _ in range(10):
|
||
net._throttle("pytest.example")
|
||
el = time.time() - t0
|
||
finally:
|
||
net._cfg["per_host_delay"] = old if old is not None else 0.6
|
||
assert el >= 0.40, "троттлинг не держит паузу: %.2fs на 10 вызовов" % el
|
||
|
||
|
||
def test_throttle_threads():
|
||
import threading
|
||
|
||
from omni import net
|
||
old = net._cfg.get("per_host_delay")
|
||
net._cfg["per_host_delay"] = 0.05
|
||
net._last.clear()
|
||
try:
|
||
t0 = time.time()
|
||
ths = [threading.Thread(target=lambda: [net._throttle("pytest.mt") for _ in range(3)]) for _ in range(6)]
|
||
[t.start() for t in ths]
|
||
[t.join() for t in ths]
|
||
el = time.time() - t0
|
||
finally:
|
||
net._cfg["per_host_delay"] = old if old is not None else 0.6
|
||
assert el >= 0.6, "8 потоков проскакивают хост одновременно: %.2fs" % el
|
||
|
||
|
||
# ---------- 5. владелец бота ----------
|
||
def test_bot_not_claimable_by_first_writer(monkeypatch):
|
||
from omni import bot
|
||
sent = []
|
||
monkeypatch.setattr(bot, "OWNERS", set())
|
||
monkeypatch.setattr(bot, "_send", lambda cid, text: sent.append(text))
|
||
bot.handle({"message": {"chat": {"id": 1}, "from": {"id": 999999}, "text": "/run abuse"}})
|
||
assert not bot.OWNERS, "бот снова назначает владельца по первому сообщению"
|
||
assert sent and "read-only" in sent[0]
|
||
|
||
|
||
# ---------- 6. реестр без дублей ----------
|
||
def test_sources_unique():
|
||
from omni import sources
|
||
urls = [s.url.rstrip("/").lower() for s in sources.all_sources() if s.url]
|
||
assert len(urls) == len(set(urls)), "в реестре дубли по url"
|
||
|
||
|
||
# ---------- 7. JOBS не растёт бесконечно ----------
|
||
def test_jobs_evicted():
|
||
from omni import serve
|
||
for i in range(500):
|
||
serve._job("t%d" % i, status="done")
|
||
assert len(serve.JOBS) <= 400, "JOBS растёт без границ: %d" % len(serve.JOBS)
|
||
|
||
|
||
# ---------- 8. ai.py не спит после последней попытки ----------
|
||
def test_ai_does_not_sleep_after_last_attempt():
|
||
src = (ROOT / "omni" / "ai.py").read_text(encoding="utf-8")
|
||
assert "time.sleep(1.5 + attempt)" not in src |