абузы: free-панели new-api на вкладке (роут /api/panels + грид), CSS вынесен из JS, панельный классификатор даёт freeapi-вес, тесты

tests/test_panel_api.py

@@ -5,11 +5,38 @@ 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")
 
 
+@pytest.fixture(autouse=True)
+def isolated_panel_data(tmp_path, monkeypatch):
+    """Never touch the operator's database, watchlist, digest or TG session."""
+    from omni import abuse, config, panel_api, store, tgsess
+    from omni.model import Item
+
+    monkeypatch.setattr(config, "DATA", tmp_path)
+    monkeypatch.setattr(config, "DB", tmp_path / "test.sqlite3")
+    monkeypatch.setattr(abuse, "DB", config.DB)
+    monkeypatch.setattr(panel_api, "WATCH", tmp_path / "watch.json")
+    monkeypatch.setattr(config, "tg_cfg", lambda: {"session": "", "limit": 100})
+    monkeypatch.setattr(tgsess, "enabled", lambda: False)
+    today = time.strftime("%Y-%m-%d")
+    store.add([
+        Item(provider="pytest", kind="post", source="pytest-panel",
+             title="Firecrawl security dataset leak announcement",
+             text="Public incident report for regression coverage.",
+             url="https://example.invalid/seed", published=today),
+        Item(provider="pytest", kind="post", source="pytest-other",
+             title="Guide to observability and performance metrics",
+             text="A completely different article about tracing.",
+             url="https://example.invalid/other", published=today),
+    ])
+
+
 def _tmp_item(tag=None):
     from omni import store
     from omni.model import Item
@@ -124,14 +151,19 @@ def test_health_counts():
 
 
 # ---------- tg ----------
-def test_tg_status_shape_and_flag():
-    from omni import panel_api
+@pytest.mark.parametrize("enabled", [False, True])
+def test_tg_status_shape_and_flag(monkeypatch, enabled):
+    from omni import config, panel_api, tgsess
 
+    monkeypatch.setattr(tgsess, "enabled", lambda: enabled)
+    monkeypatch.setattr(tgsess, "me", lambda: {"id": 123, "first_name": "Test"})
+    monkeypatch.setattr(config, "tg_cfg",
+                        lambda: {"session": "test-account", "limit": 100})
     d = panel_api.tg_status()
-    assert {"enabled", "me", "error", "limit"} <= set(d)
-    assert isinstance(d["enabled"], bool)
-    if d["enabled"]:
-        assert d["session"]
+    assert {"enabled", "me", "error", "limit", "session"} <= set(d)
+    assert d["enabled"] is enabled
+    assert d["me"] == ({"id": 123, "first_name": "Test"} if enabled else None)
+    assert d["error"] is None
 
 
 # ---------- digest ----------
@@ -205,3 +237,170 @@ def test_browse2_filters_since_and_src():
     # src + since вместе
     rows5, total5 = store.browse2(q="", limit=5, src=name, since="2099-01-01")
     assert total5 == 0 and rows5 == []
+
+
+def test_registry_dedupes_url_and_name(monkeypatch):
+    from omni import sources, source_state
+
+    first = sources.S("first", "security", "rss", "https://example.invalid/feed")
+    same_url = sources.S("alias", "security", "rss", first.url + "/")
+    same_name = sources.S("FIRST", "security", "rss", "https://example.invalid/mirror")
+    other = sources.S("other", "security", "rss", "https://example.invalid/other")
+    monkeypatch.setattr(sources, "SOURCES", [first, same_url, same_name, other])
+    monkeypatch.setattr(sources, "_gen_sources", lambda: [])
+    monkeypatch.setattr(sources, "_fresh_sources", lambda factory: [])
+    monkeypatch.setattr(sources, "_quality_sources", lambda factory: [])
+    monkeypatch.setattr(sources, "radar_sources", lambda: [])
+    monkeypatch.setattr(sources, "panel_sources", lambda: [])
+    monkeypatch.setattr(source_state, "custom", lambda: [])
+    assert sources.all_sources() == [first, other]
+
+
+def test_fresh_sources_have_valid_metadata():
+    from omni import feeds, sources
+    from omni.sources_fresh import fresh
+
+    rows = fresh(sources.S)
+    assert rows
+    assert all(isinstance(s.tags, tuple) for s in rows)
+    assert all(s.adapter in feeds.ADAPTERS for s in rows)
+    assert len({s.name for s in rows}) == len(rows)
+
+
+def test_quality_wave_is_integrated_without_collisions():
+    from omni import feeds, sources
+    from omni.sources_quality import quality_sources
+
+    wave = quality_sources(sources.S)
+    registry = sources.all_sources()
+    assert len(wave) >= 30
+    assert all(isinstance(s.tags, tuple) and s.adapter in feeds.ADAPTERS for s in wave)
+    assert {s.name for s in wave} <= {s.name for s in registry}
+    assert len(registry) == len({s.name.strip().lower() for s in registry})
+    assert len(registry) == len({s.url.rstrip("/").lower() for s in registry})
+
+
+@pytest.mark.parametrize("tags", [(), ("news",), ("news", "cve")])
+def test_fresh_generator_preserves_tag_tuples(tmp_path, tags):
+    import runpy
+
+    from omni import sources
+
+    emit = runpy.run_path(str(ROOT / "tools" / "probe_fresh.py"))["emit"]
+    row = {"name": "test_feed", "category": "security", "kind": "rss",
+           "adapter": "rss", "url": "https://example.invalid/feed", "tags": tags}
+    path = tmp_path / "generated_sources.py"
+    assert emit([row], [], path, dedupe=False) == 1
+    generated = runpy.run_path(str(path))["fresh"](sources.S)
+    assert generated[0].tags == tags
+    assert isinstance(generated[0].tags, tuple)
+
+
+def test_timeline_local_day_and_rolling_24_hours(monkeypatch):
+    import datetime
+    from omni import panel_api, store
+
+    now = datetime.datetime(2026, 10, 20, 1, 0).timestamp()
+    class FrozenDate(datetime.date):
+        @classmethod
+        def today(cls):
+            return cls(2026, 10, 20)
+
+    monkeypatch.setattr(panel_api._dt, "date", FrozenDate)
+    monkeypatch.setattr(panel_api.time, "time", lambda: now)
+    with store.db() as c:
+        c.execute("UPDATE items SET seen=? WHERE source='pytest-panel'", (now - 1800,))
+        c.execute("UPDATE items SET seen=? WHERE source='pytest-other'", (now - 23 * 3600,))
+    data = panel_api.timeline(days=2, mode="seen")
+    assert data["last24"] == 2, "rolling 24 hours must include yesterday evening"
+    assert data["today"] == 1, "local 00:30 import belongs to today's bar"
+    assert [p["n"] for p in data["points"]] == [1, 1]
+
+
+def test_run_stats_count_partial_successes():
+    from omni import store
+
+    for status in ("ok", "ok 5/10", "пусто", "Exception: timeout"):
+        store.log_run("pytest", "test", "", 0, 0, status)
+    stats = store.stats()
+    assert stats["runs"] == 4
+    assert stats["runs_ok"] == 2
+
+
+def test_timeline_last24_respects_publication_mode(monkeypatch):
+    import datetime
+    from omni import panel_api, store
+
+    now = time.time()
+    monkeypatch.setattr(panel_api.time, "time", lambda: now)
+    old = datetime.datetime.fromtimestamp(now - 48 * 3600, datetime.timezone.utc).isoformat()
+    future = datetime.datetime.fromtimestamp(now + 48 * 3600, datetime.timezone.utc).isoformat()
+    with store.db() as c:
+        c.execute("UPDATE items SET seen=?", (now - 60,))
+        c.execute("UPDATE items SET published=? WHERE source='pytest-panel'", (old,))
+        c.execute("UPDATE items SET published=? WHERE source='pytest-other'", (future,))
+    assert panel_api.timeline(mode="seen")["last24"] == 2
+    assert panel_api.timeline(mode="pub")["last24"] == 0
+
+
+def test_http_exports_match_feed_filters_and_item_find(monkeypatch):
+    import csv
+    import datetime
+    import io
+    import json
+    import threading
+    import urllib.request
+    from http.server import ThreadingHTTPServer
+    from omni import serve, store
+    from omni.model import Item
+
+    today = datetime.date.today().isoformat()
+    old = (datetime.date.today() - datetime.timedelta(days=10)).isoformat()
+    rows = [
+        Item("pytest", "post", source="quality-http", title="promo code: QUAL2026 $20 30 days",
+             text="Official promotional offer for testing.", url="https://example.invalid/quality-new", published=today),
+        Item("pytest", "post", source="quality-http", title="Legacy license costs $99",
+             text="An old pricing page.", url="https://example.invalid/quality-old", published=old),
+        Item("pytest", "post", source="quality-http", title="Latest debugging guide",
+             text="A free guide to tracing request latency.", url="https://example.invalid/quality-free", published=today),
+    ]
+    store.add(rows)
+    for status in (" OK ", "ok 2/5", "пусто", "Exception: timeout", None):
+        store.log_run("pytest", "quality-http", "", 0, 0, status)
+    monkeypatch.setattr(serve, "UI_TOKEN", "")
+    server = ThreadingHTTPServer(("127.0.0.1", 0), serve.Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    base = "http://127.0.0.1:%s" % server.server_address[1]
+    query = "?src=quality-http&days=1&priced=1&limit=1000"
+    try:
+        def get(path):
+            with urllib.request.urlopen(base + path, timeout=10) as res:
+                assert res.status == 200
+                return res.read().decode("utf-8-sig")
+
+        feed = json.loads(get("/api/items" + query))
+        export = json.loads(get("/api/export.json" + query))
+        csv_rows = list(csv.DictReader(io.StringIO(get("/api/export.csv" + query)), delimiter=";"))
+        assert feed["total"] == export["total"] == 1
+        assert [r["url"] for r in feed["items"]] == [rows[0].url]
+        assert [r["url"] for r in export["items"]] == [rows[0].url]
+        assert [r["url"] for r in csv_rows] == [rows[0].url]
+        assert "QUAL2026" in feed["items"][0]["find"]["codes"]
+        runs = json.loads(get("/api/runs?limit=2"))
+        assert runs["total"] == 5 and runs["shown"] == 2
+        assert runs["by_status"] == {"ok": 2, "empty": 1, "err": 1, "none": 1}
+        states = {r["state"] for r in json.loads(get("/api/runs?limit=10"))["items"]}
+        assert states == {"ok", "empty", "err", "none"}
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+
+def test_web_has_free_panel_block():
+    """Вкладка абузов обязана показывать free-панели new-api."""
+    html = open(os.path.join(ROOT, "web", "index.html"), encoding="utf-8").read()
+    assert "loadPanels" in html
+    assert "/api/panels" in html
+    assert "pn-grid" in html
+    assert ".pn-grid{" in html