tests/test_watch.py
Rama main · réplica de lectura del repositorio original.
"""Tests des outils de supervision élargis (lecture seule via API)."""
from app import watch
# ── Backups (canal SSH forcé) ─────────────────────────────────────────────
async def test_backups_canal_indisponible_message_clair(monkeypatch):
async def fake_channel(cmd):
return "Canal SSH lecture seule indisponible pour le moment."
monkeypatch.setattr(watch, "_status_channel", fake_channel)
report = await watch.backups_report()
assert "indisponible" in report
async def test_backups_parse_la_sortie_du_canal_force(monkeypatch):
async def fake_channel(cmd):
assert cmd == "vektor-status backups"
return "OK|vzdump-lxc-101-2026_09_23.tar.zst|1G|15 h\nOLD|vzdump-lxc-102-2026_09_20.tar.zst|18G|3 j"
monkeypatch.setattr(watch, "_status_channel", fake_channel)
report = await watch.backups_report()
assert "vzdump-lxc-101" in report
assert "vzdump-lxc-102" in report
assert "🟢" in report and "🔴" in report # vieux backup signalé
assert "2 fichiers vus" in report
async def test_backups_vide_dit_clairement_quaucun_fichier(monkeypatch):
async def fake_channel(cmd):
return "NONE|aucun vzdump trouve|0G|?"
monkeypatch.setattr(watch, "_status_channel", fake_channel)
report = await watch.backups_report()
assert "Aucun fichier de backup" in report
# ── DNS (multi-transport : UDP wire, DoH wire, DoH JSON) ─────────────────
class FakeDoHResponse:
def __init__(self, status_code=200, payload=None, content=b""):
self.status_code = status_code
self._payload = payload or {}
self.content = content
def json(self):
return self._payload
class FakeDoHClient:
def __init__(self, responses_by_url):
self._responses = responses_by_url
async def __aenter__(self):
return self
async def __aexit__(self, *exc):
return False
async def get(self, url, **kwargs):
params = kwargs.get("params") or {}
if params: # forme DoH JSON : l'URL réelle porte la query string
from urllib.parse import urlencode
url = url + "?" + urlencode(params)
return self._responses[url]
def _fake_udp_reply(packet, rcode=0, ip="93.184.216.34"):
"""Réponse DNS wire minimale : question recopiée + 1 A."""
offset = 12 # l'entête fait 12 octets ; scanner le QNAME à partir de là
while packet[offset] != 0:
offset += packet[offset] + 1
qend = offset + 5 # octet nul + QTYPE + QCLASS
question = packet[12:qend]
answer = b"\xc0\x0c" + bytes([0, 1, 0, 1, 0, 0, 0, 60, 0, 4]) + bytes(
int(p) for p in ip.split(".")
)
return (
packet[:2] # echo de l'ID de transaction
+ bytes([0x81, 0x80 | rcode])
+ b"\x00\x01\x00\x01\x00\x00\x00\x00"
+ question
+ answer
)
async def test_dns_sonde_udp_wire_et_json(monkeypatch):
sent = {}
def fake_udp(host, port, packet, timeout=4.0):
sent["target"] = f"{host}:{port}"
return _fake_udp_reply(packet)
monkeypatch.setattr(watch, "DNS_RESOLVERS", {
"pihole": "udp://192.0.2.53:53",
"cloud": "http://c.test/dns-query",
})
monkeypatch.setattr(watch, "_udp_wire_query", fake_udp)
monkeypatch.setattr(watch.httpx, "AsyncClient", lambda **k: FakeDoHClient({
"http://c.test/dns-query?name=example.com&type=A": FakeDoHResponse(
payload={"Status": 0, "Answer": [{"type": 1, "data": "93.184.216.34"}]}
),
}))
report = await watch.dns_report()
assert sent["target"] == "192.0.2.53:53"
assert "🟢 pihole" in report and "93.184.216.34" in report
assert "🟢 cloud" in report
# l'adresse du résolveur sondé doit être visible (sinon on confond les
# IP de la réponse avec celles du résolveur)
assert "192.0.2.53:53" in report
assert "→" in report
async def test_dns_doh_wire_parse_le_rcode(monkeypatch):
good = _fake_udp_reply(watch._encode_wire_query("example.com"))
monkeypatch.setattr(watch, "DNS_RESOLVERS", {
"ok": "https://doh.test/dns-query?{dns}",
"ko": "https://doh.test/dns-query?{dns}",
})
monkeypatch.setattr(watch.httpx, "AsyncClient", lambda **k: FakeDoHClient({
"https://doh.test/dns-query?" + watch._encode_doh_query("example.com"): good and FakeDoHResponse(content=good),
}))
# le second résolveur partage la même URL : on le fait échouer via un socket KO
async def boom(*a, **k):
raise OSError("down")
monkeypatch.setattr(watch, "DNS_RESOLVERS", {"ko": "udp://192.0.2.1:53"})
report = await watch.dns_report()
assert "injoignable" in report
async def test_dns_aucun_resolveur_configure(monkeypatch):
monkeypatch.setattr(watch, "DNS_RESOLVERS", {})
report = await watch.dns_report()
assert "aucun résolveur" in report
async def test_dns_sonde_interne_resolue_nxdomain_et_injoignable(monkeypatch):
"""Section noms internes : le nom interne est résolu par le Pi-hole qui
sert le mesh, NXDOMAIN (🟡) sur le récursif, 🔴 sur un résolveur mort."""
monkeypatch.setattr(watch, "DNS_INTERNAL_PROBE", "pve.internal.example")
monkeypatch.setattr(watch, "DNS_RESOLVERS", {
"pihole": "udp://192.0.2.53:53",
"recursif": "udp://192.0.2.54:5335",
"mort": "udp://192.0.2.55:53",
})
def fake_udp(host, port, packet, timeout=4.0):
if host == "192.0.2.53":
return _fake_udp_reply(packet, ip="100.64.0.1")
if host == "192.0.2.54":
return _fake_udp_reply(packet, rcode=3)
raise OSError("down")
monkeypatch.setattr(watch, "_udp_wire_query", fake_udp)
report = await watch.dns_report()
assert "🏠 **Noms internes**" in report and "pve.internal.example" in report
assert "🟢 pihole" in report and "100.64.0.1" in report
assert "🟡 recursif" in report and "NXDOMAIN" in report
assert "🔴 mort" in report and "injoignable" in report
# la légende distingue le NXDOMAIN normal du récursif d'une vraie panne
assert "récursif public" in report
async def test_dns_sonde_interne_absente_si_non_configuree(monkeypatch):
monkeypatch.setattr(watch, "DNS_INTERNAL_PROBE", "")
monkeypatch.setattr(watch, "DNS_RESOLVERS", {"pihole": "udp://192.0.2.53:53"})
monkeypatch.setattr(watch, "_udp_wire_query", lambda *a, **k: _fake_udp_reply(a[2]))
report = await watch.dns_report()
assert "Noms internes" not in report
def test_parse_wire_extrait_rcode_et_ips():
packet = watch._encode_wire_query("example.com")
rcode, ips = watch._parse_wire_response(_fake_udp_reply(packet, ip="192.0.2.7"))
assert rcode == 0
assert ips == ["192.0.2.7"]
# ── Monitoring (Healthchecks.io) ──────────────────────────────────────────
async def test_monitoring_non_configure_degrade_proprement(monkeypatch):
monkeypatch.setattr(watch, "HC_API_URL", "")
monkeypatch.setattr(watch, "HC_READ_TOKEN", "")
report = await watch.monitoring_report()
assert "non configuré" in report
assert "VEKTOR_HC_READ_TOKEN" in report
async def test_monitoring_resume_les_checks(monkeypatch):
monkeypatch.setattr(watch, "HC_API_URL", "http://hc.test")
monkeypatch.setattr(watch, "HC_READ_TOKEN", "read-token")
class FakeHCResponse:
status_code = 200
def json(self):
return {"checks": [
{"name": "stream-guard", "status": "up"},
{"name": "nightly-backup", "status": "late"},
{"name": "s3-sync", "status": "down"},
]}
class FakeHCClient:
def __init__(self, **kwargs):
pass
async def __aenter__(self):
return self
async def __aexit__(self, *exc):
return False
async def get(self, url, **kwargs):
assert "X-Api-Key" in kwargs["headers"]
return FakeHCResponse()
monkeypatch.setattr(watch.httpx, "AsyncClient", FakeHCClient)
report = await watch.monitoring_report()
assert "3 checks" in report
assert "1 ok" in report and "1 en retard" in report and "1 down" in report
assert "s3-sync" in report and "nightly-backup" in report
# ── Exposition au LLM (tool-calling) ─────────────────────────────────────
async def test_les_trois_nouveaux_outils_sont_exposes_lecture_seule():
from app import graph
names = {tool_item.name for tool_item in graph.READONLY_TOOLS}
assert {"derniers_backups", "etat_dns", "etat_monitoring"} <= names