""" Protección de fotos al estilo hec-photo-app (services/protection_service.py + media_service). Fija lo mismo que `test_protection_service.py` de HEC: * caché idempotente (no se gasta un crédito en cada apertura del portal); * un fallo definitivo se anota en ai_health, no cada reintento; * `protect_invert` se manda INVERTIDO a KIS (su API hace lo contrario de lo que promete el interruptor); * la huella de configuración: cambiar los ajustes SÍ cambia lo que se sirve. Y lo propio de este proyecto: el respaldo ante un fallo es la marca de agua, nunca el original. """ import io import pytest from PIL import Image from core import config from services import ai_health, photo_service, protection_service as P from conftest import DEMO_CODE, SCHOOL, family_dir SC = "ecole-test" def _jpeg_bytes(color=(10, 200, 10), size=(600, 400)) -> bytes: buf = io.BytesIO() Image.new("RGB", size, color).save(buf, "JPEG") return buf.getvalue() @pytest.fixture def kis(monkeypatch): """KIS simulado: devuelve un JPEG verde liso (las «caras difuminadas») y cuenta las llamadas.""" calls = {"n": 0, "forms": []} async def fake(photo_bytes, filename, api_token, cfg): calls["n"] += 1 calls["forms"].append(P.kis_form_data(cfg)) return _jpeg_bytes() monkeypatch.setattr(P, "_call_kis", fake) monkeypatch.setattr(config, "STORYFACE_API_KEY", "sf_test") ai_health._state.clear() yield calls # deja el colegio de test como estaba P.set_protection(SC, {}) for f in family_dir(SC, DEMO_CODE).glob(".protect/*"): f.unlink() import shutil shutil.rmtree(config.THUMBS_DIR / SC, ignore_errors=True) def _enable(**heads): P.set_protection(SC, {"protection_heads_enabled": True, "protection_heads": heads}) def _photo(): return family_dir(SC, DEMO_CODE) / "portrait.jpg" # ─── Configuración ──────────────────────────────────────────────────────────── def test_defaults_match_hec(): cfg = P.normalize_config({}) assert cfg["protection_heads_enabled"] is False assert cfg["protection_heads"] == {"intensity": 1.0, "max_heads": 0, "invert": False} assert cfg["protection_overlay"] == "none" assert cfg["protection_lens"] == {"numBalls": 4, "radius": 90, "zoom": 2.5, "speed": 2.0} assert cfg["protection_puzzle"]["shape"] == "mix" and cfg["protection_puzzle"]["num_pieces"] == 6 def test_partial_config_is_merged_with_defaults(): cfg = P.normalize_config({"protection_overlay": "lens", "protection_lens": {"numBalls": 7}}) assert cfg["protection_lens"]["numBalls"] == 7 and cfg["protection_lens"]["radius"] == 90 @pytest.mark.parametrize("raw", [ {"protection_overlay": "hologram"}, {"protection_heads": {"intensity": 5}}, {"protection_heads": {"max_heads": -1}}, {"protection_lens": {"numBalls": 99}}, {"protection_lens": {"zoom": "mucho"}}, {"protection_puzzle": {"shape": "triangulo"}}, {"protection_puzzle": {"num_pieces": 3}}, {"protection_puzzle": {"gap": 50}}, ]) def test_invalid_config_is_rejected(raw): with pytest.raises(ValueError): P.normalize_config(raw) def test_unknown_keys_are_ignored_not_stored(): cfg = P.normalize_config({"protection_heads": {"evil": 1}, "__proto__": {"x": 1}}) assert "evil" not in cfg["protection_heads"] and "__proto__" not in cfg def test_config_is_stored_per_school(): P.set_protection(SC, {"protection_overlay": "puzzle"}) P.set_protection("autre-ecole", {"protection_overlay": "lens"}) assert P.get_protection(SC)["protection_overlay"] == "puzzle" assert P.get_protection("autre-ecole")["protection_overlay"] == "lens" P.set_protection(SC, {}) P.set_protection("autre-ecole", {}) def test_corrupt_stored_config_falls_back_to_defaults_instead_of_breaking_the_portal(): from core.db import get_db with get_db() as conn: conn.execute("INSERT OR REPLACE INTO school_settings (school_id, key, value) VALUES (?,?,?)", (SC, "portal_protection", "{no es json")) conn.commit() assert P.get_protection(SC)["protection_overlay"] == "none" P.set_protection(SC, {}) def test_fingerprint_is_stable_across_int_and_float_and_changes_with_the_settings(): assert P.config_fingerprint({"intensity": 1, "max_heads": 0, "invert": False}) == \ P.config_fingerprint({"intensity": 1.0, "max_heads": 0, "invert": False}) base = P.config_fingerprint({"intensity": 1.0, "max_heads": 0, "invert": False}) assert base != P.config_fingerprint({"intensity": 1.5, "max_heads": 0, "invert": False}) assert base != P.config_fingerprint({"intensity": 1.0, "max_heads": 3, "invert": False}) assert base != P.config_fingerprint({"intensity": 1.0, "max_heads": 0, "invert": True}) def test_protect_invert_is_sent_inverted_to_kis(): """KIS: invert=false difumina el FONDO, true difumina la CARA. Nuestro «sin marcar» protege las caras.""" assert P.kis_form_data({"invert": False})["protect_invert"] == "true" assert P.kis_form_data({"invert": True})["protect_invert"] == "false" form = P.kis_form_data({"intensity": 1.5, "max_heads": 4}) assert form["protect_effect"] == "blur" and form["protect_intensity"] == "1.5" and form["protect_max_heads"] == "4" # ─── Caché y reintentos ─────────────────────────────────────────────────────── async def test_an_already_protected_photo_does_not_call_the_provider_again(kis): cfg = P.DEFAULT_HEADS_CONFIG first = await P.ensure_heads(_photo(), "tok", cfg, SC) second = await P.ensure_heads(_photo(), "tok", cfg, SC) assert first == second and first.exists() and kis["n"] == 1 assert first.parent.name == ".protect" and first.parent.parent == _photo().parent async def test_changing_the_settings_regenerates_and_forgets_the_stale_variant(kis): a = await P.ensure_heads(_photo(), "tok", {"intensity": 1.0, "max_heads": 0, "invert": False}, SC) b = await P.ensure_heads(_photo(), "tok", {"intensity": 2.0, "max_heads": 0, "invert": False}, SC) assert a != b and kis["n"] == 2 assert b.exists() and not a.exists(), "la variante de la configuración anterior debe limpiarse" async def test_a_definitive_failure_is_recorded_once_and_a_success_clears_it(kis, monkeypatch): async def boom(*a, **k): raise Exception("KIS apply_effect failed (HTTP 400): insufficient credits") monkeypatch.setattr(P, "_call_kis", boom) with pytest.raises(Exception): await P.ensure_heads(_photo(), "tok", P.DEFAULT_HEADS_CONFIG, SC) st = ai_health.get_status(SC) assert st["code"] == "no_credits" and st["task"] == "heads" and st["count"] == 1 monkeypatch.undo() monkeypatch.setattr(config, "STORYFACE_API_KEY", "sf_test") async def ok(*a, **k): return _jpeg_bytes() monkeypatch.setattr(P, "_call_kis", ok) await P.ensure_heads(_photo(), "tok", P.DEFAULT_HEADS_CONFIG, SC) assert ai_health.get_status(SC) is None async def test_nothing_is_called_when_protection_is_off_or_there_is_no_api_key(kis, monkeypatch): assert await P.protected_preview(_photo(), SC, 1000) is None # apagada _enable() monkeypatch.setattr(config, "STORYFACE_API_KEY", "") assert await P.protected_preview(_photo(), SC, 1000) is None # sin clave assert kis["n"] == 0 # ─── Lo que de verdad se sirve ──────────────────────────────────────────────── async def _preview(client, size="full"): photos = (await client.get(f"/api/portal/photos?code={DEMO_CODE}")).json()["photos"] p = next(p for p in photos if p["filename"] == "portrait.jpg") url = p["url"] if size == "full" else p["thumb_url"] return await client.get(url) def _mean_color(content: bytes): img = Image.open(io.BytesIO(content)).convert("RGB").resize((32, 32)) px = list(img.getdata()) return tuple(sum(c[i] for c in px) / len(px) for i in range(3)) async def test_with_protection_on_the_family_receives_the_protected_file(client, kis): _enable() r = await _preview(client) assert r.status_code == 200 and r.headers["content-type"] == "image/webp" red, green, blue = _mean_color(r.content) assert green > 150 and red < 60, "se sirve el archivo de KIS, no el original ni la marca de agua" assert kis["n"] == 1 await _preview(client) assert kis["n"] == 1, "la segunda visita sale de la caché" async def test_protection_is_lazy_it_only_costs_credits_for_photos_actually_requested(client, kis): _enable() await client.get(f"/api/portal/photos?code={DEMO_CODE}") # listar NO protege nada assert kis["n"] == 0 async def test_thumb_and_full_are_different_sizes_of_the_same_protected_photo(client, kis): _enable() full = Image.open(io.BytesIO((await _preview(client, "full")).content)) thumb = Image.open(io.BytesIO((await _preview(client, "thumb")).content)) assert max(full.size) <= 1000 and max(thumb.size) <= 480 and max(thumb.size) < max(full.size) assert kis["n"] == 1 async def test_when_kis_fails_the_fallback_is_the_watermarked_preview_never_the_original(client, kis, monkeypatch): _enable() async def boom(*a, **k): raise Exception("KIS apply_effect failed (HTTP 500)") monkeypatch.setattr(P, "_call_kis", boom) monkeypatch.setattr("utils.retry.RETRY_ATTEMPTS", 1) r = await _preview(client) assert r.status_code == 200 img = Image.open(io.BytesIO(r.content)) assert max(img.size) <= 1000 # reducida: no es el original (1200 px) assert r.content != (_photo()).read_bytes() assert ai_health.get_status(SC)["task"] == "heads" async def test_with_protection_off_the_watermarked_preview_is_served_and_no_credit_is_spent(client, kis): r = await _preview(client) assert r.status_code == 200 and kis["n"] == 0 assert len(set(Image.open(io.BytesIO(r.content)).convert("RGB").getdata())) > 20 # marca de agua presente async def test_changing_the_heads_settings_changes_what_is_served(client, kis): _enable(intensity=1.0) await _preview(client) _enable(intensity=2.0) await _preview(client) assert kis["n"] == 2 and {f["protect_intensity"] for f in kis["forms"]} == {"1.0", "2.0"} async def test_protected_files_never_leak_through_a_direct_url(client, kis): _enable() await _preview(client) for path in (f"/photos/{SC}/cam1/{DEMO_CODE}/.protect/portrait_heads_x.jpg", f"/thumbs/{SC}/x.webp"): assert (await client.get(path)).status_code == 404 async def test_a_school_cannot_be_served_another_schools_protected_photo(client, kis): _enable() r = await _preview(client) url = r.request.url.path.replace(DEMO_CODE, "OTH-2002") assert (await client.get(f"{url}?exp=1&sig=x")).status_code == 403 # ─── Configuración por API ──────────────────────────────────────────────────── async def test_protection_api_roundtrip_validation_and_permissions(client, admin_headers, director_headers): body = {"protection": {"protection_heads_enabled": True, "protection_overlay": "lens", "protection_heads": {"intensity": 1.5, "max_heads": 3, "invert": True}, "protection_lens": {"numBalls": 5}}} r = await client.put("/api/admin/protection", json=body, headers=admin_headers) assert r.status_code == 200 and r.json()["protection"]["protection_heads"]["max_heads"] == 3 got = (await client.get("/api/admin/protection", headers=admin_headers)).json() assert got["protection"]["protection_lens"]["numBalls"] == 5 bad = await client.put("/api/admin/protection", json={"protection": {"protection_overlay": "x"}}, headers=admin_headers) assert bad.status_code == 400 assert (await client.put("/api/admin/protection", json=body, headers=director_headers)).status_code == 403 assert (await client.get("/api/admin/protection", headers=director_headers)).status_code == 200 # el director la ve await client.put("/api/admin/protection", json={"protection": {}}, headers=admin_headers) async def test_portal_receives_the_overlay_config_but_not_the_internals(client, admin_headers): await client.put("/api/admin/protection", json={"protection": {"protection_overlay": "puzzle", "protection_puzzle": {"num_pieces": 8}}}, headers=admin_headers) prot = (await client.get(f"/api/portal/photos?code={DEMO_CODE}")).json()["protection"] assert prot["overlay"] == "puzzle" and prot["puzzle"]["num_pieces"] == 8 assert set(prot) == {"overlay", "lens", "puzzle", "heads"} await client.put("/api/admin/protection", json={"protection": {}}, headers=admin_headers) async def test_protection_change_is_audited(client, admin_headers): await client.put("/api/admin/protection", json={"protection": {"protection_heads_enabled": True}}, headers=admin_headers) ev = (await client.get("/api/admin/audit?action=protection.update", headers=admin_headers)).json()["events"] assert ev and ev[0]["details"]["heads"] is True await client.put("/api/admin/protection", json={"protection": {}}, headers=admin_headers) # ─── Limpieza ───────────────────────────────────────────────────────────────── async def test_cleanup_drops_protected_and_preview_copies_whose_original_is_gone(client, kis): from services import cleanup_service _enable() await _preview(client) folder = family_dir(SC, DEMO_CODE) protected = list((folder / ".protect").glob("portrait_heads_*")) previews = list(photo_service.mirror_path(_photo(), "").parent.glob("portrait.*")) assert protected and previews # una foto temporal para no tocar las de los demás tests tmp = folder / "temporal.jpg" tmp.write_bytes(_jpeg_bytes()) tmp_prot = folder / ".protect" / "temporal_heads_aaaaaaaaaa.jpg" tmp_prot.write_bytes(b"x") tmp_prev = photo_service.mirror_path(tmp, ".wm.1000.webp") tmp_prev.parent.mkdir(parents=True, exist_ok=True) tmp_prev.write_bytes(b"x") tmp.unlink() cleanup_service.prune_orphan_derivatives() assert not tmp_prot.exists() and not tmp_prev.exists() assert all(p.exists() for p in protected), "las copias de fotos que SÍ existen no se tocan" def test_deleting_a_photo_removes_its_protected_and_preview_copies(tmp_path, monkeypatch): monkeypatch.setattr(config, "PHOTOS_DIR", tmp_path / "photos") monkeypatch.setattr(config, "THUMBS_DIR", tmp_path / "thumbs") folder = tmp_path / "photos" / "sc" / "cam" / "MAR-AAA-111" folder.mkdir(parents=True) src = folder / "a.jpg" src.write_bytes(_jpeg_bytes()) (folder / ".protect").mkdir() prot = folder / ".protect" / "a_heads_1234567890.jpg" prot.write_bytes(b"x") prev = photo_service.mirror_path(src, ".wm.1000.webp") prev.parent.mkdir(parents=True) prev.write_bytes(b"x") assert photo_service.delete_photo("sc", "MAR-AAA-111", "a.jpg") is True assert not prot.exists() and not prev.exists() and not src.exists()