rpa_vision_v3/tests/unit/test_sanitize_log_entries.py

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