208 lines
8.1 KiB
Python
208 lines
8.1 KiB
Python
"""Панель v2: API-слой panel_api + регистрация роутов в serve. Без сети."""
|
|
import inspect
|
|
import os
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(ROOT))
|
|
os.environ.setdefault("PYTHONIOENCODING", "utf-8")
|
|
|
|
|
|
def _tmp_item(tag=None):
|
|
from omni import store
|
|
from omni.model import Item
|
|
|
|
tag = tag or str(time.time_ns())
|
|
it = Item(provider="pytest", kind="post", source="pytest-panel",
|
|
title="pytest panel %s" % tag, text="текст <b>%s</b> & xss" % tag,
|
|
url="https://example.invalid/p/%s" % tag, published=time.strftime("%Y-%m-%d"))
|
|
store.add([it])
|
|
with store.db() as c:
|
|
row = c.execute("SELECT id FROM items WHERE url=?", (it.url,)).fetchone()
|
|
return int(row[0]), tag
|
|
|
|
|
|
# ---------- timeline ----------
|
|
def test_timeline_shape_and_modes():
|
|
from omni import panel_api
|
|
|
|
for mode in ("pub", "seen"):
|
|
d = panel_api.timeline(days=7, mode=mode)
|
|
assert d["ok"] and d["mode"] == mode
|
|
assert len(d["points"]) == 7, "ровно N дней, пустые заполнены нулями"
|
|
assert d["total"] == sum(p["n"] for p in d["points"])
|
|
for p in d["points"]:
|
|
assert set(p) == {"day", "n", "cve", "sev"}
|
|
assert p["cve"] <= p["n"] and p["sev"] <= p["n"]
|
|
assert "last24" in panel_api.timeline(days=2)
|
|
|
|
|
|
def test_timeline_days_clamped():
|
|
"""0/отрицательное → 1 день, None → 30, огромное → потолок 365."""
|
|
from omni import panel_api
|
|
|
|
assert len(panel_api.timeline(days=None)["points"]) == 30
|
|
assert len(panel_api.timeline(days=0)["points"]) == 1
|
|
assert len(panel_api.timeline(days=-5)["points"]) == 1
|
|
assert len(panel_api.timeline(days=10_000)["points"]) == 365
|
|
|
|
|
|
# ---------- детали записи ----------
|
|
def test_item_detail_cleans_html_and_finds_siblings():
|
|
from omni import panel_api
|
|
|
|
iid, tag = _tmp_item()
|
|
d = panel_api.item_detail(iid)
|
|
assert d["ok"]
|
|
it = d["item"]
|
|
text = it["text"]
|
|
assert "<b>" not in text and "<" not in text, "сырые теги остались"
|
|
assert "&" not in text, "двойное экранирование"
|
|
assert tag in text
|
|
assert isinstance(it["related_same_source"], list)
|
|
assert all(r["source"] == "pytest-panel" for r in it["related_same_source"])
|
|
assert panel_api.item_detail(10 ** 12) == {"error": "not found"}
|
|
|
|
|
|
# ---------- подсказки ----------
|
|
def test_suggest_needs_two_chars_and_finds_source():
|
|
from omni import panel_api
|
|
|
|
assert panel_api.suggest("a") == {"titles": [], "sources": [], "categories": []}
|
|
d = panel_api.suggest("firecrawl")
|
|
assert isinstance(d["titles"], list) and isinstance(d["sources"], list)
|
|
if d["sources"]:
|
|
assert {"name", "category"} <= set(d["sources"][0])
|
|
|
|
|
|
# ---------- watchlist ----------
|
|
def test_watch_add_list_remove():
|
|
from omni import panel_api
|
|
|
|
q = "pytest-watch-%d" % time.time_ns()
|
|
r = panel_api.watch_add(q)
|
|
assert r["ok"] and r["rule"]["q"] == q
|
|
rid = r["rule"]["id"]
|
|
try:
|
|
rules = panel_api.watch_list()["rules"]
|
|
assert any(x["id"] == rid for x in rules)
|
|
assert panel_api.watch_remove(rid)["ok"]
|
|
assert not any(x["id"] == rid for x in panel_api.watch_list()["rules"])
|
|
finally:
|
|
panel_api.watch_remove(rid)
|
|
assert "error" in panel_api.watch_add("x")
|
|
|
|
|
|
# ---------- детали источника ----------
|
|
def test_source_detail_shape():
|
|
from omni import panel_api, sources
|
|
|
|
s = sources.all_sources()[0]
|
|
d = panel_api.source_detail(s.name)
|
|
assert d["ok"] and d["source"]["name"] == s.name
|
|
for k in ("category", "adapter", "kind", "url", "tags", "limit"):
|
|
assert k in d["source"], "нет поля source." + k
|
|
for k in ("state", "stored", "latest", "runs"):
|
|
assert k in d, "нет поля " + k
|
|
assert isinstance(d["latest"], list) and isinstance(d["runs"], list)
|
|
assert d["stored"] >= 0
|
|
assert panel_api.source_detail("нет-такого-источника") == {"error": "not found"}
|
|
|
|
|
|
# ---------- health ----------
|
|
def test_health_counts():
|
|
from omni import panel_api
|
|
|
|
d = panel_api.health()
|
|
assert d["ok"]
|
|
assert d["items"] > 0 and d["sources"] > 0
|
|
assert {"alive", "runs", "uptime", "db_mb"} <= set(d)
|
|
assert 0 <= d["alive"] <= d["sources"], "живых не может быть больше, чем всех"
|
|
assert d["db_mb"] >= 0 and d["uptime"] >= 0
|
|
|
|
|
|
# ---------- tg ----------
|
|
def test_tg_status_shape_and_flag():
|
|
from omni import panel_api
|
|
|
|
d = panel_api.tg_status()
|
|
assert {"enabled", "me", "error", "limit"} <= set(d)
|
|
assert isinstance(d["enabled"], bool)
|
|
if d["enabled"]:
|
|
assert d["session"]
|
|
|
|
|
|
# ---------- digest ----------
|
|
def test_digest_shape_and_day_window():
|
|
from pathlib import Path
|
|
|
|
from omni import panel_api
|
|
|
|
d = panel_api.digest(days=3, limit=10)
|
|
assert d["ok"] and d["days"] == 3 and d["n"] >= 0
|
|
md = d["md"]
|
|
assert md.startswith("# omni · дайджест")
|
|
for head in ("## Главное по score", "## Уязвимости", "## Сливы и утечки", "## Абузы"):
|
|
assert head in md, "в дайджесте нет раздела " + head
|
|
p = Path(d["path"])
|
|
assert p.exists() and p.read_text(encoding="utf-8").startswith("# omni · дайджест")
|
|
# clamp: 0 → 1 день, 999 → 30
|
|
assert panel_api.digest(days=0)["days"] == 1
|
|
assert panel_api.digest(days=999)["days"] == 30
|
|
|
|
|
|
# ---------- export ----------
|
|
def test_export_respects_filters_and_cap():
|
|
from omni import panel_api
|
|
|
|
d = panel_api.export_json(q="firecrawl", limit=5)
|
|
assert d["ok"] and d["count"] == len(d["items"]) <= 5
|
|
assert d["total"] >= d["count"], "total — это всё по фильтру, count — срез"
|
|
for it in d["items"]:
|
|
assert {"id", "title", "url", "source", "published", "types",
|
|
"price", "currency", "score"} <= set(it)
|
|
d2 = panel_api.export_json(typ="leak", limit=3)
|
|
assert d2["ok"] and all("leak" in (it["types"] or "") for it in d2["items"])
|
|
|
|
|
|
# ---------- serve: роуты зарегистрированы ----------
|
|
def test_serve_registers_panel_routes():
|
|
from omni import serve
|
|
|
|
src = inspect.getsource(serve)
|
|
for path in ("/api/item/", "/api/timeline", "/api/suggest", "/api/health",
|
|
"/api/tg/status", "/api/tg/search", "/api/tg/pull", "/api/source",
|
|
"/api/watch", "/api/digest", "/api/export.json"):
|
|
assert path in src, "роут не зарегистрирован: " + path
|
|
assert "browse2" in src, "browse2 не используется в serve"
|
|
|
|
|
|
# ---------- web: панель v2 содержит новые разделы ----------
|
|
def test_web_panel_v2_has_new_sections():
|
|
html = (ROOT / "web" / "index.html").read_text(encoding="utf-8")
|
|
for frag in ('id:"overview"', 'id:"feed"', 'id:"abuse"', 'id:"sources"', 'id:"runs"',
|
|
'id:"tg"', 'id:"keys"', 'id:"digest"', 'id:"watch"',
|
|
"/api/timeline", "/api/tg/status", "/api/suggest", "watchlist",
|
|
"/api/export.json"):
|
|
assert frag in html, "в панели нет " + frag
|
|
|
|
|
|
# ---------- store.browse2: since/src ----------
|
|
def test_browse2_filters_since_and_src():
|
|
from omni import store
|
|
|
|
rows, total = store.browse2(q="", limit=5, sort="seen", since="2026-01-01")
|
|
assert isinstance(rows, list) and total >= len(rows)
|
|
rows2, total2 = store.browse2(q="", limit=5, sort="seen", since="2099-01-01")
|
|
assert total2 == 0 and rows2 == [], "since не применился"
|
|
rows3, _ = store.browse2(q="", limit=200, sort="seen")
|
|
assert rows3, "нет строк для проверки src-фильтра"
|
|
name = rows3[0]["source"]
|
|
rows4, total4 = store.browse2(q="", limit=200, src=name)
|
|
assert total4 > 0 and all(x["source"] == name for x in rows4), "src не применился"
|
|
# src + since вместе
|
|
rows5, total5 = store.browse2(q="", limit=5, src=name, since="2099-01-01")
|
|
assert total5 == 0 and rows5 == []
|