FabriceandClaude Opus 5.5
ada3d76f50
- admin.sh : RTG lancé directement sur le Mac (plus de conteneur), nginx joint l'interface par host.docker.internal:8765 ; variante conteneur conservée (admin-docker.sh, docker-compose-conteneur.yml) - Compression : réglages lus dans les fichiers (CRF, codec), refaire seulement le 1080p ou le 720p, seulement si les réglages ont changé ; réglages vidéo propres à chaque voyage ; « Compression actuelle » affichée - Plus d'arrêt automatique après 2 h d'inactivité (--arret-inactif, 0 par défaut) ; arrêt propre qui attend la fin des encodages en cours - Aide contextuelle dans l'interface (rtg/interface/aide) ; README centré sur l'installation et la mise en place - Cœurs attribués au conteneur détectés (cgroup) et signalés Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Mf2kATMtqVABHSYdxJMnim
241 lines
12 KiB
Python
241 lines
12 KiB
Python
"""Interface : jours imposés aux traces sans heures, suggestion d'après les
|
|
photos, et quelques échanges avec le serveur local."""
|
|
import os
|
|
import threading
|
|
import urllib.parse
|
|
import urllib.request
|
|
from datetime import date
|
|
from http.server import ThreadingHTTPServer
|
|
|
|
import pytest
|
|
|
|
from rtg import config, sources
|
|
from rtg import acces, corrections
|
|
from rtg.interface import serveur
|
|
from rtg.interface.securite import Gardien
|
|
from test_chaine import _photo, pytestmark # noqa: F401 (mêmes prérequis)
|
|
|
|
|
|
def _etape(chemin, lat0, lon0):
|
|
pts = "".join(f'<trkpt lat="{lat0 + i * 0.01:.5f}" lon="{lon0 + i * 0.012:.5f}"></trkpt>' for i in range(50))
|
|
with open(chemin, "w") as f:
|
|
f.write(f'<?xml version="1.0"?><gpx version="1.1" creator="68degres.fr" '
|
|
f'xmlns="http://www.topografix.com/GPX/1/1"><trk><trkseg>{pts}</trkseg></trk></gpx>')
|
|
|
|
|
|
@pytest.fixture
|
|
def voyage_sans_heures(tmp_path):
|
|
racine = tmp_path / "site"
|
|
voyage = racine / "Ecosse-2027"
|
|
gpx, photos = tmp_path / "gpx", tmp_path / "photos"
|
|
for d in (voyage, gpx, photos):
|
|
d.mkdir(parents=True)
|
|
_etape(gpx / "Etape 1.gpx", 55.9, 3.2)
|
|
_etape(gpx / "Etape 2.gpx", 56.7, 2.24)
|
|
_photo(str(photos / "P1.jpg"), "2027:06:11 10:00:00", gps=(56.75, 2.30))
|
|
config.ecrire_toml(str(racine / "site.toml"), {"titre": "Test"})
|
|
config.ecrire_toml(str(voyage / "voyage.toml"), {
|
|
"titre": "Écosse", "sources": {"traces": [str(gpx)], "videos": [], "photos": [str(photos)]}})
|
|
return racine, voyage
|
|
|
|
|
|
def _cfg(racine, voyage):
|
|
return config.reglages_voyage(str(voyage), config.reglages_site(str(racine)))
|
|
|
|
|
|
def test_jour_impose_et_suggestion(voyage_sans_heures):
|
|
racine, voyage = voyage_sans_heures
|
|
a = sources.analyser(str(voyage), _cfg(racine, voyage))
|
|
assert not a.journees and len(a.non_datees) == 2
|
|
medias = sources.positions_medias(str(voyage), a)
|
|
par_nom = {os.path.basename(f): p for f, p in a.non_datees.items()}
|
|
assert sources.suggerer_jour(par_nom["Etape 2.gpx"], medias) == (date(2027, 6, 11), 1)
|
|
assert sources.suggerer_jour(par_nom["Etape 1.gpx"], medias) == (None, 0)
|
|
|
|
config.modifier_toml(str(voyage / "voyage.toml"), {"dates_traces": {"Etape 1.gpx": "2027-06-10"}})
|
|
a = sources.analyser(str(voyage), _cfg(racine, voyage))
|
|
assert sorted(a.journees) == [date(2027, 6, 10)]
|
|
assert [os.path.basename(f) for f in a.non_datees] == ["Etape 2.gpx"]
|
|
# Le titre et les sources ont survécu à la réécriture du fichier
|
|
assert config.lire_toml(str(voyage / "voyage.toml"))["titre"] == "Écosse"
|
|
|
|
|
|
@pytest.fixture
|
|
def client(voyage_sans_heures):
|
|
racine, voyage = voyage_sans_heures
|
|
app = serveur.Application(str(racine))
|
|
srv = ThreadingHTTPServer(("127.0.0.1", 0), None)
|
|
port = srv.server_address[1]
|
|
gardien = Gardien(port)
|
|
srv.RequestHandlerClass = serveur.creer_gestionnaire(app, port, gardien)
|
|
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
|
sid = gardien.ouvrir_session()
|
|
|
|
def appel(chemin, donnees=None, htmx=True, session=True):
|
|
req = urllib.request.Request(f"http://127.0.0.1:{port}{chemin}",
|
|
data=urllib.parse.urlencode(donnees).encode() if donnees else None)
|
|
if htmx:
|
|
req.add_header("HX-Request", "true")
|
|
if session:
|
|
req.add_header("Cookie", f"{gardien.cookie}={sid}")
|
|
try:
|
|
with urllib.request.urlopen(req) as r:
|
|
return r.status, r.read().decode()
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, e.read().decode()
|
|
appel.port, appel.gardien = port, gardien
|
|
yield appel, racine, voyage
|
|
srv.shutdown()
|
|
|
|
|
|
def test_serveur(client):
|
|
appel, racine, voyage = client
|
|
code, html = appel("/")
|
|
assert code == 200 and "Écosse" in html
|
|
code, html = appel("/voyage/analyse?v=Ecosse-2027")
|
|
assert "2 traces à rattacher" in html and "11 juin 2027" in html
|
|
# Répartir : un jour par trace dans l'ordre des noms
|
|
code, html = appel("/voyage/dates-suite", {"v": "Ecosse-2027", "debut": "2027-06-10"})
|
|
assert "2 journées" in html
|
|
assert config.lire_toml(str(voyage / "voyage.toml"))["dates_traces"] == {
|
|
"Etape 1.gpx": "2027-06-10", "Etape 2.gpx": "2027-06-11"}
|
|
# Requête de modification sans l'en-tête htmx : refusée
|
|
code, _ = appel("/voyage/date", {"v": "Ecosse-2027", "fichier": "Etape 1.gpx", "jour": ""}, htmx=False)
|
|
assert code == 403
|
|
# Voyage hors du site : refusé
|
|
code, html = appel("/voyage/analyse?v=../../etc")
|
|
assert "introuvable" in html.lower()
|
|
|
|
|
|
class _SansRedirection(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, *a, **k):
|
|
return None
|
|
|
|
|
|
def test_acces(client, tmp_path, monkeypatch):
|
|
appel, racine, voyage = client
|
|
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config"))
|
|
monkeypatch.setenv("APPDATA", str(tmp_path / "config"))
|
|
base = f"http://127.0.0.1:{appel.port}"
|
|
ouvreur = urllib.request.build_opener(_SansRedirection)
|
|
|
|
def brut(chemin):
|
|
try:
|
|
return ouvreur.open(base + chemin)
|
|
except urllib.error.HTTPError as e:
|
|
return e
|
|
# Sans session : renvoyé vers la connexion, y compris pour les fichiers du site
|
|
for chemin in ("/", "/site/site.toml", "/voyage?v=Ecosse-2027"):
|
|
r = brut(chemin)
|
|
assert r.status == 303 and r.headers["Location"] == "/connexion", chemin
|
|
assert appel("/", session=False)[0] == 200 # htmx : HX-Redirect, corps vide
|
|
# La clé d'accès ouvre une session (cookie HttpOnly), une fausse clé non
|
|
r = brut("/?jeton=" + appel.gardien.jeton)
|
|
assert r.status == 303 and "HttpOnly" in r.headers["Set-Cookie"] and "SameSite=Strict" in r.headers["Set-Cookie"]
|
|
assert brut("/?jeton=faux").headers["Location"] == "/connexion"
|
|
# Mot de passe : faux, juste, puis blocage après 5 échecs
|
|
acces.definir("un mot de passe solide")
|
|
code, html = appel("/connexion", {"mot_de_passe": "non"}, session=False)
|
|
assert "incorrect" in html
|
|
ok, _ = appel.gardien.essayer("10.0.0.1", "un mot de passe solide")
|
|
assert ok
|
|
for _ in range(5):
|
|
appel.gardien.essayer("10.0.0.2", "faux")
|
|
ok, message = appel.gardien.essayer("10.0.0.2", "un mot de passe solide")
|
|
assert not ok and "Trop d'essais" in message
|
|
# En-têtes de sécurité
|
|
r = urllib.request.urlopen(urllib.request.Request(base + "/connexion"))
|
|
assert r.headers["X-Frame-Options"] == "DENY" and "frame-ancestors 'none'" in r.headers["Content-Security-Policy"]
|
|
|
|
|
|
def test_corrections(client):
|
|
appel, racine, voyage = client
|
|
# Médias : une photo localisée, une sans position
|
|
with open(voyage / "export_photos.csv", "w") as f:
|
|
f.write('"P1.jpg",2027:06:11 10:00:00,56.75,2.30,\n"P2.jpg",2027:06:11 11:00:00,,,\n')
|
|
code, texte = appel("/voyage/position", {"v": "Ecosse-2027", "fichier": "P2.jpg", "lat": "56.8", "lon": "2.36"})
|
|
assert '"deplace": true' in texte
|
|
cfg = config.reglages_voyage(str(voyage), config.reglages_site(str(racine)))
|
|
assert corrections.positions(cfg) == {"P2.jpg": (56.8, 2.36)}
|
|
# Zone masquée du voyage, puis zone privée du site
|
|
appel("/voyage/masque", {"v": "Ecosse-2027", "action": "ajouter", "lat": "56.7", "lon": "2.24", "rayon": "3000"})
|
|
appel("/voyage/masque", {"v": "Ecosse-2027", "action": "ajouter", "portee": "site",
|
|
"lat": "48.8", "lon": "2.1", "rayon": "500"})
|
|
cfg = config.reglages_voyage(str(voyage), config.reglages_site(str(racine)))
|
|
assert corrections.masques(cfg) == [(56.7, 2.24, 3000.0), (48.8, 2.1, 500.0)]
|
|
# Génération : la photo placée apparaît, le début d'Etape 2 est retiré
|
|
config.modifier_toml(str(voyage / "voyage.toml"), {"dates_traces": {"Etape 1.gpx": "2027-06-10",
|
|
"Etape 2.gpx": "2027-06-11"}})
|
|
from rtg import pages, sources
|
|
cfg = config.reglages_voyage(str(voyage), config.reglages_site(str(racine)))
|
|
sources.ranger(str(voyage), cfg, sources.analyser(str(voyage), cfg))
|
|
v = pages.Voyage(str(voyage), cfg)
|
|
assert set(v.df_p["Fichier"]) == {"P1.jpg", "P2.jpg"}
|
|
v.lire_traces()
|
|
assert v.points_masques > 0
|
|
for pistes in v.traces.values():
|
|
for piste, pts in pistes:
|
|
assert all(corrections.distance_m(la, lo, 56.7, 2.24) > 3000 for la, lo in pts)
|
|
|
|
|
|
def test_derriere_le_serveur_web(voyage_sans_heures, tmp_path, monkeypatch):
|
|
"""--proxy --prefixe /admin : adresses préfixées, mot de passe, cookie Secure."""
|
|
racine, voyage = voyage_sans_heures
|
|
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config"))
|
|
monkeypatch.setenv("APPDATA", str(tmp_path / "config"))
|
|
acces.definir("un mot de passe solide")
|
|
app = serveur.Application(str(racine), "/admin")
|
|
srv = ThreadingHTTPServer(("127.0.0.1", 0), None)
|
|
port = srv.server_address[1]
|
|
gardien = Gardien(port, proxy=True, prefixe=app.prefixe)
|
|
srv.RequestHandlerClass = serveur.creer_gestionnaire(app, port, gardien)
|
|
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
|
ouvreur = urllib.request.build_opener(_SansRedirection)
|
|
entetes = {"Host": "roadtrip.example", "X-Forwarded-For": "203.0.113.7", "X-Forwarded-Proto": "https"}
|
|
|
|
def requete(chemin, donnees=None, extra=None):
|
|
req = urllib.request.Request(f"http://127.0.0.1:{port}{chemin}", headers=dict(entetes, **(extra or {})),
|
|
data=urllib.parse.urlencode(donnees).encode() if donnees else None)
|
|
try:
|
|
return ouvreur.open(req)
|
|
except urllib.error.HTTPError as e:
|
|
return e
|
|
try:
|
|
r = requete("/admin/")
|
|
assert r.status == 303 and r.headers["Location"] == "/admin/connexion"
|
|
assert requete("/").status == 404 # hors du préfixe
|
|
assert requete("/admin/connexion").status == 200
|
|
r = requete("/admin/connexion", {"mot_de_passe": "un mot de passe solide"}, {"HX-Request": "true"})
|
|
cookie = r.headers["Set-Cookie"]
|
|
assert r.headers["HX-Redirect"] == "/admin/" and "Path=/admin/" in cookie and "Secure" in cookie
|
|
sid = cookie.split(";")[0]
|
|
r = requete("/admin/", extra={"Cookie": sid})
|
|
html = r.read().decode()
|
|
assert r.status == 200 and 'href="/admin/nouveau"' in html and 'href="/"' in html # Voir le site
|
|
# Les échecs sont comptés par visiteur (X-Forwarded-For), pas pour le serveur web
|
|
for _ in range(5):
|
|
requete("/admin/connexion", {"mot_de_passe": "faux"}, {"HX-Request": "true"})
|
|
assert gardien.attente("203.0.113.7") > 0 and gardien.attente("127.0.0.1") == 0
|
|
finally:
|
|
srv.shutdown()
|
|
serveur.ENV.globals["base"] = ""
|
|
|
|
|
|
def test_aide(client):
|
|
appel, racine, voyage = client
|
|
code, html = appel("/aide")
|
|
assert code == 200 and "Prise en main" in html and 'href="/aide/generer"' in html
|
|
code, html = appel("/aide/generer?panneau=1")
|
|
assert "<h1>Générer</h1>" in html and "CRF" in html
|
|
code, html = appel("/aide/index?panneau=1")
|
|
assert 'hx-get="/aide/sources?panneau=1"' in html # liens internes au panneau
|
|
code, html = appel("/aide/nexistepas")
|
|
assert "introuvable" in html
|
|
# Chaque lien entre pages d'aide mène à une page qui existe
|
|
import glob, re as _re
|
|
dossier = os.path.join(os.path.dirname(serveur.__file__), "aide")
|
|
sujets = {os.path.basename(f)[:-3] for f in glob.glob(os.path.join(dossier, "*.md"))}
|
|
for f in glob.glob(os.path.join(dossier, "*.md")):
|
|
for cible in _re.findall(r"\]\(([a-z]+)\.md\)", open(f, encoding="utf-8").read()):
|
|
assert cible in sujets, (f, cible)
|