""" test_cleanup_service.py — nettoyage RGPD (services/cleanup_service.py), à l'origine de l'incident du 2026-09-03 où des photos de familles ont été supprimées sans trace. Utilise exclusivement des dossiers temporaires (tmp_path, monkeypatch) : ne touche JAMAIS aux vrais dossiers de photos. """ import os import time import uuid from datetime import timedelta from core import config from core.db import get_db from core.timeutil import utcnow, utcnow_iso from services import audit_service, cleanup_service, school_service def _touch(path, age_days: float): path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(b"fake-image-content") mtime = time.time() - age_days * 86400 os.utime(path, (mtime, mtime)) def _patch_dirs(monkeypatch, tmp_path): monkeypatch.setattr(cleanup_service, "PHOTOS_DIR", tmp_path / "photos") monkeypatch.setattr(cleanup_service, "INCOMING_DIR", tmp_path / "incoming") monkeypatch.setattr(cleanup_service, "THUMBS_DIR", tmp_path / "thumbs") return tmp_path / "photos" def test_deletes_only_files_older_than_retention(tmp_path, monkeypatch): photos_dir = _patch_dirs(monkeypatch, tmp_path) old_photo = photos_dir / "OLD-1234" / "portrait.jpg" recent_photo = photos_dir / "NEW-5678" / "portrait.jpg" non_image = photos_dir / "OLD-1234" / "meta.json" _touch(old_photo, age_days=90) _touch(recent_photo, age_days=5) _touch(non_image, age_days=90) assert cleanup_service.run_once() == 1 assert not old_photo.exists() assert recent_photo.exists() assert non_image.exists(), "meta.json n'est pas une image — ne doit jamais être supprimé" def test_retention_follows_each_schools_setting(tmp_path, monkeypatch): """Deux écoles, deux durées : 20 jours chez l'une, 120 chez l'autre.""" photos_dir = _patch_dirs(monkeypatch, tmp_path) short = school_service.create_school("Rétention courte", slug=f"short-{uuid.uuid4().hex[:5]}")["id"] long_ = school_service.create_school("Rétention longue", slug=f"long-{uuid.uuid4().hex[:5]}")["id"] school_service.update_settings(short, {"rgpd_days": 20}) school_service.update_settings(long_, {"rgpd_days": 120}) with get_db() as conn: for code, sid in (("SHT-0001", short), ("LNG-0001", long_)): conn.execute("INSERT OR REPLACE INTO families (code, family_name, created_at, school_id) VALUES (?,?,?,?)", (code, code, utcnow_iso(), sid)) conn.commit() a = photos_dir / "SHT-0001" / "a.jpg" b = photos_dir / "LNG-0001" / "b.jpg" _touch(a, age_days=30) # > 20 j → supprimée _touch(b, age_days=30) # < 120 j → conservée assert cleanup_service.run_once() == 1 assert not a.exists() assert b.exists() def test_deletion_is_logged_as_critical_audit_event_with_school(tmp_path, monkeypatch): photos_dir = _patch_dirs(monkeypatch, tmp_path) sid = school_service.create_school("Audit école", slug=f"aud-{uuid.uuid4().hex[:5]}")["id"] with get_db() as conn: conn.execute("INSERT OR REPLACE INTO families (code, family_name, created_at, school_id) VALUES ('AUD-0001','A',?,?)", (utcnow_iso(), sid)) conn.commit() _touch(photos_dir / "AUD-0001" / "fratrie.jpg", age_days=90) cleanup_service.run_once() matching = [e for e in audit_service.list_events(action="photo.rgpd_delete", limit=50) if e["resource_id"] == "AUD-0001"] assert matching, "toute suppression RGPD doit être tracée" assert matching[0]["severity"] == "critical" assert matching[0]["details"]["files_deleted"] == 1 assert matching[0]["school_id"] == sid def test_empty_family_folder_is_pruned_after_deletion(tmp_path, monkeypatch): photos_dir = _patch_dirs(monkeypatch, tmp_path) family_dir = photos_dir / "PRU-0001" _touch(family_dir / "only_photo.jpg", age_days=90) cleanup_service.run_once() assert not family_dir.exists() def test_nothing_deleted_when_all_photos_are_recent(tmp_path, monkeypatch): photos_dir = _patch_dirs(monkeypatch, tmp_path) _touch(photos_dir / "FRESH-01" / "a.jpg", age_days=1) _touch(photos_dir / "FRESH-01" / "b.webp", age_days=10) assert cleanup_service.run_once() == 0 assert (photos_dir / "FRESH-01" / "a.jpg").exists() def test_orphan_watermarked_previews_are_removed_with_the_original(tmp_path, monkeypatch): photos_dir = _patch_dirs(monkeypatch, tmp_path) _touch(photos_dir / "PRV-0001" / "x.jpg", age_days=90) preview = tmp_path / "thumbs" / "PRV-0001" / "x.wm.webp" _touch(preview, age_days=1) cleanup_service.run_once() assert not preview.exists() def test_mass_deletion_triggers_an_alert(tmp_path, monkeypatch): photos_dir = _patch_dirs(monkeypatch, tmp_path) alerts = [] monkeypatch.setattr(cleanup_service.alert_service, "notify", lambda key, subject, body="": alerts.append(key)) monkeypatch.setattr(config, "RGPD_ALERT_THRESHOLD", 3) for i in range(5): _touch(photos_dir / "MAS-0001" / f"{i}.jpg", age_days=90) cleanup_service.run_once() assert "rgpd-mass-delete" in alerts def test_abandoned_unpaid_orders_expire_and_paid_ones_survive(tmp_path, monkeypatch): _patch_dirs(monkeypatch, tmp_path) from conftest import insert_order, item old = (utcnow() - timedelta(hours=config.UNPAID_ORDER_TTL_HOURS + 2)).isoformat() stale, stale_failed, fresh, paid = (str(uuid.uuid4()) for _ in range(4)) insert_order(stale, "TST-1001", "ecole-test", [item()], payment_status="unpaid", created_at=old) insert_order(stale_failed, "TST-1001", "ecole-test", [item()], payment_status="failed", created_at=old) insert_order(fresh, "TST-1001", "ecole-test", [item()], payment_status="unpaid") insert_order(paid, "TST-1001", "ecole-test", [item()], payment_status="paid", created_at=old) assert cleanup_service.expire_unpaid_orders() >= 2 with get_db() as conn: get = lambda i: conn.execute("SELECT status, payment_status FROM orders WHERE id=?", (i,)).fetchone() assert (get(stale)["status"], get(stale)["payment_status"]) == ("cancelled", "expired") assert get(stale_failed)["status"] == "cancelled" assert get(fresh)["status"] == "pending" assert (get(paid)["status"], get(paid)["payment_status"]) == ("pending", "paid")