pyssh-client/tests/test_ai_service.py

"""AIRequestWorker : execution d'une requete IA dans un vrai QThread en
arriere-plan, jamais sur le thread Qt principal - avec dispatch correct selon
le type de requete et remontee des erreurs via signal plutot qu'exception."""
import pathlib
import sys

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))

import threading

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

from app.ai.provider import AIError, AIProvider
from app.ai.service import (
    KIND_ASK,
    KIND_DIAGNOSE,
    KIND_EXPLAIN,
    KIND_GENERATE,
    KIND_SOLUTION,
    KIND_TEST,
    AIRequestWorker,
)

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


class RecordingProvider(AIProvider):
    """Enregistre la methode appelee et le thread d'execution, pour verifier
    a la fois le dispatch correct et le fait que ca tourne hors du thread principal."""

    def __init__(self, fail_with: Exception | None = None):
        self.calls = []
        self.threads = []
        self.fail_with = fail_with

    def chat(self, messages, *, temperature=None):
        self.calls.append(("chat", messages))
        self.threads.append(threading.current_thread())
        if self.fail_with:
            raise self.fail_with
        return "reponse-chat"

    def explain_output(self, output):
        self.calls.append(("explain_output", output))
        self.threads.append(threading.current_thread())
        if self.fail_with:
            raise self.fail_with
        return "reponse-explain"

    def diagnose(self, output):
        self.calls.append(("diagnose", output))
        self.threads.append(threading.current_thread())
        if self.fail_with:
            raise self.fail_with
        return "reponse-diagnose"

    def propose_solution(self, output):
        self.calls.append(("propose_solution", output))
        self.threads.append(threading.current_thread())
        if self.fail_with:
            raise self.fail_with
        return "reponse-solution"

    def generate_command(self, request):
        self.calls.append(("generate_command", request))
        self.threads.append(threading.current_thread())
        if self.fail_with:
            raise self.fail_with
        return "reponse-generate"

    def ask(self, question, context=""):
        self.calls.append(("ask", (question, context)))
        self.threads.append(threading.current_thread())
        if self.fail_with:
            raise self.fail_with
        return "reponse-ask"


def run_worker(worker, timeout_ms=5000):
    results = {}

    def on_ok(text):
        results["ok"] = text

    def on_error(text):
        results["error"] = text

    worker.finished_ok.connect(on_ok)
    worker.finished_error.connect(on_error)
    worker.start()
    worker.wait(timeout_ms)
    # laisser la boucle d'evenements livrer les signaux en file d'attente (cross-thread)
    QCoreApplication.processEvents()
    return results


main_thread = threading.current_thread()

print("=== 1. Dispatch correct par 'kind' (explain/diagnose/solution/generate/ask/test) ===")
cases = [
    (KIND_EXPLAIN, "explain_output", "sortie X", "reponse-explain"),
    (KIND_DIAGNOSE, "diagnose", "erreur Y", "reponse-diagnose"),
    (KIND_SOLUTION, "propose_solution", "probleme Z", "reponse-solution"),
    (KIND_GENERATE, "generate_command", "trouver les gros fichiers", "reponse-generate"),
]
for kind, expected_method, payload, expected_reply in cases:
    provider = RecordingProvider()
    worker = AIRequestWorker(provider, kind, payload)
    results = run_worker(worker)
    print(f"  kind={kind!r} -> methode appelee: {provider.calls[0][0]!r}, resultat: {results!r}")
    assert provider.calls == [(expected_method, payload)]
    assert results.get("ok") == expected_reply
    assert "error" not in results
print("  OK")

print()
print("=== 2. KIND_ASK transmet aussi le contexte ===")
provider = RecordingProvider()
worker = AIRequestWorker(provider, KIND_ASK, "pourquoi ?", context="ligne de contexte")
results = run_worker(worker)
print("  appel:", provider.calls[0], "resultat:", results)
assert provider.calls == [("ask", ("pourquoi ?", "ligne de contexte"))]
assert results.get("ok") == "reponse-ask"
print("  OK")

print()
print("=== 3. KIND_TEST appelle chat() avec un message minimal ===")
provider = RecordingProvider()
worker = AIRequestWorker(provider, KIND_TEST, "")
results = run_worker(worker)
print("  appel:", provider.calls[0], "resultat:", results)
assert provider.calls[0][0] == "chat"
assert results.get("ok") == "reponse-chat"
print("  OK")

print()
print("=== 4. AIError est capturee et remontee via finished_error, jamais levee ===")
provider = RecordingProvider(fail_with=AIError("cle API invalide"))
worker = AIRequestWorker(provider, KIND_EXPLAIN, "sortie")
results = run_worker(worker)
print("  resultat:", results)
assert results.get("error") == "cle API invalide"
assert "ok" not in results
print("  OK")

print()
print("=== 5. Exception inattendue capturee aussi (jamais de crash silencieux du thread) ===")
provider = RecordingProvider(fail_with=RuntimeError("boum"))
worker = AIRequestWorker(provider, KIND_EXPLAIN, "sortie")
results = run_worker(worker)
print("  resultat:", results)
assert "error" in results and "boum" in results["error"]
print("  OK")

print()
print("=== 6. Type de requete inconnu -> erreur claire, pas de plantage ===")
provider = RecordingProvider()
worker = AIRequestWorker(provider, "kind-inexistant", "x")
results = run_worker(worker)
print("  resultat:", results)
assert "error" in results and "inconnu" in results["error"].lower()
print("  OK")

print()
print("=== 7. L'execution se fait bien dans un thread separe du thread principal ===")
provider = RecordingProvider()
worker = AIRequestWorker(provider, KIND_EXPLAIN, "sortie")
run_worker(worker)
print("  thread principal:", main_thread, "thread d'execution:", provider.threads[0])
assert provider.threads[0] is not main_thread
assert isinstance(worker, __import__("PySide6.QtCore", fromlist=["QThread"]).QThread)
print("  OK")

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