openmind/backend/tests/test_cli.py

"""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"]