omni-parse/tests/test_regress.py

"""Регресс-тесты под пункты прожарки. Без сети: только чистые функции и локальная база.

Запуск: 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


# ---------- 9. tg_web: служебное и пустыши отсекаются ----------
def test_tg_service_filtered():
    from omni.feeds import SERVICE_RE
    assert SERVICE_RE.match("Channel created")
    assert SERVICE_RE.match("Channel photo removed")
    assert SERVICE_RE.match("pinned")
    assert not SERVICE_RE.match("Слив базы 2 млн аккаунтов")


def test_tg_title_from_body():
    from omni.feeds import _item
    from omni.sources import Source
    src = Source("t_x", "abuse", "html", "https://t.me/s/x", "tg_web", ("abuse",), 5, "", {})
    it = _item(src, "post", "https://t.me/x/1", "service",
               "СЛИВ: 1300+ АККАУНТОВ ДЛЯ ВАС!" + chr(10) + "подробности внутри")
    assert "АККАУНТОВ" in it.title


# ---------- 10. сигнал абуза в score ----------
def test_abuse_boost():
    from omni import rank

    class I:
        provider, kind, published, price = "forums", "post", "", None
        score = 0.0

    a, b = I(), I()
    a.title, b.title = "leak dump credentials 40M accounts", "weekly product update notes for teams"
    a.text = b.text = "x" * 600
    assert rank.score(a).score > rank.score(b).score



# ---------- 11. бот: /tag читается без владельца ----------
def test_tag_command_readonly(monkeypatch):
    from omni import bot
    sent = []
    monkeypatch.setattr(bot, "OWNERS", set())
    monkeypatch.setattr(bot, "_send", lambda cid, text, **kw: sent.append(text))
    bot.handle({"message": {"chat": {"id": 1}, "from": {"id": 1}, "text": "/tag"}})
    assert sent and "типы записей" in sent[0], sent[:1]
    assert "/tag" in bot.CMDS