843e2061fe
Les pages /aide, /settings et /logs référeçaient encore Tasker (payant) pour
la détection de présence. Mise à jour de la procédure (Wi-Fi ou géofence +
action Trigger URL POST), des libellés et messages, et du test qui vérifiait
la présence du mot « Tasker ».
💘 Generated with Crush
Assisted-by: Crush:glm-5.2
343 lines
14 KiB
Python
343 lines
14 KiB
Python
"""Smoke tests pour les nouvelles routes /logs et /aide + journalisation présence."""
|
|
import sys
|
|
|
|
# DATA_DIR temporaire est positionné par conftest.py avant l'import de models.
|
|
import models # noqa: F401 (réexport pratique pour les tests ci-dessous)
|
|
|
|
sys.path.insert(0, ".")
|
|
import main
|
|
from fastapi.testclient import TestClient
|
|
|
|
client = TestClient(main.app)
|
|
|
|
|
|
def _login_as(uid: str):
|
|
"""Crée un user avec mdp puis se connecte (renvoie les cookies signés)."""
|
|
import bcrypt
|
|
models.save_auth(uid, email=f"{uid}@x.fr",
|
|
password_hash=bcrypt.hashpw(b"x" * 12, bcrypt.gensalt()).decode())
|
|
r = client.post("/login/password", data={"email": f"{uid}@x.fr", "password": "x" * 12},
|
|
follow_redirects=False)
|
|
assert r.status_code == 302, r.text
|
|
# Récupère le cookie user_id réellement signé par le serveur
|
|
return {"user_id": r.cookies.get("user_id")}
|
|
|
|
|
|
COOKIES = _login_as("alice")
|
|
|
|
|
|
# ── Route /logs ──────────────────────────────────────────────────────────────
|
|
|
|
def test_logs_page_requires_auth():
|
|
# Client frais, sans cookie, doit être redirigé vers /login
|
|
unauth = TestClient(main.app)
|
|
r = unauth.get("/logs", follow_redirects=False)
|
|
assert r.status_code == 302
|
|
assert "/login" in r.headers["location"]
|
|
|
|
|
|
def test_logs_page_empty_for_new_user():
|
|
r = client.get("/logs", cookies=COOKIES)
|
|
assert r.status_code == 200
|
|
assert "Journaux" in r.text
|
|
assert "Aucun événement" in r.text # état initial
|
|
|
|
|
|
def test_logs_page_shows_entries_after_presence():
|
|
token = models.load_notif_config("alice")["token"]
|
|
# Simule une arrivée Automation
|
|
client.post(f"/presence/{token}/arrivee")
|
|
# Simule un départ
|
|
client.post(f"/presence/{token}/depart")
|
|
|
|
r = client.get("/logs", cookies=COOKIES)
|
|
assert r.status_code == 200
|
|
assert "arrivée" in r.text
|
|
assert "départ" in r.text
|
|
assert "Présence" in r.text
|
|
assert "Aucun événement" not in r.text
|
|
|
|
|
|
def test_logs_filter_query_param():
|
|
# Toutes les entrées sont de type "presence" à ce stade
|
|
r_all = client.get("/logs", cookies=COOKIES)
|
|
r_notif = client.get("/logs?filter=notif", cookies=COOKIES)
|
|
r_pres = client.get("/logs?filter=presence", cookies=COOKIES)
|
|
|
|
# Le filtre "notif" ne doit PAS montrer les entrées presence
|
|
assert "arrivée" not in r_notif.text
|
|
# Le filtre "presence" DOIT les montrer
|
|
assert "arrivée" in r_pres.text
|
|
assert "départ" in r_pres.text
|
|
# Le total Tout inclut tout
|
|
assert "arrivée" in r_all.text
|
|
|
|
|
|
# ── Route /aide ──────────────────────────────────────────────────────────────
|
|
|
|
def test_aide_page_requires_auth():
|
|
unauth = TestClient(main.app)
|
|
r = unauth.get("/aide", follow_redirects=False)
|
|
assert r.status_code == 302
|
|
assert "/login" in r.headers["location"]
|
|
|
|
|
|
def test_aide_page_content():
|
|
r = client.get("/aide", cookies=COOKIES)
|
|
assert r.status_code == 200
|
|
# Vérifie la présence des sections clés
|
|
for needle in ["ntfy", "Automation", "Aide", "topic", "géofence", "/logs"]:
|
|
assert needle in r.text, f"missing {needle!r}"
|
|
|
|
|
|
# ── Logging via notifications.send_ntfy (échec) ─────────────────────────────
|
|
|
|
def test_send_ntfy_failure_logs_entry():
|
|
# Serveur injoignable (host public sans résolution) → échec → entrée de log "échec"
|
|
import notifications
|
|
notifications.send_ntfy(
|
|
"https://ntfy-injoignable.invalid", "topic-injoignable",
|
|
"msg test", "titre test", user_id="alice",
|
|
)
|
|
logs = models.load_logs("alice")
|
|
assert any(l["event"] == "échec" and l["type"] == "notif" for l in logs)
|
|
|
|
|
|
# ── Navigation ──────────────────────────────────────────────────────────────
|
|
|
|
def test_nav_links_present():
|
|
r = client.get("/aide", cookies=COOKIES)
|
|
assert 'href="/logs"' in r.text
|
|
assert 'href="/aide"' in r.text
|
|
assert 'href="/settings"' in r.text
|
|
|
|
|
|
# ── Isolation par user ──────────────────────────────────────────────────────
|
|
|
|
def test_logs_isolated_per_user():
|
|
# Bob ne doit pas voir les logs d'alice
|
|
bob_cookies = _login_as("bob")
|
|
r = client.get("/logs", cookies=bob_cookies)
|
|
assert "arrivée" not in r.text
|
|
assert "Aucun événement" in r.text
|
|
|
|
|
|
# ── Rotation des logs ───────────────────────────────────────────────────────
|
|
|
|
def test_log_rotation_caps_at_max():
|
|
# Alice a déjà quelques entrées ; on en ajoute largement > MAX_LOGS
|
|
for i in range(models.MAX_LOGS + 50):
|
|
models.append_log("alice", "notif", "envoyée", f"msg {i}")
|
|
logs = models.load_logs("alice")
|
|
assert len(logs) == models.MAX_LOGS
|
|
# Le plus récent est bien le dernier inséré
|
|
assert logs[0]["message"] == f"msg {models.MAX_LOGS + 49}"
|
|
|
|
|
|
def test_invalid_log_type_rejected():
|
|
try:
|
|
models.append_log("alice", "invalid_type", "x", "y")
|
|
assert False, "doit lever une AssertionError"
|
|
except AssertionError:
|
|
pass
|
|
|
|
|
|
# ── Bouton de test ntfy (/settings/test-ntfy) ────────────────────────────────
|
|
|
|
def test_test_ntfy_requires_auth():
|
|
unauth = TestClient(main.app)
|
|
r = unauth.post("/settings/test-ntfy", follow_redirects=False)
|
|
assert r.status_code == 302
|
|
assert "/login" in r.headers["location"]
|
|
|
|
|
|
def test_test_ntfy_without_topic_returns_error():
|
|
# Alice n'a pas de topic configuré → message d'erreur
|
|
models.save_notif_config("alice", ntfy_topic="", ntfy_server="https://ntfy.sh")
|
|
r = client.post("/settings/test-ntfy", cookies=COOKIES)
|
|
assert r.status_code == 200
|
|
assert "settings-error" in r.text
|
|
assert "topic" in r.text.lower()
|
|
|
|
|
|
def test_test_ntfy_with_bad_server_returns_failure():
|
|
# Topic renseigné mais serveur injoignable → échec
|
|
models.save_notif_config("alice", ntfy_topic="topic-test", ntfy_server="https://ntfy-injoignable.invalid")
|
|
r = client.post("/settings/test-ntfy", cookies=COOKIES)
|
|
assert r.status_code == 200
|
|
assert "settings-error" in r.text
|
|
assert "Échec" in r.text
|
|
# L'entrée de log d'échec doit être présente
|
|
logs = models.load_logs("alice")
|
|
assert any(l["event"] == "échec" and l["type"] == "notif" for l in logs)
|
|
|
|
|
|
def test_test_ntfy_logs_to_journal():
|
|
"""Le bouton de test doit écrire dans le journal (même en échec)."""
|
|
models.save_notif_config("alice", ntfy_topic="topic-test-2", ntfy_server="https://ntfy-injoignable.invalid")
|
|
client.post("/settings/test-ntfy", cookies=COOKIES)
|
|
logs = models.load_logs("alice")
|
|
# L'entrée la plus récente doit être l'échec du test (message contient "Test pointeuse")
|
|
assert any(l["event"] == "échec" and "Test pointeuse" in l["message"] for l in logs)
|
|
|
|
|
|
# ── Jeton d'accès ntfy (serveur protégé) ─────────────────────────────────────
|
|
|
|
def test_send_ntfy_passes_authorization_header(monkeypatch):
|
|
"""Quand un token est fourni, send_ntfy doit envoyer un header Authorization: Bearer."""
|
|
import notifications
|
|
captured = {}
|
|
|
|
class _DummyReq:
|
|
def __init__(self, url, data, headers, method):
|
|
captured["url"] = url
|
|
captured["headers"] = headers
|
|
captured["method"] = method
|
|
|
|
class _DummyResp:
|
|
def close(self): pass
|
|
|
|
def _fake_urlopen(req, timeout):
|
|
captured["timeout"] = timeout
|
|
return _DummyResp()
|
|
|
|
monkeypatch.setattr(notifications.urllib.request, "Request", _DummyReq)
|
|
monkeypatch.setattr(notifications.urllib.request, "urlopen", _fake_urlopen)
|
|
|
|
ok = notifications.send_ntfy(
|
|
"https://ntfy.example", "topic-x",
|
|
"msg", "titre", user_id="alice", token="tk_abc123",
|
|
)
|
|
assert ok is True
|
|
assert captured["headers"]["Authorization"] == "Bearer tk_abc123"
|
|
|
|
|
|
def test_send_ntfy_omits_authorization_when_no_token(monkeypatch):
|
|
"""Sans token, pas de header Authorization."""
|
|
import notifications
|
|
captured = {}
|
|
|
|
class _DummyReq:
|
|
def __init__(self, url, data, headers, method):
|
|
captured["headers"] = headers
|
|
|
|
class _DummyResp:
|
|
def close(self): pass
|
|
|
|
monkeypatch.setattr(notifications.urllib.request, "Request", _DummyReq)
|
|
monkeypatch.setattr(notifications.urllib.request, "urlopen", lambda req, timeout: _DummyResp())
|
|
|
|
notifications.send_ntfy("https://x", "t", "m", "ti", user_id=None, token="")
|
|
assert "Authorization" not in captured["headers"]
|
|
|
|
|
|
def test_settings_save_persists_ntfy_token():
|
|
"""Le token saisi dans /settings doit être persisté."""
|
|
models.save_notif_config("alice", ntfy_topic="t", ntfy_server="https://ntfy.sh")
|
|
client.post("/settings", cookies=COOKIES, data={
|
|
"ntfy_topic": "topic-persisted",
|
|
"ntfy_server": "https://ntfy.arkel.fr",
|
|
"ntfy_token": "tk_persisted_123",
|
|
"rappel_debut_h": "8",
|
|
"rappel_fin_h": "19",
|
|
})
|
|
cfg = models.load_notif_config("alice")
|
|
assert cfg["ntfy_token"] == "tk_persisted_123"
|
|
assert cfg["ntfy_topic"] == "topic-persisted"
|
|
|
|
|
|
# ── Régénération du token de présence (Tasker) ───────────────────────────────
|
|
|
|
def test_regenerate_presence_token_requires_auth():
|
|
unauth = TestClient(main.app)
|
|
r = unauth.post("/settings/regenerate-presence-token", follow_redirects=False)
|
|
assert r.status_code == 302
|
|
assert "/login" in r.headers["location"]
|
|
|
|
|
|
def test_regenerate_presence_token_invalidates_old_urls():
|
|
"""Le nouveau token remplace l'ancien : les anciennes URLs ne marchent plus."""
|
|
old_token = models.load_notif_config("alice")["token"]
|
|
|
|
r = client.post("/settings/regenerate-presence-token", cookies=COOKIES)
|
|
assert r.status_code == 200
|
|
# Le HTML retourné contient de nouvelles URLs avec un nouveau token
|
|
assert old_token not in r.text
|
|
assert "/presence/" in r.text
|
|
assert "Token régénéré" in r.text
|
|
|
|
# L'ancienne URL doit maintenant renvoyer 404 (token inconnu)
|
|
r_old = client.post(f"/presence/{old_token}/arrivee")
|
|
assert r_old.status_code == 404
|
|
|
|
# La nouvelle URL doit fonctionner
|
|
new_cfg = models.load_notif_config("alice")
|
|
r_new = client.post(f"/presence/{new_cfg['token']}/arrivee")
|
|
assert r_new.status_code == 200
|
|
assert r_new.json() == {"ok": True}
|
|
|
|
|
|
def test_regenerate_preserves_other_notif_fields():
|
|
"""La régénération ne doit pas effacer topic/server/token ntfy."""
|
|
models.save_notif_config(
|
|
"alice", ntfy_topic="topic-x", ntfy_server="https://ntfy.arkel.fr",
|
|
ntfy_token="tk_xyz",
|
|
)
|
|
before = models.load_notif_config("alice")
|
|
new_token = models.regenerate_presence_token("alice")
|
|
after = models.load_notif_config("alice")
|
|
|
|
assert after["token"] == new_token
|
|
assert before["token"] != new_token
|
|
# Les autres champs sont préservés
|
|
assert after["ntfy_topic"] == "topic-x"
|
|
assert after["ntfy_server"] == "https://ntfy.arkel.fr"
|
|
assert after["ntfy_token"] == "tk_xyz"
|
|
|
|
|
|
# ── Changements de paramètres tracés dans le journal ─────────────────────────
|
|
|
|
def test_settings_save_logged():
|
|
"""Sauvegarder les réglages ntfy doit écrire une entrée de journal."""
|
|
models.save_notif_config("alice", ntfy_topic="t", ntfy_server="https://ntfy.sh")
|
|
before = len([l for l in models.load_logs("alice") if l["type"] == "settings"])
|
|
client.post("/settings", cookies=COOKIES, data={
|
|
"ntfy_topic": "topic-new",
|
|
"ntfy_server": "https://ntfy.arkel.fr",
|
|
"ntfy_token": "tk_abc",
|
|
"rappel_debut_h": "8",
|
|
"rappel_fin_h": "19",
|
|
})
|
|
after = [l for l in models.load_logs("alice") if l["type"] == "settings"]
|
|
assert len(after) == before + 1
|
|
latest = after[0]
|
|
assert latest["event"] == "Réglages ntfy"
|
|
assert "topic-new" in latest["message"]
|
|
assert "ntfy.arkel.fr" in latest["message"]
|
|
|
|
|
|
def test_settings_plages_save_logged():
|
|
"""Sauvegarder les plages horaires doit écrire une entrée de journal."""
|
|
before = len([l for l in models.load_logs("alice") if l["type"] == "settings"])
|
|
client.post("/settings/plages", cookies=COOKIES, data={
|
|
"matin_debut": "09:00", "matin_fin": "12:00",
|
|
"aprem_debut": "14:00", "aprem_fin": "17:00",
|
|
"pause_dejeuner_fin": "13:30",
|
|
"arrivee_visee": "", "depart_vise": "",
|
|
})
|
|
after = [l for l in models.load_logs("alice") if l["type"] == "settings"]
|
|
assert len(after) == before + 1
|
|
latest = after[0]
|
|
assert latest["event"] == "Plages horaires"
|
|
assert "09:00" in latest["message"]
|
|
|
|
|
|
def test_regenerate_presence_token_logged():
|
|
"""Régénérer le token de présence doit écrire une entrée de journal."""
|
|
before = len([l for l in models.load_logs("alice") if l["type"] == "settings"])
|
|
client.post("/settings/regenerate-presence-token", cookies=COOKIES)
|
|
after = [l for l in models.load_logs("alice") if l["type"] == "settings"]
|
|
assert len(after) == before + 1
|
|
latest = after[0]
|
|
assert "régénéré" in latest["event"].lower()
|