merge(R2): DialogResolver MVP P0 (worktree a86565d0)

This commit is contained in:
Dom
2026-05-24 17:53:35 +02:00
7 changed files with 995 additions and 0 deletions

View File

@@ -0,0 +1,141 @@
"""Tests intégration HTTP de POST /api/v1/dialog/resolve — R2 MVP P0.
Vérifie :
- flag OFF par défaut → 503.
- flag ON + match catalogue → 200 + payload conforme.
- flag ON + pas de match → 200 + matched=False + policy=pause.
"""
from __future__ import annotations
import sys
from pathlib import Path
import pytest
_ROOT = str(Path(__file__).resolve().parents[2])
if _ROOT not in sys.path:
sys.path.insert(0, _ROOT)
pytestmark = pytest.mark.integration
_TEST_API_TOKEN = "test_dialog_resolver_endpoint_token"
@pytest.fixture
def client(monkeypatch):
"""TestClient FastAPI avec token. Le flag RPA_DIALOG_RESOLVER_ENABLED
est géré par chaque test (par défaut absent → 503)."""
monkeypatch.setenv("RPA_API_TOKEN", _TEST_API_TOKEN)
from fastapi.testclient import TestClient
from agent_v0.server_v1 import api_stream
monkeypatch.setattr(api_stream, "API_TOKEN", _TEST_API_TOKEN)
return TestClient(api_stream.app, raise_server_exceptions=False)
def _auth_headers():
return {"Authorization": f"Bearer {_TEST_API_TOKEN}"}
class TestDialogResolveEndpointFlag:
"""Flag OFF par défaut — protection anti-régression."""
def test_disabled_by_default_returns_503(self, client, monkeypatch):
# flag explicitement absent → 503 attendu.
monkeypatch.delenv("RPA_DIALOG_RESOLVER_ENABLED", raising=False)
resp = client.post(
"/api/v1/dialog/resolve",
json={"current_title": "Confirmer l'enregistrement", "evidence_texts": []},
headers=_auth_headers(),
)
assert resp.status_code == 503
assert "RPA_DIALOG_RESOLVER_ENABLED" in resp.text
class TestDialogResolveEndpointEnabled:
"""Flag ON : l'endpoint retourne une résolution conforme."""
@pytest.fixture(autouse=True)
def _enable_flag(self, monkeypatch):
monkeypatch.setenv("RPA_DIALOG_RESOLVER_ENABLED", "true")
def test_match_confirm_save_overwrite(self, client):
resp = client.post(
"/api/v1/dialog/resolve",
json={
"current_title": "Confirmer l'enregistrement",
"evidence_texts": [],
"machine_id": "pc-alpha",
},
headers=_auth_headers(),
)
assert resp.status_code == 200
body = resp.json()
assert body["matched"] is True
assert body["dialog_id"] == "confirm-save-overwrite"
assert body["policy"] == "auto"
assert body["action"] is not None
assert body["action"]["button_label"] == "Oui"
def test_match_notepad_with_evidence(self, client):
resp = client.post(
"/api/v1/dialog/resolve",
json={
"current_title": "Bloc-notes",
"evidence_texts": ["Ne pas enregistrer"],
"machine_id": "pc-alpha",
},
headers=_auth_headers(),
)
assert resp.status_code == 200
body = resp.json()
assert body["matched"] is True
assert body["dialog_id"] == "notepad-unsaved-changes"
assert body["policy"] == "auto"
def test_no_match_returns_pause(self, client):
resp = client.post(
"/api/v1/dialog/resolve",
json={
"current_title": "Fenêtre inconnue XYZ",
"evidence_texts": [],
"machine_id": "pc-alpha",
},
headers=_auth_headers(),
)
assert resp.status_code == 200
body = resp.json()
assert body["matched"] is False
assert body["dialog_id"] == ""
assert body["policy"] == "pause"
assert body["action"] is None
def test_match_windows_uac_returns_pause(self, client):
resp = client.post(
"/api/v1/dialog/resolve",
json={
"current_title": "Contrôle de compte d'utilisateur",
"evidence_texts": ["Voulez-vous autoriser cette application"],
"machine_id": "pc-alpha",
},
headers=_auth_headers(),
)
assert resp.status_code == 200
body = resp.json()
assert body["matched"] is True
assert body["dialog_id"] == "windows-uac"
assert body["policy"] == "pause"
assert body["action"] is None
def test_requires_auth(self, client):
resp = client.post(
"/api/v1/dialog/resolve",
json={"current_title": "Confirmer l'enregistrement", "evidence_texts": []},
# pas de header → 401.
)
assert resp.status_code == 401

View File

@@ -0,0 +1,227 @@
"""Tests unitaires DialogResolver — R2 MVP P0.
Couvre :
- match par titre seul (catalogue de base).
- match titre + evidence (Bloc-notes : titre ambigu).
- pas de match → policy conservative "pause".
- override déclaratif autorisé pour Easily/non-système.
- validateur sécurité : refuse `auto`/`skip` sur préfixes
`windows-` / `defender-` (modaux SYSTÈME).
"""
from __future__ import annotations
import sys
from pathlib import Path
import pytest
_ROOT = str(Path(__file__).resolve().parents[2])
if _ROOT not in sys.path:
sys.path.insert(0, _ROOT)
from agent_v0.server_v1.core.dialog import ( # noqa: E402
DeclarativeOverride,
DialogResolver,
KNOWN_DIALOGS,
SystemModalOverrideError,
system_modals_cannot_be_overridden,
)
pytestmark = pytest.mark.unit
class TestCatalog:
"""Le catalogue P0 contient au moins les 10 entrées requises."""
REQUIRED_IDS = {
"confirm-save-overwrite",
"notepad-unsaved-changes",
"windows-file-explorer",
"easily-save-unconfirmed",
"easily-overwrite-file",
"easily-confirm-action",
"easily-clinical-warning",
"windows-uac",
"windows-hello-credui",
"edge-update",
}
def test_catalog_has_p0_entries(self):
assert self.REQUIRED_IDS.issubset(KNOWN_DIALOGS.keys())
def test_system_modals_not_overridable(self):
"""UAC + Hello + file-explorer system : `declarative_override` False."""
for modal_id in ("windows-uac", "windows-hello-credui"):
assert KNOWN_DIALOGS[modal_id].declarative_override is False
assert KNOWN_DIALOGS[modal_id].policy == "pause"
def test_clinical_warning_not_overridable(self):
"""Clinical warning : décision médicale, jamais auto."""
spec = KNOWN_DIALOGS["easily-clinical-warning"]
assert spec.policy == "pause"
assert spec.declarative_override is False
class TestResolveByTitleOnly:
"""Match sur titre seul, sans evidence."""
def setup_method(self):
self.resolver = DialogResolver()
def test_confirm_save_overwrite_fr(self):
res = self.resolver.resolve("Confirmer l'enregistrement")
assert res.matched
assert res.dialog_id == "confirm-save-overwrite"
assert res.policy == "auto"
assert res.action is not None
assert res.action["button_label"] == "Oui"
def test_confirm_save_overwrite_en(self):
res = self.resolver.resolve("Confirm Save As")
assert res.matched
assert res.dialog_id == "confirm-save-overwrite"
def test_windows_uac_pause(self):
res = self.resolver.resolve(
"Contrôle de compte d'utilisateur",
evidence_texts=["Voulez-vous autoriser cette application"],
)
assert res.matched
assert res.dialog_id == "windows-uac"
assert res.policy == "pause"
assert res.action is None
def test_windows_file_explorer_fallback_pause(self):
"""Cas replay 4c38dbb8 — fallback file-explorer → pause humaine."""
res = self.resolver.resolve("rpa_vision : Explorateur de fichiers")
assert res.matched
assert res.dialog_id == "windows-file-explorer"
assert res.policy == "pause"
def test_no_match_returns_pause(self):
res = self.resolver.resolve("Fenêtre métier inconnue 42")
assert res.matched is False
assert res.dialog_id == ""
assert res.policy == "pause"
class TestResolveByTitleAndEvidence:
"""Match qui exige une evidence en plus du titre (Bloc-notes)."""
def setup_method(self):
self.resolver = DialogResolver()
def test_notepad_unsaved_changes_with_evidence(self):
res = self.resolver.resolve(
"Bloc-notes",
evidence_texts=[
"Voulez-vous enregistrer les modifications ?",
"Ne pas enregistrer",
],
)
assert res.matched
assert res.dialog_id == "notepad-unsaved-changes"
assert res.policy == "auto"
assert res.action["button_label"] == "Enregistrer"
def test_notepad_without_evidence_does_not_match(self):
"""Titre `Bloc-notes` seul est trop générique : pas de match."""
res = self.resolver.resolve("test.txt - Bloc-notes")
# Pas d'evidence → la spec evidence-required ne matche pas → pas matched.
assert res.matched is False
assert res.policy == "pause"
def test_case_insensitive_and_accent_resilient(self):
"""Normalisation : casse + apostrophes typographiques tolérées."""
res = self.resolver.resolve("CONFIRMER LENREGISTREMENT")
assert res.matched
assert res.dialog_id == "confirm-save-overwrite"
class TestSecurityValidator:
"""Validateur déclaratif : interdit auto/skip sur SYSTÈME."""
def test_uac_auto_override_raises(self):
override = DeclarativeOverride(dialog_id="windows-uac", policy="auto")
with pytest.raises(SystemModalOverrideError):
system_modals_cannot_be_overridden(override)
def test_uac_skip_override_raises(self):
override = DeclarativeOverride(dialog_id="windows-uac", policy="skip")
with pytest.raises(SystemModalOverrideError):
system_modals_cannot_be_overridden(override)
def test_defender_auto_override_raises(self):
override = DeclarativeOverride(
dialog_id="defender-smartscreen-app", policy="auto",
)
with pytest.raises(SystemModalOverrideError):
system_modals_cannot_be_overridden(override)
def test_uac_pause_override_passes(self):
"""`pause` reste valide même sur SYSTÈME (no-op)."""
override = DeclarativeOverride(dialog_id="windows-uac", policy="pause")
assert system_modals_cannot_be_overridden(override) is override
def test_easily_auto_override_passes(self):
"""Easily n'est pas SYSTÈME → surcharge auto autorisée par le validateur."""
override = DeclarativeOverride(
dialog_id="easily-save-unconfirmed",
policy="auto",
button_label="Enregistrer",
)
assert system_modals_cannot_be_overridden(override) is override
class TestDeclarativeOverrideAtResolve:
"""Le DialogResolver applique l'override quand la spec l'autorise."""
def setup_method(self):
self.resolver = DialogResolver()
def test_override_applied_on_overridable_spec(self):
"""edge-update : default `skip`, surchargeable en `auto`."""
override = DeclarativeOverride(
dialog_id="edge-update",
policy="auto",
button_label="Plus tard",
)
res = self.resolver.resolve(
"Microsoft Edge",
evidence_texts=["Microsoft Edge a été mis à jour"],
declarative_override=override,
)
assert res.matched
assert res.policy == "auto"
assert res.action is not None
assert res.action["button_label"] == "Plus tard"
def test_override_rejected_at_resolve_for_system_modal(self):
"""Defense in depth : resolve() refuse aussi auto sur SYSTÈME."""
override = DeclarativeOverride(dialog_id="windows-uac", policy="auto")
with pytest.raises(SystemModalOverrideError):
self.resolver.resolve(
"Contrôle de compte d'utilisateur",
evidence_texts=["Voulez-vous autoriser cette application"],
declarative_override=override,
)
def test_override_ignored_on_non_overridable_clinical_warning(self):
"""Clinical warning : surcharge ignorée silencieusement (declarative_override=False)."""
override = DeclarativeOverride(
dialog_id="easily-clinical-warning",
policy="auto",
button_label="OK",
)
res = self.resolver.resolve(
"Avertissement clinique",
evidence_texts=["Allergie connue"],
declarative_override=override,
)
assert res.matched
# La spec déclare declarative_override=False → policy reste pause.
assert res.policy == "pause"