2320acf3c5
- omni/vault.py: пул ключей парсеров (env-first, data/vault.json, ротация, cooldown 401/402/403=1ч, 429=10м), только маски в выводе - tools/build_vault.py: сборка пула из MASTER_KEYS.csv и локальных дампов (значения не в git) - tools/vault_check.py: живая проверка с маскированием - searchers: firecrawl/tavily/exa/serper/jina/apify берут ключ из vault, если нет env - omni radar: домены+каналы из radar_server.db -> реестр, --import-posts заливает посты каналов - radar-источники помечены opt_in: общий обход не тормозят (869 против 3337 всего) - omni/sources_cn.py + tools/wave_cn.py: 9 живых китайских источников (52pojie, nodeseek, kanxue, anquanke, 4hou, xz.aliyun, ithome, cnbeta, v2ex) - net.get(proxy=...): принудительный прокси для CF-площадок - /api/keys + строка ключей в панели - tests/test_vault_radar.py: 7 тестов, всего 29 passed
61 lines
2.3 KiB
Python
61 lines
2.3 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()
|
|
assert u not in seen, "дубль url: %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"]) |