132 lines
4.1 KiB
Python
132 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from starlette.testclient import TestClient
|
|
|
|
import dispatcher
|
|
from watchdog_store import ensure_watchdog_tables
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
SCHEMA_SQL = (ROOT / "config" / "schema.sql").read_text()
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path, monkeypatch):
|
|
db_path = tmp_path / "wd.db"
|
|
conn = sqlite3.connect(db_path)
|
|
conn.executescript(SCHEMA_SQL)
|
|
ensure_watchdog_tables(conn)
|
|
conn.close()
|
|
monkeypatch.setattr(dispatcher.cfg.database, "path", str(db_path))
|
|
monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
|
|
monkeypatch.setattr(dispatcher.cfg.routing, "require_vision", False)
|
|
monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")
|
|
with TestClient(dispatcher.app) as c:
|
|
yield c
|
|
|
|
|
|
def test_watchdog_status_empty_db(client):
|
|
resp = client.get("/admin/api/watchdog/status")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["last_tick"] is None
|
|
assert data["open_alerts"] == 0
|
|
|
|
|
|
def test_watchdog_loops_empty_db(client):
|
|
resp = client.get("/admin/api/watchdog/loops")
|
|
assert resp.status_code == 200
|
|
assert resp.json() == []
|
|
|
|
|
|
def test_watchdog_channels_empty_db(client):
|
|
resp = client.get("/admin/api/watchdog/channels")
|
|
assert resp.status_code == 200
|
|
assert resp.json() == []
|
|
|
|
|
|
def test_watchdog_channels_create(client):
|
|
resp = client.post(
|
|
"/admin/api/watchdog/channels",
|
|
json={"channel_name": "desktop", "enabled": True, "min_severity": "warning"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["ok"] is True
|
|
listing = client.get("/admin/api/watchdog/channels").json()
|
|
assert len(listing) == 1
|
|
assert listing[0]["channel_name"] == "desktop"
|
|
assert listing[0]["enabled"] == 1
|
|
|
|
|
|
def test_watchdog_channels_update_min_severity(client):
|
|
client.post(
|
|
"/admin/api/watchdog/channels",
|
|
json={"channel_name": "desktop", "enabled": True, "min_severity": "warning"},
|
|
)
|
|
client.post(
|
|
"/admin/api/watchdog/channels",
|
|
json={"channel_name": "desktop", "min_severity": "critical"},
|
|
)
|
|
listing = client.get("/admin/api/watchdog/channels").json()
|
|
assert listing[0]["min_severity"] == "critical"
|
|
assert listing[0]["enabled"] == 1
|
|
|
|
|
|
def test_watchdog_channels_requires_name(client):
|
|
resp = client.post("/admin/api/watchdog/channels", json={})
|
|
assert resp.status_code == 422
|
|
|
|
|
|
def test_watchdog_channels_invalid_severity(client):
|
|
resp = client.post(
|
|
"/admin/api/watchdog/channels",
|
|
json={"channel_name": "x", "min_severity": "bogus"},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
def test_watchdog_test_alert_no_channels(client):
|
|
resp = client.post(
|
|
"/admin/api/watchdog/test-alert",
|
|
json={"severity": "warning"},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["sent"] == 0
|
|
|
|
|
|
def test_watchdog_test_alert_invalid_severity(client):
|
|
resp = client.post(
|
|
"/admin/api/watchdog/test-alert",
|
|
json={"severity": "bogus"},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
def test_watchdog_status_after_tick(tmp_path, monkeypatch):
|
|
import sqlite3 as _sqlite3
|
|
db_path = tmp_path / "wd_ticks.db"
|
|
conn = _sqlite3.connect(db_path)
|
|
conn.row_factory = _sqlite3.Row
|
|
conn.executescript(SCHEMA_SQL)
|
|
ensure_watchdog_tables(conn)
|
|
conn.execute(
|
|
"INSERT INTO watchdog_ticks (ticked_at, sessions_seen, outcome) VALUES (?, ?, ?)",
|
|
("2026-09-25T10:00:00", 5, "ok"),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
monkeypatch.setattr(dispatcher.cfg.database, "path", str(db_path))
|
|
monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
|
|
monkeypatch.setattr(dispatcher.cfg.routing, "require_vision", False)
|
|
monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")
|
|
with TestClient(dispatcher.app) as c:
|
|
resp = c.get("/admin/api/watchdog/status")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["last_tick"] is not None
|
|
assert data["last_tick"]["sessions_seen"] == 5
|