61 lines
2.4 KiB
Python
61 lines
2.4 KiB
Python
"""Тесты на «одно место»: vault ключей, opt-in радара, китайские источники."""
|
|
import json
|
|
|
|
from omni import sources, vault
|
|
|
|
|
|
def test_vault_env_first(monkeypatch):
|
|
monkeypatch.setenv("FIRECRAWL_API_KEY", "fc-env-key")
|
|
assert vault.get("firecrawl") == "fc-env-key"
|
|
|
|
|
|
def test_vault_mask_never_shows_value():
|
|
m = vault.mask("fc-1234567890abcdef")
|
|
assert "1234567890" not in m
|
|
assert m.endswith("ef")
|
|
assert vault.mask("short") == "…"
|
|
|
|
|
|
def test_vault_roundrobin_and_cooldown(monkeypatch):
|
|
data = {"zzsvc": [{"key": "k1"}, {"key": "k2"}]}
|
|
monkeypatch.setattr(vault, "_load", lambda: (data, {}))
|
|
monkeypatch.setattr(vault, "ENV", {})
|
|
got = {vault.get("zzsvc", offset=i) for i in range(6)}
|
|
assert got == {"k1", "k2"}, "пул должен ротироваться по обоим ключам"
|
|
monkeypatch.setattr(vault, "_load", lambda: (data, {"zzsvc": {"k1": 9e12, "k2": 9e12}}))
|
|
assert vault.get("zzsvc") in {"k1", "k2"}, "при остывании всё равно отдаём ключ"
|
|
|
|
|
|
def test_radar_sources_opt_in():
|
|
r = sources.radar_sources()
|
|
if not r:
|
|
return
|
|
assert all((s.extra or {}).get("opt_in") for s in r)
|
|
plain = sources.enabled_sources()
|
|
assert not any((s.extra or {}).get("opt_in") for s in plain), "радар не должен лезть в общий обход"
|
|
assert sources.pick(tags="radar"), "по тегу radar источники должны находиться"
|
|
|
|
|
|
def test_cn_sources_registered():
|
|
cn = [s for s in sources.all_sources() if s.name.startswith("cn_")]
|
|
assert cn, "китайские источники должны быть в реестре"
|
|
assert all(s.url.startswith("http") for s in cn)
|
|
assert all(s.adapter in ("html_listing", "rss") for s in cn)
|
|
|
|
|
|
def test_all_sources_no_dup_url():
|
|
seen = {}
|
|
for s in sources.all_sources():
|
|
u = ((s.url or "").rstrip("/").lower(), s.adapter)
|
|
assert u not in seen, "дубль (url, adapter): %s" % s.name
|
|
seen[u] = s.name
|
|
|
|
|
|
def test_radar_json_shape():
|
|
from pathlib import Path
|
|
p = Path("data/radar_sources.json")
|
|
if not p.exists():
|
|
return
|
|
d = json.loads(p.read_text(encoding="utf-8"))
|
|
assert isinstance(d.get("domains"), list) and isinstance(d.get("channels"), list)
|
|
assert all(len(x) >= 2 for x in d["domains"]) |