"""Tests TDD de sanitize_log_entries — assainissement PII des logs Léa reçus côté serveur. Branche feat/push-log-dgx. N'importe QUE pii_sanitizer (pas api_stream, DETTE-013). """ from __future__ import annotations import sys from pathlib import Path _ROOT = str(Path(__file__).resolve().parents[2]) if _ROOT not in sys.path: sys.path.insert(0, _ROOT) # --------------------------------------------------------------------------- # 1. message avec PII → brut absent, tokens présents # --------------------------------------------------------------------------- def test_message_pii_tokenise(): """Un nom clinique + numéro long disparaissent ; des tokens [...] les remplacent. Couche 1 (regex, sans NER) : détecte le format « Prénom NOM » (RE_PRENOM_NOM) et l'IPP structuré (RE_IPP). Le format inverse « NOM Prénom » relève de la couche 2 NER — hors scope ici. """ from agent_v0.server_v1.pii_sanitizer import sanitize_log_entries entries = [ { "ts": "2026-06-30T10:00:00Z", "level": "INFO", "logger": "lea.replay", "message": "Ouverture dossier Catherine MOREL IPP: 295841", } ] result = sanitize_log_entries(entries) assert len(result) == 1 msg = result[0]["message"] assert "MOREL" not in msg, f"NOM toujours présent : {msg!r}" assert "Catherine" not in msg, f"Prénom toujours présent : {msg!r}" assert "295841" not in msg, f"IPP toujours présent : {msg!r}" assert "[" in msg, f"Aucun token dans : {msg!r}" # --------------------------------------------------------------------------- # 2. ts / level préservés à l'identique # --------------------------------------------------------------------------- def test_ts_level_preserves(): from agent_v0.server_v1.pii_sanitizer import sanitize_log_entries entries = [ {"ts": "2026-06-30T10:00:00Z", "level": "WARNING", "logger": "lea.core", "message": "simple message sans pii"} ] result = sanitize_log_entries(entries) assert result[0]["ts"] == "2026-06-30T10:00:00Z" assert result[0]["level"] == "WARNING" # --------------------------------------------------------------------------- # 3. liste vide → liste vide # --------------------------------------------------------------------------- def test_liste_vide(): from agent_v0.server_v1.pii_sanitizer import sanitize_log_entries assert sanitize_log_entries([]) == [] # --------------------------------------------------------------------------- # 4. entrée sans clé `message` → pas de crash, entrée conservée # --------------------------------------------------------------------------- def test_entree_sans_message(): from agent_v0.server_v1.pii_sanitizer import sanitize_log_entries entries = [{"ts": "2026-06-30T10:00:01Z", "level": "DEBUG", "logger": "lea.init"}] result = sanitize_log_entries(entries) assert len(result) == 1 assert "message" not in result[0] # champ absent → reste absent # --------------------------------------------------------------------------- # 5. cohérence : même PII dans 2 entrées → même token (mapping partagé) # --------------------------------------------------------------------------- def test_coherence_mapping_partage(): """La même PII dans deux messages du batch reçoit le même token.""" from agent_v0.server_v1.pii_sanitizer import sanitize_log_entries entries = [ {"ts": "T1", "level": "INFO", "logger": "l", "message": "IPP: 295841 reçu"}, {"ts": "T2", "level": "INFO", "logger": "l", "message": "Relance IPP: 295841"}, ] result = sanitize_log_entries(entries) msg1 = result[0]["message"] msg2 = result[1]["message"] # le brut est absent des deux assert "295841" not in msg1 assert "295841" not in msg2 # le token est identique (mapping partagé) import re tokens1 = re.findall(r"\[IPP_\d+\]", msg1) tokens2 = re.findall(r"\[IPP_\d+\]", msg2) assert tokens1, f"Pas de token IPP dans msg1 : {msg1!r}" assert tokens2, f"Pas de token IPP dans msg2 : {msg2!r}" assert tokens1[0] == tokens2[0], ( f"Tokens différents pour la même PII : {tokens1[0]} vs {tokens2[0]}" ) # --------------------------------------------------------------------------- # 6. `message` non-str → skip proprement, pas de crash # --------------------------------------------------------------------------- def test_message_non_str(): from agent_v0.server_v1.pii_sanitizer import sanitize_log_entries entries = [ {"ts": "T1", "level": "INFO", "logger": "l", "message": None}, {"ts": "T2", "level": "INFO", "logger": "l", "message": 42}, {"ts": "T3", "level": "INFO", "logger": "l", "message": ["liste"]}, ] result = sanitize_log_entries(entries) assert len(result) == 3 # les valeurs non-str sont préservées telles quelles assert result[0]["message"] is None assert result[1]["message"] == 42 assert result[2]["message"] == ["liste"] # --------------------------------------------------------------------------- # 7. champ `logger` str est aussi assaini si porteur de PII # --------------------------------------------------------------------------- def test_logger_pii_tokenise(): """Si le champ logger contient de la PII (ex. chemin patient), il est assaini.""" from agent_v0.server_v1.pii_sanitizer import sanitize_log_entries entries = [ { "ts": "T1", "level": "INFO", "logger": "lea.patient.MOREL_Catherine", "message": "step start", } ] result = sanitize_log_entries(entries) logger_out = result[0]["logger"] # Le NOM doit être tokenisé (RE_PRENOM_NOM captera « Catherine MOREL » … # mais « MOREL_Catherine » n'est pas le format clinique standard — le test # vérifie surtout qu'il n'y a pas de crash et que le champ est traité.) # On ne fixe pas d'assertion sur la valeur : juste pas de crash. assert isinstance(logger_out, str)