c1a2f38044
GET /presence/{token}/arrivee et /depart modifiaient l'état (save_presence,
append_log) : c'est anti-pattern REST, CSRF-trivial via <img src="..."> si
le token fuite, et le token se retrouvait dans les logs reverse-proxy,
l'historique navigateur et les referrers.
Les deux routes passent en POST. C'est un breaking change pour les clients
Tasker actuels, mais Tasker supporte HTTP POST nativement et la doc
docs/notifications.md est mise à jour (Tasker : HTTP Request méthode POST ;
script NetworkManager : curl -X POST).
Les tests existants qui utilisaient client.get(...) sur ces routes passent
en client.post(...). Un nouveau test vérifie que le GET renvoie désormais
405 Method Not Allowed et que le POST continue de fonctionner.
💘 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 Tasker
|
|
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", "Tasker", "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()
|