snoopy
9ede2c5e68
Fail-fast sur la configuration. Une limite de debit illisible (10/hours, 10/banana, -1/hour) arrete desormais le demarrage avec un message nommant la variable, au lieu de desactiver silencieusement la protection : seules les valeurs 0, off, none, disabled et vide desactivent volontairement. Meme traitement pour MAX_SHARED_MAPS, MAX_SHARE_SIZE_MB, DEFAULT_SHARE_EXPIRY_DAYS et chaque entree de TRUSTED_PROXY_CIDRS. X-Forwarded-For n'est plus cru sur la seule foi de TRUST_PROXY_HEADERS : la connexion doit aussi provenir d'un reseau declare dans TRUSTED_PROXY_CIDRS (ipaddress, IPv4/IPv6/CIDR). Activer la confiance sans declarer de reseau est refuse au demarrage, faute de quoi la configuration semblerait active sans l'etre. L'adresse retenue est normalisee : les differentes ecritures d'une meme IPv6 ne donnent plus des compteurs distincts. MAX_SHARED_MAPS : comptage et insertion serialises par un verrou de processus, pour que deux creations simultanees ne franchissent pas ensemble la derniere place. Documente comme valable pour un service applicatif unique. Ajoute APP_BIND pour ne pas publier 8080 sur toutes les interfaces en production, une option backup --keep pour la retention, et documente Nginx, Traefik en override, cron et systemd. Corrige un bug de rendu du build local : mon edition avait reecrit docker-entrypoint.sh en CRLF, rendant le shebang introuvable dans le conteneur. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
231 lines
8.0 KiB
Python
231 lines
8.0 KiB
Python
"""CLI d'administration : compteurs, listage, suppression, purge, sauvegarde."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from datetime import timedelta
|
|
from pathlib import Path
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
from app import cli
|
|
from app.db.session import SessionLocal
|
|
from app.models.share import SharedMap, utcnow
|
|
from tests.conftest import build_map
|
|
|
|
|
|
def _create(client: TestClient, title: str = "Carte") -> dict:
|
|
response = client.post("/api/shares", json={"title": title, "map": build_map()})
|
|
assert response.status_code == 201, response.text
|
|
return response.json()
|
|
|
|
|
|
def _expire(share_token: str) -> None:
|
|
with SessionLocal() as session:
|
|
share = session.execute(
|
|
select(SharedMap).where(SharedMap.share_token == share_token)
|
|
).scalar_one()
|
|
share.expires_at = utcnow() - timedelta(days=1)
|
|
session.commit()
|
|
|
|
|
|
def _count() -> int:
|
|
with SessionLocal() as session:
|
|
return len(session.execute(select(SharedMap)).scalars().all())
|
|
|
|
|
|
def test_stats_reports_counts_without_secrets(client: TestClient, capsys) -> None:
|
|
created = _create(client, "Infrastructure")
|
|
_expire(_create(client, "Ancienne")["share_token"])
|
|
|
|
assert cli.main(["stats"]) == 0
|
|
out = capsys.readouterr().out
|
|
|
|
assert "Partages : 2" in out
|
|
assert "actifs : 1" in out
|
|
assert "expires : 1" in out
|
|
assert "Taille" in out
|
|
# Ni jeton, ni hash, ni contenu de carte.
|
|
assert created["management_token"] not in out
|
|
assert created["share_token"] not in out
|
|
assert "map_data" not in out
|
|
|
|
|
|
def test_shares_list_masks_tokens(client: TestClient, capsys) -> None:
|
|
created = _create(client, "Infrastructure")
|
|
|
|
assert cli.main(["shares", "list"]) == 0
|
|
out = capsys.readouterr().out
|
|
|
|
assert "Infrastructure" in out
|
|
assert created["share_token"] not in out
|
|
assert created["management_token"] not in out
|
|
# Le jeton est identifiable sans etre utilisable.
|
|
assert created["share_token"][:4] in out
|
|
|
|
|
|
def test_shares_list_filters(client: TestClient, capsys) -> None:
|
|
active = _create(client, "Active")
|
|
expired = _create(client, "Expiree")
|
|
_expire(expired["share_token"])
|
|
|
|
cli.main(["shares", "list", "--active"])
|
|
out = capsys.readouterr().out
|
|
assert "Active" in out and "Expiree" not in out
|
|
|
|
cli.main(["shares", "list", "--expired"])
|
|
out = capsys.readouterr().out
|
|
assert "Expiree" in out and "Active" not in out
|
|
assert active["share_token"] not in out
|
|
|
|
|
|
def test_shares_list_limit(client: TestClient, capsys) -> None:
|
|
for index in range(5):
|
|
_create(client, f"Carte {index}")
|
|
|
|
cli.main(["shares", "list", "--limit", "2"])
|
|
body = capsys.readouterr().out.strip().splitlines()
|
|
# En-tete, separateur, puis deux lignes.
|
|
assert len(body) == 4
|
|
|
|
|
|
def test_shares_delete_requires_confirmation_flag(client: TestClient, capsys) -> None:
|
|
created = _create(client)
|
|
|
|
# Entree non interactive et pas de --yes : rien n'est supprime.
|
|
assert cli.main(["shares", "delete", created["share_token"]]) == 1
|
|
assert _count() == 1
|
|
|
|
assert cli.main(["shares", "delete", created["share_token"], "--yes"]) == 0
|
|
assert "1 partage supprime." in capsys.readouterr().out
|
|
assert _count() == 0
|
|
|
|
|
|
def test_shares_delete_unknown_token(client: TestClient, capsys) -> None:
|
|
assert cli.main(["shares", "delete", "inexistant", "--yes"]) == 1
|
|
assert "Aucun partage" in capsys.readouterr().out
|
|
|
|
|
|
def test_purge_expired_dry_run_changes_nothing(client: TestClient, capsys) -> None:
|
|
_create(client, "Active")
|
|
_expire(_create(client, "Expiree")["share_token"])
|
|
|
|
assert cli.main(["shares", "purge-expired", "--dry-run"]) == 0
|
|
out = capsys.readouterr().out
|
|
assert "1 partage(s) expire(s)" in out
|
|
assert "simulation" in out
|
|
assert _count() == 2
|
|
|
|
|
|
def test_purge_expired_removes_only_expired(client: TestClient, capsys) -> None:
|
|
_create(client, "Active")
|
|
_expire(_create(client, "Expiree")["share_token"])
|
|
|
|
assert cli.main(["shares", "purge-expired", "--yes"]) == 0
|
|
assert "1 partage(s) supprime(s)." in capsys.readouterr().out
|
|
|
|
with SessionLocal() as session:
|
|
remaining = session.execute(select(SharedMap.title)).scalars().all()
|
|
assert remaining == ["Active"]
|
|
|
|
|
|
def test_purge_without_expired_shares(client: TestClient, capsys) -> None:
|
|
_create(client)
|
|
assert cli.main(["shares", "purge-expired", "--yes"]) == 0
|
|
assert "0 partage(s) supprime(s)" in capsys.readouterr().out
|
|
|
|
|
|
def test_backup_produces_a_usable_database(client: TestClient, tmp_path: Path, capsys) -> None:
|
|
_create(client, "Infrastructure")
|
|
destination = tmp_path / "backups" / "openmind.db"
|
|
|
|
assert cli.main(["backup", str(destination)]) == 0
|
|
assert "Sauvegarde ecrite" in capsys.readouterr().out
|
|
assert destination.exists()
|
|
|
|
# La copie contient bien les donnees et se relit sans l'application.
|
|
connection = sqlite3.connect(destination)
|
|
try:
|
|
titles = [row[0] for row in connection.execute("SELECT title FROM shared_maps")]
|
|
finally:
|
|
connection.close()
|
|
assert titles == ["Infrastructure"]
|
|
|
|
|
|
def test_backup_refuses_to_overwrite_without_flag(client: TestClient, tmp_path: Path) -> None:
|
|
_create(client)
|
|
destination = tmp_path / "openmind.db"
|
|
assert cli.main(["backup", str(destination)]) == 0
|
|
assert cli.main(["backup", str(destination)]) == 1
|
|
assert cli.main(["backup", str(destination), "--yes"]) == 0
|
|
|
|
|
|
def test_restore_refuses_a_foreign_file(client: TestClient, tmp_path: Path, capsys) -> None:
|
|
intruder = tmp_path / "autre.db"
|
|
sqlite3.connect(intruder).execute("CREATE TABLE autre (id INTEGER)")
|
|
|
|
assert cli.main(["restore", str(intruder), "--yes"]) == 1
|
|
assert "shared_maps" in capsys.readouterr().out
|
|
# La base courante est intacte.
|
|
assert _count() == 0
|
|
|
|
|
|
def test_restore_puts_back_a_backup(client: TestClient, tmp_path: Path, capsys) -> None:
|
|
_create(client, "Avant sauvegarde")
|
|
backup = tmp_path / "snapshot.db"
|
|
cli.main(["backup", str(backup)])
|
|
capsys.readouterr()
|
|
|
|
# On modifie la base apres la sauvegarde.
|
|
with SessionLocal() as session:
|
|
for share in session.execute(select(SharedMap)).scalars():
|
|
session.delete(share)
|
|
session.commit()
|
|
assert _count() == 0
|
|
|
|
assert cli.main(["restore", str(backup), "--yes"]) == 0
|
|
out = capsys.readouterr().out
|
|
assert "Sauvegarde de securite" in out
|
|
assert "Restauration terminee" in out
|
|
|
|
with SessionLocal() as session:
|
|
titles = session.execute(select(SharedMap.title)).scalars().all()
|
|
assert titles == ["Avant sauvegarde"]
|
|
|
|
|
|
def test_backup_retention_keeps_the_most_recent(client: TestClient, tmp_path: Path, capsys) -> None:
|
|
"""--keep N ne conserve que les N sauvegardes les plus recentes du dossier."""
|
|
import os
|
|
import time
|
|
|
|
_create(client, "Infrastructure")
|
|
folder = tmp_path / "backups"
|
|
|
|
for index in range(5):
|
|
destination = folder / f"openmind-{index}.db"
|
|
assert cli.main(["backup", str(destination), "--yes"]) == 0
|
|
# Horodatage distinct : le tri de retention doit etre deterministe.
|
|
os.utime(destination, (time.time() + index, time.time() + index))
|
|
capsys.readouterr()
|
|
|
|
latest = folder / "openmind-latest.db"
|
|
assert cli.main(["backup", str(latest), "--keep", "3"]) == 0
|
|
out = capsys.readouterr().out
|
|
assert "Ancienne sauvegarde supprimee" in out
|
|
|
|
remaining = sorted(path.name for path in folder.glob("*.db"))
|
|
assert len(remaining) == 3
|
|
# La sauvegarde qui vient d'etre ecrite est toujours conservee, et compte
|
|
# dans les trois : on ne finit jamais avec une de plus que demande.
|
|
assert "openmind-latest.db" in remaining
|
|
|
|
|
|
def test_backup_without_keep_deletes_nothing(client: TestClient, tmp_path: Path) -> None:
|
|
_create(client)
|
|
folder = tmp_path / "backups"
|
|
for index in range(3):
|
|
assert cli.main(["backup", str(folder / f"openmind-{index}.db")]) == 0
|
|
assert cli.main(["backup", str(folder / "openmind-3.db")]) == 0
|
|
assert len(list(folder.glob("*.db"))) == 4
|