220 lines
8.0 KiB
Python
220 lines
8.0 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
|
||
|
||
|
||
# ---------- 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
|
||
|
||
|
||
|
||
# ---------- 12. сессия Telegram: выключена без env, CVE в заголовке ----------
|
||
def test_tgsess_disabled_without_env(monkeypatch, tmp_path):
|
||
from omni import config, tgsess
|
||
monkeypatch.setattr(config, "DATA", tmp_path) # пустая папка: data/tg.json нет
|
||
for k in ("OMNI_TG_SESSION", "OMNI_TG_API_ID", "OMNI_TG_API_HASH"):
|
||
monkeypatch.delenv(k, raising=False)
|
||
assert tgsess.enabled() is False
|
||
assert tgsess.fetch(["thehackernews"], 5) == []
|
||
assert tgsess._key("@SomeChan/") == "SomeChan"
|
||
|
||
|
||
def test_tgsess_picks_up_file_config(monkeypatch, tmp_path):
|
||
"""Креды из data/tg.json подхватываются без env."""
|
||
import json as _json
|
||
from omni import config, tgsess
|
||
monkeypatch.setattr(config, "DATA", tmp_path)
|
||
for k in ("OMNI_TG_SESSION", "OMNI_TG_API_ID", "OMNI_TG_API_HASH", "OMNI_TG_LIMIT"):
|
||
monkeypatch.delenv(k, raising=False)
|
||
(tmp_path / "tg.json").write_text(_json.dumps(
|
||
{"session": "C:/tmp/x", "api_id": "123", "api_hash": "abc", "limit": "7"}), encoding="utf-8")
|
||
cfg = config.tg_cfg()
|
||
assert cfg["session"] == "C:/tmp/x" and cfg["limit"] == "7"
|
||
assert tgsess.enabled() is True
|
||
|
||
|
||
def test_item_cve_prefix():
|
||
from types import SimpleNamespace
|
||
from omni.feeds import _item
|
||
src = SimpleNamespace(name="tg_cvefeed", tags=["cve"])
|
||
it = _item(src, "post", "https://t.me/cvefeed/1", "Improper neutralization of input",
|
||
"Details: CVE-2026-12345 affects the parser badly.")
|
||
assert it.title.startswith("CVE-2026-12345"), it.title
|