pyssh-client/tests/test_ai_panel.py

"""AIPanel : point d'entree unique des requetes IA d'un onglet - verifie que
la verification de securite (avis + secrets) est bien invoquee avant tout
envoi, que les reponses/erreurs mettent a jour la conversation, et qu'une
reponse perimee (generation obsolete) est ignoree."""
import sys, pathlib
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))

from unittest.mock import patch

from PySide6.QtCore import QCoreApplication
from PySide6.QtWidgets import QApplication

from app.ai.provider import AIProvider, AIError
from app.ai.service import KIND_EXPLAIN, KIND_ASK
from app.dialogs.ai_prompts import MASK, SEND, CANCEL
from app.ui.ai_panel import AIPanel

app = QApplication.instance() or QApplication(sys.argv)


class FakeProvider(AIProvider):
    def __init__(self, reply="reponse IA", fail_with=None):
        self.reply = reply
        self.fail_with = fail_with
        self.chat_calls = []

    def chat(self, messages, *, temperature=None):
        self.chat_calls.append(messages)
        if self.fail_with:
            raise self.fail_with
        return self.reply


def pump(ms=3000):
    """Laisse le worker en arriere-plan se terminer et livrer son signal."""
    QCoreApplication.processEvents()
    import time
    deadline = time.time() + ms / 1000
    while time.time() < deadline:
        QCoreApplication.processEvents()
        time.sleep(0.01)


NOTICE_OK = patch("app.ui.ai_panel.ensure_ai_notice_acknowledged", return_value=True)
SEND_ASIS = patch("app.ui.ai_panel.confirm_sensitive_content", return_value=SEND)


print("=== 1. Aucun fournisseur configure : message clair, pas de requete lancee ===")
panel = AIPanel(provider_factory=lambda: None)
panel.run_request(KIND_EXPLAIN, "erreur X", label="Expliquer cette sortie")
assert "non configure" in panel.conversation.toPlainText().lower()
assert panel._worker is None
print("  OK")

print()
print("=== 2. Avis de principe refuse : rien n'est envoye ===")
provider = FakeProvider()
panel = AIPanel(provider_factory=lambda: provider)
with patch("app.ui.ai_panel.ensure_ai_notice_acknowledged", return_value=False):
    panel.run_request(KIND_EXPLAIN, "erreur X", label="Expliquer cette sortie")
assert provider.chat_calls == []
assert panel._worker is None
print("  OK")

print()
print("=== 3. Secret detecte + annulation : rien n'est envoye ===")
provider = FakeProvider()
panel = AIPanel(provider_factory=lambda: provider)
with NOTICE_OK, patch("app.ui.ai_panel.confirm_sensitive_content", return_value=CANCEL):
    panel.run_request(KIND_EXPLAIN, "password=abc123", label="Expliquer cette sortie")
assert provider.chat_calls == []
print("  OK")

print()
print("=== 4. Secret detecte + masquage : le texte envoye est redige ===")
provider = FakeProvider(reply="analyse")
panel = AIPanel(provider_factory=lambda: provider)
with NOTICE_OK, patch("app.ui.ai_panel.confirm_sensitive_content", return_value=MASK):
    panel.run_request(KIND_EXPLAIN, "password=abc123", label="Expliquer cette sortie")
    pump()
sent_content = provider.chat_calls[0][-1]["content"]
print("  contenu envoye:", sent_content)
assert "abc123" not in sent_content
assert "[MASQUE]" in sent_content
assert panel._last_response == "analyse"
print("  OK")

print()
print("=== 5. Requete reussie : reponse affichee et memorisee (Copier reponse) ===")
provider = FakeProvider(reply="Tout va bien.")
panel = AIPanel(provider_factory=lambda: provider)
with NOTICE_OK, SEND_ASIS:
    panel.run_request(KIND_EXPLAIN, "sortie normale", label="Expliquer cette sortie")
    pump()
assert "Tout va bien." in panel.conversation.toPlainText()
panel._copy_last_response()
assert QApplication.clipboard().text() == "Tout va bien."
print("  OK")

print()
print("=== 6. Erreur du fournisseur : affichee dans la conversation, pas d'exception ===")
provider = FakeProvider(fail_with=AIError("cle API invalide"))
panel = AIPanel(provider_factory=lambda: provider)
with NOTICE_OK, SEND_ASIS:
    panel.run_request(KIND_EXPLAIN, "sortie", label="Expliquer cette sortie")
    pump()
assert "cle API invalide" in panel.conversation.toPlainText()
print("  OK")

print()
print("=== 7. Effacer la conversation : generation incrementee, contenu vide ===")
provider = FakeProvider(reply="reponse")
panel = AIPanel(provider_factory=lambda: provider)
with NOTICE_OK, SEND_ASIS:
    panel.run_request(KIND_EXPLAIN, "sortie", label="Expliquer cette sortie")
    pump()
gen_before = panel._generation
panel.clear_conversation()
assert panel._generation == gen_before + 1
assert panel.conversation.toPlainText() == ""
assert panel._last_response == ""
print("  OK")

print()
print("=== 8. Une reponse d'une generation perimee est ignoree (annulation cooperative) ===")
provider = FakeProvider(reply="reponse tardive")
panel = AIPanel(provider_factory=lambda: provider)
with NOTICE_OK, SEND_ASIS:
    panel.run_request(KIND_EXPLAIN, "sortie", label="Expliquer cette sortie")
    worker = panel._worker
    panel.clear_conversation()  # invalide la generation avant que le worker n'ait fini
    worker.wait(3000)
    QCoreApplication.processEvents()
assert "reponse tardive" not in panel.conversation.toPlainText()
print("  OK")

print()
print("=== 9. Question libre (panneau) : contexte joint, question affichee ===")
provider = FakeProvider(reply="42")
panel = AIPanel(provider_factory=lambda: provider)
panel.prompt_question(context="ligne de contexte")
assert panel.question_row.isVisible()
panel.question_edit.setText("pourquoi ?")
with NOTICE_OK, SEND_ASIS:
    panel._send_question()
    pump()
assert not panel.question_row.isVisible()
last_messages = provider.chat_calls[-1]
assert any("ligne de contexte" in m["content"] for m in last_messages)
assert last_messages[-1]["content"] == "pourquoi ?"
assert "42" in panel.conversation.toPlainText()
print("  OK")

print()
print("=== 10. Une requete deja en cours bloque la suivante (une a la fois) ===")
provider = FakeProvider(reply="lente")
panel = AIPanel(provider_factory=lambda: provider)
with NOTICE_OK, SEND_ASIS:
    panel.run_request(KIND_EXPLAIN, "sortie 1", label="Expliquer cette sortie")
    first_worker = panel._worker
    panel.run_request(KIND_EXPLAIN, "sortie 2", label="Expliquer cette sortie")
    assert panel._worker is first_worker, "la seconde requete n'aurait pas du demarrer un nouveau worker"
    assert "deja en cours" in panel.busy_label.text().lower()
    pump()
print("  OK")

print()
print("AI PANEL : TOUS LES TESTS OK")