diff --git a/app/ai/__init__.py b/app/ai/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/app/ai/danger.py b/app/ai/danger.py
new file mode 100644
index 0000000..a19a4c4
--- /dev/null
+++ b/app/ai/danger.py
@@ -0,0 +1,37 @@
+"""Detection locale simple de commandes potentiellement destructives, proposees
+par l'assistant IA. N'a pas besoin d'etre parfaite : sert a afficher un
+avertissement, jamais a bloquer ou executer quoi que ce soit automatiquement."""
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+_DANGEROUS_PATTERNS: list[tuple[str, re.Pattern]] = [
+ ("rm -rf (suppression recursive forcee)", re.compile(r"\brm\s+(-\w*\s+)*-\w*[rf]\w*[rf]\w*\b", re.IGNORECASE)),
+ ("mkfs (formatage)", re.compile(r"\bmkfs(\.\w+)?\b", re.IGNORECASE)),
+ ("dd (ecriture bas niveau sur un disque)", re.compile(r"\bdd\s+.*\bif=", re.IGNORECASE | re.DOTALL)),
+ ("wipefs (effacement des signatures de systeme de fichiers)", re.compile(r"\bwipefs\b", re.IGNORECASE)),
+ ("shutdown (extinction)", re.compile(r"\bshutdown\b", re.IGNORECASE)),
+ ("reboot (redemarrage)", re.compile(r"\breboot\b", re.IGNORECASE)),
+ ("iptables : purge des regles", re.compile(r"\biptables\b.*(-F|--flush)\b", re.IGNORECASE)),
+ ("nft : purge des regles", re.compile(r"\bnft\b.*\bflush\b", re.IGNORECASE)),
+ ("DROP DATABASE", re.compile(r"\bDROP\s+DATABASE\b", re.IGNORECASE)),
+ ("DELETE FROM sans WHERE", re.compile(r"\bDELETE\s+FROM\s+\S+\s*;", re.IGNORECASE)),
+ ("docker system prune (suppression en masse)", re.compile(r"\bdocker\s+system\s+prune\b", re.IGNORECASE)),
+ ("zpool destroy", re.compile(r"\bzpool\s+destroy\b", re.IGNORECASE)),
+ ("lvremove (suppression de volume logique)", re.compile(r"\blvremove\b", re.IGNORECASE)),
+ ("vgremove (suppression de groupe de volumes)", re.compile(r"\bvgremove\b", re.IGNORECASE)),
+]
+
+
+@dataclass
+class DangerMatch:
+ label: str
+
+
+def check_dangerous(command: str) -> list[DangerMatch]:
+ return [DangerMatch(label) for label, pattern in _DANGEROUS_PATTERNS if pattern.search(command)]
+
+
+def is_dangerous(command: str) -> bool:
+ return bool(check_dangerous(command))
diff --git a/app/ai/provider.py b/app/ai/provider.py
new file mode 100644
index 0000000..4d55c06
--- /dev/null
+++ b/app/ai/provider.py
@@ -0,0 +1,154 @@
+"""Abstraction du fournisseur IA : une API compatible OpenAI (endpoint chat
+completions), avec URL de base et modele configurables - pas de liste de
+fournisseurs ou de modeles figee en dur, pour supporter OpenAI, OpenRouter, un
+proxy interne ou un LLM local exposant la meme API.
+
+AIProvider porte les 4 usages metier (explain_output/diagnose/generate_command/
+ask), tous construits au-dessus de chat(), seule methode que doit implementer
+un fournisseur concret (OpenAICompatibleProvider)."""
+from __future__ import annotations
+
+import abc
+import json
+import urllib.error
+import urllib.request
+
+
+class AIError(Exception):
+ """Message deja comprehensible, pret a etre affiche a l'utilisateur."""
+
+
+_SYSTEM_PROMPT_TROUBLESHOOT = """Tu es un assistant specialise en administration systeme, Linux, Windows,
+reseau, SSH, Docker, virtualisation et troubleshooting.
+
+Analyse la sortie terminal fournie par l'utilisateur.
+
+Explique :
+1. ce que signifie la sortie ;
+2. la cause probable ;
+3. les verifications recommandees ;
+4. les commandes eventuellement utiles.
+
+Ne suppose jamais qu'une commande a ete executee.
+Ne pretends jamais connaitre des elements qui ne sont pas presents.
+Signale explicitement les hypotheses."""
+
+_SYSTEM_PROMPT_DIAGNOSE = _SYSTEM_PROMPT_TROUBLESHOOT + """
+
+Structure ta reponse ainsi :
+Cause probable
+Explication
+Verifications
+Commandes proposees
+Niveau de risque"""
+
+_SYSTEM_PROMPT_COMMAND = """Tu es un assistant specialise en administration systeme, Linux, Windows,
+reseau, SSH, Docker et virtualisation.
+
+L'utilisateur decrit ce qu'il veut accomplir en langage naturel. Propose UNE
+commande (ou une courte suite de commandes) qui repond a la demande, dans un
+bloc de code, avec une breve explication de ce qu'elle fait.
+
+Ne suppose jamais le systeme d'exploitation ou l'environnement si l'utilisateur
+ne l'a pas precise : demande-le, ou propose une variante par systeme si
+pertinent. La commande proposee ne sera jamais executee automatiquement : elle
+sera seulement affichee a l'utilisateur pour verification."""
+
+
+class AIProvider(abc.ABC):
+ @abc.abstractmethod
+ def chat(self, messages: list[dict], *, temperature: float | None = None) -> str:
+ """Renvoie le texte de la reponse. Leve AIError en cas d'echec."""
+
+ def explain_output(self, output: str) -> str:
+ return self.chat([
+ {"role": "system", "content": _SYSTEM_PROMPT_TROUBLESHOOT},
+ {"role": "user", "content": f"Explique cette sortie terminal :\n\n{output}"},
+ ])
+
+ def diagnose(self, output: str) -> str:
+ return self.chat([
+ {"role": "system", "content": _SYSTEM_PROMPT_DIAGNOSE},
+ {"role": "user", "content": f"Diagnostique cette erreur :\n\n{output}"},
+ ])
+
+ def propose_solution(self, output: str) -> str:
+ return self.chat([
+ {"role": "system", "content": _SYSTEM_PROMPT_TROUBLESHOOT},
+ {"role": "user", "content": (
+ "Propose une solution concrete a ce probleme, avec les commandes "
+ f"eventuellement necessaires :\n\n{output}"
+ )},
+ ])
+
+ def generate_command(self, request: str) -> str:
+ return self.chat([
+ {"role": "system", "content": _SYSTEM_PROMPT_COMMAND},
+ {"role": "user", "content": request},
+ ])
+
+ def ask(self, question: str, context: str = "") -> str:
+ messages = [{"role": "system", "content": _SYSTEM_PROMPT_TROUBLESHOOT}]
+ if context:
+ messages.append({"role": "user", "content": f"Contexte (sortie terminal selectionnee) :\n{context}"})
+ messages.append({"role": "user", "content": question})
+ return self.chat(messages)
+
+
+class OpenAICompatibleProvider(AIProvider):
+ def __init__(
+ self, base_url: str, api_key: str, model: str,
+ timeout: float = 30.0, temperature: float = 0.3,
+ ):
+ self.base_url = base_url.rstrip("/")
+ self.api_key = api_key
+ self.model = model
+ self.timeout = timeout
+ self.temperature = temperature
+
+ def chat(self, messages: list[dict], *, temperature: float | None = None) -> str:
+ if not self.base_url:
+ raise AIError("Aucune URL de base configuree pour l'assistant IA.")
+ if not self.model:
+ raise AIError("Aucun modele configure pour l'assistant IA.")
+
+ url = f"{self.base_url}/chat/completions"
+ payload = {
+ "model": self.model,
+ "messages": messages,
+ "temperature": self.temperature if temperature is None else temperature,
+ }
+ headers = {"Content-Type": "application/json"}
+ if self.api_key:
+ headers["Authorization"] = f"Bearer {self.api_key}"
+ request = urllib.request.Request(
+ url, data=json.dumps(payload).encode("utf-8"), headers=headers, method="POST",
+ )
+
+ try:
+ with urllib.request.urlopen(request, timeout=self.timeout) as response:
+ raw = response.read()
+ except urllib.error.HTTPError as exc:
+ raise AIError(self._describe_http_error(exc)) from exc
+ except urllib.error.URLError as exc:
+ if isinstance(exc.reason, TimeoutError):
+ raise AIError("Le fournisseur IA n'a pas repondu a temps (delai depasse).") from exc
+ raise AIError(f"Impossible de joindre le fournisseur IA : {exc.reason}") from exc
+
+ try:
+ parsed = json.loads(raw.decode("utf-8"))
+ return parsed["choices"][0]["message"]["content"]
+ except (json.JSONDecodeError, KeyError, IndexError, TypeError, UnicodeDecodeError) as exc:
+ raise AIError(f"Reponse du fournisseur IA invalide ou inattendue : {exc}") from exc
+
+ @staticmethod
+ def _describe_http_error(exc: urllib.error.HTTPError) -> str:
+ if exc.code == 401:
+ return "Authentification refusee (401) : verifiez la cle API."
+ if exc.code == 429:
+ return "Trop de requetes (429) : le fournisseur limite le debit, reessayez plus tard."
+ try:
+ body = exc.read().decode("utf-8", errors="replace")
+ except Exception: # noqa: BLE001
+ body = ""
+ return f"Erreur du fournisseur IA ({exc.code}) : {(body or exc.reason)[:300]}"
diff --git a/app/ai/redaction.py b/app/ai/redaction.py
new file mode 100644
index 0000000..e27715b
--- /dev/null
+++ b/app/ai/redaction.py
@@ -0,0 +1,52 @@
+"""Detection/redaction basique de secrets avant envoi a un fournisseur IA
+externe : cles privees SSH, jetons Bearer, cles AWS, en-tetes Authorization,
+password=/token=/api_key=, et longues chaines ressemblant a un secret.
+
+Heuristique volontairement simple (comme la detection de commandes
+dangereuses) : mieux vaut avertir a tort sur un hash ou un identifiant anodin
+que laisser passer un vrai secret."""
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+_PATTERNS: list[tuple[str, re.Pattern]] = [
+ ("Cle privee (PEM/OpenSSH)", re.compile(
+ r"-----BEGIN (?:RSA |EC |OPENSSH |DSA |ENCRYPTED )?PRIVATE KEY-----"
+ )),
+ ("Jeton Bearer", re.compile(r"\bBearer\s+[A-Za-z0-9\-_.=]{10,}", re.IGNORECASE)),
+ ("Cle d'acces AWS", re.compile(r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b")),
+ ("En-tete Authorization", re.compile(r"\bAuthorization\s*:\s*\S+", re.IGNORECASE)),
+ ("Mot de passe (password=)", re.compile(r"\b(?:password|passwd|pwd)\s*[:=]\s*\S+", re.IGNORECASE)),
+ ("Cle API (api_key=)", re.compile(r"\b(?:api[_-]?key|apikey)\s*[:=]\s*\S+", re.IGNORECASE)),
+ ("Jeton (token=)", re.compile(r"\btoken\s*[:=]\s*\S+", re.IGNORECASE)),
+ ("Chaine longue ressemblant a un secret", re.compile(r"\b[A-Za-z0-9_\-]{32,}\b")),
+]
+
+
+@dataclass
+class DetectedSecret:
+ label: str
+ excerpt: str
+
+
+def detect_secrets(text: str) -> list[DetectedSecret]:
+ """Renvoie les categories detectees (au plus une occurrence par categorie,
+ pour un resume lisible), dans l'ordre des motifs (du plus specifique au
+ plus generique)."""
+ found: list[DetectedSecret] = []
+ for label, pattern in _PATTERNS:
+ match = pattern.search(text)
+ if match:
+ excerpt = match.group(0)
+ if len(excerpt) > 60:
+ excerpt = excerpt[:57] + "..."
+ found.append(DetectedSecret(label, excerpt))
+ return found
+
+
+def redact(text: str) -> str:
+ redacted = text
+ for _label, pattern in _PATTERNS:
+ redacted = pattern.sub("[MASQUE]", redacted)
+ return redacted
diff --git a/app/ai/service.py b/app/ai/service.py
new file mode 100644
index 0000000..75babce
--- /dev/null
+++ b/app/ai/service.py
@@ -0,0 +1,54 @@
+"""Execution des requetes IA en arriere-plan : aucun appel reseau ne doit
+jamais bloquer le thread Qt."""
+from __future__ import annotations
+
+from PySide6.QtCore import QThread, Signal
+
+from app.ai.provider import AIProvider, AIError
+
+KIND_EXPLAIN = "explain"
+KIND_DIAGNOSE = "diagnose"
+KIND_SOLUTION = "solution"
+KIND_GENERATE = "generate"
+KIND_ASK = "ask"
+KIND_TEST = "test"
+
+
+class AIRequestWorker(QThread):
+ """Une requete = un worker jetable. L'annulation est cooperative : la
+ requete HTTP en cours (bornee par son timeout) va a son terme, mais le
+ signal de resultat peut etre ignore par l'appelant (voir AIPanel)."""
+
+ finished_ok = Signal(str)
+ finished_error = Signal(str)
+
+ def __init__(self, provider: AIProvider, kind: str, payload: str, context: str = "", parent=None):
+ super().__init__(parent)
+ self.provider = provider
+ self.kind = kind
+ self.payload = payload
+ self.context = context
+
+ def run(self) -> None:
+ try:
+ result = self._execute()
+ self.finished_ok.emit(result)
+ except AIError as exc:
+ self.finished_error.emit(str(exc))
+ except Exception as exc: # noqa: BLE001 - ne jamais laisser un thread planter silencieusement
+ self.finished_error.emit(f"Erreur inattendue : {exc}")
+
+ def _execute(self) -> str:
+ if self.kind == KIND_EXPLAIN:
+ return self.provider.explain_output(self.payload)
+ if self.kind == KIND_DIAGNOSE:
+ return self.provider.diagnose(self.payload)
+ if self.kind == KIND_SOLUTION:
+ return self.provider.propose_solution(self.payload)
+ if self.kind == KIND_GENERATE:
+ return self.provider.generate_command(self.payload)
+ if self.kind == KIND_ASK:
+ return self.provider.ask(self.payload, context=self.context)
+ if self.kind == KIND_TEST:
+ return self.provider.chat([{"role": "user", "content": "Reponds uniquement 'OK'."}])
+ raise AIError(f"Type de requete IA inconnu : {self.kind}")
diff --git a/app/ai/settings.py b/app/ai/settings.py
new file mode 100644
index 0000000..12b773e
--- /dev/null
+++ b/app/ai/settings.py
@@ -0,0 +1,85 @@
+"""Configuration de l'assistant IA (optionnel, desactive par defaut).
+
+La cle API n'est jamais stockee ici : seuls les parametres non sensibles
+(activation, fournisseur, URL, modele, timeout, temperature) le sont, via
+app.app_settings. La cle passe exclusivement par CredentialStore
+(app.credentials), jamais dans settings.json/sessions.json/logs."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from app.app_settings import get_setting, set_setting
+from app.credentials import CredentialStore
+from app.ai.provider import AIProvider, OpenAICompatibleProvider
+
+AI_API_KEY_CREDENTIAL = "ai:api_key"
+
+DEFAULT_BASE_URL = "https://api.openai.com/v1"
+DEFAULT_PROVIDER_NAME = "OpenAI"
+DEFAULT_MODEL = ""
+DEFAULT_TIMEOUT = 30.0
+DEFAULT_TEMPERATURE = 0.3
+
+
+@dataclass
+class AISettings:
+ enabled: bool = False
+ provider_name: str = DEFAULT_PROVIDER_NAME
+ base_url: str = DEFAULT_BASE_URL
+ model: str = DEFAULT_MODEL
+ timeout: float = DEFAULT_TIMEOUT
+ temperature: float = DEFAULT_TEMPERATURE
+
+
+def load_ai_settings() -> AISettings:
+ try:
+ timeout = float(get_setting("ai_timeout", DEFAULT_TIMEOUT))
+ except (TypeError, ValueError):
+ timeout = DEFAULT_TIMEOUT
+ try:
+ temperature = float(get_setting("ai_temperature", DEFAULT_TEMPERATURE))
+ except (TypeError, ValueError):
+ temperature = DEFAULT_TEMPERATURE
+ return AISettings(
+ enabled=bool(get_setting("ai_enabled", False)),
+ provider_name=get_setting("ai_provider_name", DEFAULT_PROVIDER_NAME) or DEFAULT_PROVIDER_NAME,
+ base_url=get_setting("ai_base_url", DEFAULT_BASE_URL) or DEFAULT_BASE_URL,
+ model=get_setting("ai_model", DEFAULT_MODEL) or DEFAULT_MODEL,
+ timeout=timeout,
+ temperature=temperature,
+ )
+
+
+def save_ai_settings(settings: AISettings) -> None:
+ set_setting("ai_enabled", settings.enabled)
+ set_setting("ai_provider_name", settings.provider_name)
+ set_setting("ai_base_url", settings.base_url)
+ set_setting("ai_model", settings.model)
+ set_setting("ai_timeout", settings.timeout)
+ set_setting("ai_temperature", settings.temperature)
+
+
+def load_api_key(credential_store: CredentialStore) -> str:
+ return credential_store.get_secret(AI_API_KEY_CREDENTIAL) or ""
+
+
+def save_api_key(credential_store: CredentialStore, api_key: str) -> bool:
+ if api_key:
+ return credential_store.set_secret(AI_API_KEY_CREDENTIAL, api_key)
+ credential_store.delete_secret(AI_API_KEY_CREDENTIAL)
+ return True
+
+
+def build_provider_from_settings(credential_store: CredentialStore) -> AIProvider | None:
+ """None si l'assistant est desactive : point d'appel unique pour tous les
+ endroits de l'UI qui ont besoin d'un fournisseur pret a l'emploi."""
+ settings = load_ai_settings()
+ if not settings.enabled:
+ return None
+ return OpenAICompatibleProvider(
+ base_url=settings.base_url,
+ api_key=load_api_key(credential_store),
+ model=settings.model,
+ timeout=settings.timeout,
+ temperature=settings.temperature,
+ )
diff --git a/app/controllers/connection_controller.py b/app/controllers/connection_controller.py
index 3684fd7..e83399e 100644
--- a/app/controllers/connection_controller.py
+++ b/app/controllers/connection_controller.py
@@ -33,6 +33,7 @@ class ConnectionController(QObject):
broadcast: BroadcastManager,
snippet_store=None,
credential_store: CredentialStore | None = None,
+ ai_provider_factory=None,
parent_widget: QWidget | None = None,
):
super().__init__(parent_widget)
@@ -41,6 +42,7 @@ class ConnectionController(QObject):
self.broadcast = broadcast
self.snippet_store = snippet_store
self.credential_store = credential_store or NullCredentialStore()
+ self.ai_provider_factory = ai_provider_factory
self._parent_widget = parent_widget
self.tabs = QTabWidget(parent_widget)
@@ -108,6 +110,7 @@ class ConnectionController(QObject):
return SessionTabWidget(
session, secret, bastion_target=bastion_target, bastion_password=bastion_password,
snippet_store=self.snippet_store, broadcast=self.broadcast,
+ ai_provider_factory=self.ai_provider_factory, credential_store=self.credential_store,
)
def build_sftp_tab(self, session: Session) -> SFTPTabWidget | None:
@@ -149,7 +152,10 @@ class ConnectionController(QObject):
def new_local_terminal(self, shell: str = DEFAULT_SHELL) -> LocalTerminalTabWidget:
"""Cree un terminal local cable (multi-send, SFTP detecte), sans l'ajouter
au QTabWidget : reutilise pour un onglet normal comme pour un panneau."""
- tab = LocalTerminalTabWidget(shell, snippet_store=self.snippet_store, broadcast=self.broadcast)
+ tab = LocalTerminalTabWidget(
+ shell, snippet_store=self.snippet_store, broadcast=self.broadcast,
+ ai_provider_factory=self.ai_provider_factory, credential_store=self.credential_store,
+ )
tab.sftp_requested.connect(self.quick_sftp)
self.broadcast.register(tab)
return tab
diff --git a/app/dialogs/ai_command_dialog.py b/app/dialogs/ai_command_dialog.py
new file mode 100644
index 0000000..4a0fbac
--- /dev/null
+++ b/app/dialogs/ai_command_dialog.py
@@ -0,0 +1,79 @@
+"""Affiche une commande proposee par l'assistant IA : jamais executee
+automatiquement, seulement copiable ou inseree dans le terminal (sans Entree)
+apres verification par l'utilisateur."""
+from __future__ import annotations
+
+import re
+
+from PySide6.QtWidgets import (
+ QDialog, QVBoxLayout, QHBoxLayout, QTextEdit, QPushButton, QLabel, QApplication,
+)
+
+_CODE_BLOCK_RE = re.compile(r"```(?:[a-zA-Z0-9_+-]*\n)?(.*?)```", re.DOTALL)
+
+
+def extract_command(response: str) -> str:
+ """Prend le contenu du premier bloc de code s'il y en a un (le prompt
+ systeme en demande un), sinon la reponse entiere : mieux vaut proposer un
+ texte un peu trop large que perdre silencieusement la commande."""
+ match = _CODE_BLOCK_RE.search(response)
+ text = match.group(1) if match else response
+ return text.strip()
+
+
+class AICommandDialog(QDialog):
+ def __init__(self, response: str, dangerous_labels: list[str] | None = None, parent=None):
+ super().__init__(parent)
+ self.setWindowTitle("Commande generee par l'assistant IA")
+ self.setMinimumSize(520, 360)
+ self.command = extract_command(response)
+ self.insert_requested = False
+
+ layout = QVBoxLayout(self)
+
+ if dangerous_labels:
+ warning = QLabel(
+ "⚠ Commande potentiellement destructive : " + ", ".join(dangerous_labels)
+ )
+ warning.setWordWrap(True)
+ warning.setStyleSheet("color: #c0392b; font-weight: bold;")
+ layout.addWidget(warning)
+
+ layout.addWidget(QLabel("Reponse complete de l'assistant :"))
+ self.response_view = QTextEdit(response)
+ self.response_view.setReadOnly(True)
+ layout.addWidget(self.response_view, 1)
+
+ layout.addWidget(QLabel("Commande proposee (celle qui sera copiee/inseree) :"))
+ self.command_view = QTextEdit(self.command)
+ self.command_view.setFixedHeight(70)
+ layout.addWidget(self.command_view)
+
+ self.status_label = QLabel("Cette commande ne sera jamais executee automatiquement.")
+ self.status_label.setObjectName("statusLabel")
+ layout.addWidget(self.status_label)
+
+ btn_row = QHBoxLayout()
+ self.copy_btn = QPushButton("Copier")
+ self.insert_btn = QPushButton("Inserer dans le terminal")
+ self.cancel_btn = QPushButton("Annuler")
+ for b in (self.copy_btn, self.insert_btn, self.cancel_btn):
+ btn_row.addWidget(b)
+ layout.addLayout(btn_row)
+
+ self.copy_btn.clicked.connect(self._copy)
+ self.insert_btn.clicked.connect(self._insert)
+ self.cancel_btn.clicked.connect(self.reject)
+
+ def _current_command(self) -> str:
+ # l'utilisateur peut corriger la commande proposee avant de l'utiliser
+ return self.command_view.toPlainText().strip()
+
+ def _copy(self) -> None:
+ QApplication.clipboard().setText(self._current_command())
+ self.status_label.setText("Copie dans le presse-papiers.")
+
+ def _insert(self) -> None:
+ self.command = self._current_command()
+ self.insert_requested = True
+ self.accept()
diff --git a/app/dialogs/ai_prompts.py b/app/dialogs/ai_prompts.py
new file mode 100644
index 0000000..e0b5caf
--- /dev/null
+++ b/app/dialogs/ai_prompts.py
@@ -0,0 +1,70 @@
+"""Confirmations de securite avant tout envoi de contenu au fournisseur IA :
+un avertissement de principe affiche une seule fois, puis une verification
+systematique des secrets evidents a chaque envoi d'une selection de terminal
+ou d'une demande libre."""
+from __future__ import annotations
+
+from PySide6.QtWidgets import QMessageBox
+
+from app.app_settings import get_setting, set_setting
+from app.ai.redaction import detect_secrets
+
+_NOTICE_ACK_KEY = "ai_notice_acknowledged"
+
+MASK = "mask"
+SEND = "send"
+CANCEL = "cancel"
+
+
+def ensure_ai_notice_acknowledged(parent) -> bool:
+ """Affiche l'avertissement de principe une seule fois (jamais reaffiche
+ ensuite, l'accord etant persiste). Renvoie False si l'utilisateur refuse :
+ l'appelant ne doit alors rien envoyer."""
+ if get_setting(_NOTICE_ACK_KEY, False):
+ return True
+ box = QMessageBox(parent)
+ box.setWindowTitle("Assistant IA")
+ box.setIcon(QMessageBox.Information)
+ box.setText(
+ "Le texte selectionne dans le terminal (ou votre question) va etre envoye "
+ "au fournisseur IA que vous avez configure, en dehors de cet ordinateur.\n\n"
+ "PySSH tente de detecter les secrets evidents (mots de passe, cles, jetons) "
+ "avant chaque envoi et vous permet de les masquer, mais ne peut pas tout "
+ "detecter : verifiez ce que vous envoyez.\n\n"
+ "Ce message ne sera plus affiche."
+ )
+ box.setStandardButtons(QMessageBox.Ok | QMessageBox.Cancel)
+ box.setDefaultButton(QMessageBox.Ok)
+ accepted = box.exec() == QMessageBox.Ok
+ if accepted:
+ set_setting(_NOTICE_ACK_KEY, True)
+ return accepted
+
+
+def confirm_sensitive_content(parent, text: str) -> str:
+ """Renvoie MASK, SEND ou CANCEL. Si aucun secret evident n'est detecte,
+ renvoie SEND directement sans rien afficher."""
+ secrets = detect_secrets(text)
+ if not secrets:
+ return SEND
+ labels = "\n".join(f"- {s.label}" for s in secrets)
+ box = QMessageBox(parent)
+ box.setWindowTitle("Donnees potentiellement sensibles")
+ box.setIcon(QMessageBox.Warning)
+ box.setText(
+ "Des donnees potentiellement sensibles ont ete detectees dans le texte "
+ f"qui serait envoye au fournisseur IA :\n\n{labels}\n\n"
+ "« Masquer automatiquement » les remplace par [MASQUE] avant l'envoi.\n"
+ "« Voir le contenu » l'envoie tel quel, sans modification."
+ )
+ mask_btn = box.addButton("Masquer automatiquement", QMessageBox.AcceptRole)
+ send_btn = box.addButton("Voir le contenu", QMessageBox.DestructiveRole)
+ box.addButton("Annuler", QMessageBox.RejectRole)
+ box.setDefaultButton(mask_btn)
+ box.exec()
+ clicked = box.clickedButton()
+ if clicked is mask_btn:
+ return MASK
+ if clicked is send_btn:
+ return SEND
+ return CANCEL
diff --git a/app/dialogs/ai_settings_dialog.py b/app/dialogs/ai_settings_dialog.py
new file mode 100644
index 0000000..13474cf
--- /dev/null
+++ b/app/dialogs/ai_settings_dialog.py
@@ -0,0 +1,161 @@
+"""Configuration de l'assistant IA (optionnel, desactive par defaut) : aucune
+liste figee de fournisseurs/modeles, juste une URL de base et un nom de modele
+compatibles avec l'API OpenAI (OpenAI, OpenRouter, proxy interne, LLM local...)."""
+from __future__ import annotations
+
+from PySide6.QtWidgets import (
+ QDialog, QVBoxLayout, QFormLayout, QLineEdit, QCheckBox, QDoubleSpinBox,
+ QPushButton, QDialogButtonBox, QMessageBox, QLabel,
+)
+
+from app.ai.settings import AISettings, load_ai_settings, save_ai_settings, load_api_key, save_api_key
+from app.ai.provider import OpenAICompatibleProvider
+from app.ai.service import KIND_TEST, AIRequestWorker
+from app.credentials import CredentialStore, NullCredentialStore
+
+
+class AISettingsDialog(QDialog):
+ def __init__(self, credential_store: CredentialStore | None = None, parent=None):
+ super().__init__(parent)
+ self.credential_store = credential_store or NullCredentialStore()
+ self._test_worker: AIRequestWorker | None = None
+ self.setWindowTitle("Assistant IA")
+ self.setMinimumWidth(460)
+
+ settings = load_ai_settings()
+ api_key = load_api_key(self.credential_store)
+
+ layout = QVBoxLayout(self)
+ form = QFormLayout()
+
+ self.enabled_check = QCheckBox("Activer l'assistant IA")
+ self.enabled_check.setChecked(settings.enabled)
+ self.enabled_check.setToolTip(
+ "Desactive par defaut. Aucune donnee n'est envoyee a un service externe "
+ "tant que cette case n'est pas cochee et qu'un modele n'est pas configure."
+ )
+ form.addRow("", self.enabled_check)
+
+ self.provider_name_edit = QLineEdit(settings.provider_name)
+ self.provider_name_edit.setPlaceholderText("ex : OpenAI, OpenRouter, LLM local...")
+ form.addRow("Fournisseur :", self.provider_name_edit)
+
+ self.base_url_edit = QLineEdit(settings.base_url)
+ self.base_url_edit.setPlaceholderText("https://api.openai.com/v1")
+ form.addRow("URL de base :", self.base_url_edit)
+
+ self.model_edit = QLineEdit(settings.model)
+ self.model_edit.setPlaceholderText("ex : gpt-4o-mini")
+ form.addRow("Modele :", self.model_edit)
+
+ self.api_key_edit = QLineEdit(api_key)
+ self.api_key_edit.setEchoMode(QLineEdit.Password)
+ self.api_key_edit.setPlaceholderText("Laisser vide si le fournisseur n'en requiert pas")
+ if not self.credential_store.is_available():
+ self.api_key_edit.setEnabled(False)
+ self.api_key_edit.setPlaceholderText("Aucun trousseau securise disponible sur ce systeme")
+ form.addRow("Cle API :", self.api_key_edit)
+
+ self.timeout_spin = QDoubleSpinBox()
+ self.timeout_spin.setRange(1.0, 600.0)
+ self.timeout_spin.setSuffix(" s")
+ self.timeout_spin.setValue(settings.timeout)
+ form.addRow("Delai d'attente :", self.timeout_spin)
+
+ self.temperature_spin = QDoubleSpinBox()
+ self.temperature_spin.setRange(0.0, 2.0)
+ self.temperature_spin.setSingleStep(0.1)
+ self.temperature_spin.setValue(settings.temperature)
+ form.addRow("Temperature :", self.temperature_spin)
+
+ layout.addLayout(form)
+
+ note = QLabel(
+ "La cle API est memorisee dans le trousseau securise du systeme, jamais "
+ "dans un fichier de configuration. L'assistant est entierement optionnel : "
+ "aucune fonctionnalite existante n'en depend."
+ )
+ note.setWordWrap(True)
+ note.setObjectName("statusLabel")
+ layout.addWidget(note)
+
+ test_row = QVBoxLayout()
+ self.test_btn = QPushButton("Tester la connexion")
+ self.test_btn.clicked.connect(self._test_connection)
+ self.test_status = QLabel("")
+ self.test_status.setWordWrap(True)
+ test_row.addWidget(self.test_btn)
+ test_row.addWidget(self.test_status)
+ layout.addLayout(test_row)
+
+ buttons = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
+ buttons.accepted.connect(self._on_accept)
+ buttons.rejected.connect(self.reject)
+ layout.addWidget(buttons)
+
+ # ---------- lecture du formulaire ----------
+ def _form_settings(self) -> AISettings:
+ return AISettings(
+ enabled=self.enabled_check.isChecked(),
+ provider_name=self.provider_name_edit.text().strip() or "OpenAI",
+ base_url=self.base_url_edit.text().strip(),
+ model=self.model_edit.text().strip(),
+ timeout=self.timeout_spin.value(),
+ temperature=self.temperature_spin.value(),
+ )
+
+ def _build_provider(self) -> OpenAICompatibleProvider | None:
+ settings = self._form_settings()
+ if not settings.base_url or not settings.model:
+ QMessageBox.warning(
+ self, "Configuration incomplete",
+ "Renseignez au moins l'URL de base et le modele avant de tester.",
+ )
+ return None
+ return OpenAICompatibleProvider(
+ base_url=settings.base_url,
+ api_key=self.api_key_edit.text().strip(),
+ model=settings.model,
+ timeout=settings.timeout,
+ temperature=settings.temperature,
+ )
+
+ # ---------- test de connexion ----------
+ def _test_connection(self) -> None:
+ if self._test_worker is not None and self._test_worker.isRunning():
+ return
+ provider = self._build_provider()
+ if provider is None:
+ return
+ self.test_btn.setEnabled(False)
+ self.test_status.setText("Test en cours...")
+ self._test_worker = AIRequestWorker(provider, KIND_TEST, "")
+ self._test_worker.finished_ok.connect(self._on_test_ok)
+ self._test_worker.finished_error.connect(self._on_test_error)
+ self._test_worker.start()
+
+ def _on_test_ok(self, _text: str) -> None:
+ self.test_btn.setEnabled(True)
+ self.test_status.setText("✓ Connexion reussie.")
+
+ def _on_test_error(self, message: str) -> None:
+ self.test_btn.setEnabled(True)
+ self.test_status.setText(f"✗ Echec : {message}")
+
+ # ---------- validation ----------
+ def _on_accept(self) -> None:
+ settings = self._form_settings()
+ if settings.enabled and (not settings.base_url or not settings.model):
+ QMessageBox.warning(
+ self, "Configuration incomplete",
+ "Renseignez au moins l'URL de base et le modele pour activer l'assistant IA.",
+ )
+ return
+ save_ai_settings(settings)
+ save_api_key(self.credential_store, self.api_key_edit.text().strip())
+ self.accept()
+
+ def closeEvent(self, event) -> None:
+ if self._test_worker is not None and self._test_worker.isRunning():
+ self._test_worker.wait(2000)
+ super().closeEvent(event)
diff --git a/app/main_window.py b/app/main_window.py
index 605ba4e..50c99ca 100644
--- a/app/main_window.py
+++ b/app/main_window.py
@@ -21,6 +21,8 @@ from app.dialogs.known_hosts_dialog import KnownHostsDialog
from app.dialogs.workspace_dialog import WorkspaceDialog
from app.dialogs.snippet_dialog import SnippetManagerDialog
from app.dialogs.quick_connect_dialog import QuickConnectDialog
+from app.dialogs.ai_settings_dialog import AISettingsDialog
+from app.ai.settings import build_provider_from_settings
from app.theme import THEME_LABELS, apply_theme, current_theme, save_theme
from app.workspace import WorkspaceStore
from app.snippets import SnippetStore
@@ -52,7 +54,9 @@ class MainWindow(QMainWindow):
self.broadcast = BroadcastManager(self._iter_tabs, self)
self.connection_controller = ConnectionController(
self.session_store, self.key_manager, self.broadcast,
- snippet_store=self.snippet_store, credential_store=self.credential_store, parent_widget=self,
+ snippet_store=self.snippet_store, credential_store=self.credential_store,
+ ai_provider_factory=lambda: build_provider_from_settings(self.credential_store),
+ parent_widget=self,
)
self.session_controller = SessionController(
self.session_store, self.key_manager, self.broadcast.button, parent_widget=self,
@@ -189,6 +193,11 @@ class MainWindow(QMainWindow):
manage_snippets_action.triggered.connect(self.open_snippet_manager)
snippets_menu.addAction(manage_snippets_action)
+ ai_menu = self.menuBar().addMenu("&Assistant IA")
+ ai_settings_action = QAction("Parametres...", self)
+ ai_settings_action.triggered.connect(self.open_ai_settings)
+ ai_menu.addAction(ai_settings_action)
+
help_menu = self.menuBar().addMenu("&Aide")
about_action = QAction("A propos", self)
about_action.triggered.connect(self.show_about)
@@ -246,6 +255,10 @@ class MainWindow(QMainWindow):
dialog = SnippetManagerDialog(self.snippet_store, parent=self)
dialog.exec()
+ def open_ai_settings(self) -> None:
+ dialog = AISettingsDialog(credential_store=self.credential_store, parent=self)
+ dialog.exec()
+
def open_quick_connect(self) -> None:
dialog = QuickConnectDialog(
self.session_store, self.connection_controller, self.workspace_manager,
diff --git a/app/ui/ai_panel.py b/app/ui/ai_panel.py
new file mode 100644
index 0000000..2e7a913
--- /dev/null
+++ b/app/ui/ai_panel.py
@@ -0,0 +1,171 @@
+"""Panneau lateral de l'assistant IA : conversation en memoire uniquement
+(jamais persistee, jamais journalisee), une requete a la fois.
+
+Point d'entree unique (`run_request`) pour toutes les actions IA d'un onglet :
+c'est ici, et seulement ici, que se fait la verification de securite (avis
+unique + detection de secrets) avant tout envoi au fournisseur configure.
+"""
+from __future__ import annotations
+
+import html
+
+from PySide6.QtWidgets import (
+ QWidget, QVBoxLayout, QHBoxLayout, QTextEdit, QPushButton, QLabel, QLineEdit, QApplication,
+)
+
+from app.ai.service import AIRequestWorker, KIND_ASK
+from app.ai.redaction import redact
+from app.dialogs.ai_prompts import ensure_ai_notice_acknowledged, confirm_sensitive_content, MASK, CANCEL
+
+_KIND_LABELS = {
+ "explain": "Expliquer cette sortie",
+ "diagnose": "Diagnostiquer cette erreur",
+ "solution": "Proposer une solution",
+ "ask": "Question",
+}
+
+
+def _escape(text: str) -> str:
+ return html.escape(text).replace("\n", "
")
+
+
+class AIPanel(QWidget):
+ """`provider_factory` est un callable sans argument renvoyant un AIProvider
+ pret a l'emploi, ou None si l'assistant est desactive/non configure -
+ rappele a chaque requete pour refleter immediatement un changement de
+ parametres, sans avoir a rouvrir l'onglet."""
+
+ def __init__(self, provider_factory, parent=None):
+ super().__init__(parent)
+ self.provider_factory = provider_factory
+ self._worker: AIRequestWorker | None = None
+ self._generation = 0
+ self._last_response = ""
+ self._pending_context = ""
+
+ layout = QVBoxLayout(self)
+ layout.setContentsMargins(6, 6, 6, 6)
+
+ header = QHBoxLayout()
+ header.addWidget(QLabel("🤖 Assistant IA"), 1)
+ close_btn = QPushButton("✕")
+ close_btn.setFixedWidth(28)
+ close_btn.setToolTip("Masquer le panneau")
+ close_btn.clicked.connect(self.hide)
+ header.addWidget(close_btn)
+ layout.addLayout(header)
+
+ self.conversation = QTextEdit()
+ self.conversation.setReadOnly(True)
+ layout.addWidget(self.conversation, 1)
+
+ self.busy_label = QLabel("")
+ self.busy_label.setObjectName("statusLabel")
+ layout.addWidget(self.busy_label)
+
+ self.question_row = QWidget()
+ question_layout = QHBoxLayout(self.question_row)
+ question_layout.setContentsMargins(0, 0, 0, 0)
+ self.question_edit = QLineEdit()
+ self.question_edit.setPlaceholderText("Votre question...")
+ self.question_edit.returnPressed.connect(self._send_question)
+ send_btn = QPushButton("Envoyer")
+ send_btn.clicked.connect(self._send_question)
+ question_layout.addWidget(self.question_edit, 1)
+ question_layout.addWidget(send_btn)
+ self.question_row.hide()
+ layout.addWidget(self.question_row)
+
+ btn_row = QHBoxLayout()
+ self.copy_btn = QPushButton("Copier reponse")
+ self.new_question_btn = QPushButton("Nouvelle question")
+ self.clear_btn = QPushButton("Effacer la conversation")
+ self.copy_btn.clicked.connect(self._copy_last_response)
+ self.new_question_btn.clicked.connect(lambda: self.prompt_question(""))
+ self.clear_btn.clicked.connect(self.clear_conversation)
+ for b in (self.copy_btn, self.new_question_btn, self.clear_btn):
+ btn_row.addWidget(b)
+ layout.addLayout(btn_row)
+
+ # ---------- conversation ----------
+ def clear_conversation(self) -> None:
+ self._generation += 1 # toute reponse en vol devient perimee et sera ignoree
+ self.conversation.clear()
+ self._last_response = ""
+ self.busy_label.clear()
+ self.question_row.hide()
+
+ def _copy_last_response(self) -> None:
+ if self._last_response:
+ QApplication.clipboard().setText(self._last_response)
+
+ # ---------- question libre ----------
+ def prompt_question(self, context: str = "") -> None:
+ """Affiche le panneau et la zone de saisie d'une question. `context`
+ est un texte optionnel (ex : selection du terminal) joint a la
+ question, propose a la meme verification de secrets avant l'envoi."""
+ self.show()
+ self._pending_context = context
+ self.question_row.show()
+ self.question_edit.setFocus()
+
+ def _send_question(self) -> None:
+ question = self.question_edit.text().strip()
+ if not question:
+ return
+ self.question_edit.clear()
+ self.question_row.hide()
+ context, self._pending_context = self._pending_context, ""
+ self.run_request(KIND_ASK, question, context=context, label=_KIND_LABELS["ask"])
+
+ # ---------- requetes ----------
+ def run_request(self, kind: str, payload: str, context: str = "", label: str = "") -> None:
+ if self._worker is not None and self._worker.isRunning():
+ self.busy_label.setText("Une requete est deja en cours, patientez...")
+ return
+ provider = self.provider_factory() if self.provider_factory else None
+ if provider is None:
+ self.show()
+ self.conversation.append("Assistant IA non configure (menu Assistant IA > Parametres...).")
+ return
+
+ if not ensure_ai_notice_acknowledged(self):
+ return
+ to_check = payload + ("\n" + context if context else "")
+ decision = confirm_sensitive_content(self, to_check)
+ if decision == CANCEL:
+ return
+ if decision == MASK:
+ payload = redact(payload)
+ context = redact(context) if context else context
+
+ self.show()
+ label = label or _KIND_LABELS.get(kind, kind)
+ self.conversation.append(f"{html.escape(label)}
{_escape(payload)}")
+ self.busy_label.setText("Interrogation du fournisseur IA...")
+ generation = self._generation
+ self._worker = AIRequestWorker(provider, kind, payload, context=context)
+ self._worker.finished_ok.connect(lambda text: self._on_ok(generation, text))
+ self._worker.finished_error.connect(lambda message: self._on_error(generation, message))
+ self._worker.start()
+
+ def _on_ok(self, generation: int, text: str) -> None:
+ if generation != self._generation:
+ return
+ self.busy_label.clear()
+ self._last_response = text
+ self.conversation.append(f"{_escape(text)}
")
+
+ def _on_error(self, generation: int, message: str) -> None:
+ if generation != self._generation:
+ return
+ self.busy_label.clear()
+ self.conversation.append(f"Erreur : {_escape(message)}")
+
+ def shutdown(self) -> None:
+ """A appeler avant la destruction du panneau (fermeture de l'onglet) :
+ une requete deja en vol devient perimee (voir `_generation`) et son
+ thread dispose d'un court delai pour se terminer proprement."""
+ self._generation += 1
+ if self._worker is not None and self._worker.isRunning():
+ self._worker.wait(2000)
diff --git a/app/ui/tabs.py b/app/ui/tabs.py
index cdc977a..a43998f 100644
--- a/app/ui/tabs.py
+++ b/app/ui/tabs.py
@@ -15,7 +15,7 @@ from PySide6.QtCore import Qt, Signal, QTimer
from PySide6.QtGui import QKeySequence, QShortcut
from PySide6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QToolButton, QLabel, QMessageBox, QSplitter,
- QFrame, QLineEdit, QFileDialog, QMenu,
+ QFrame, QLineEdit, QFileDialog, QMenu, QInputDialog,
)
from app.session_manager import Session, AUTH_KEY, AUTH_PASSWORD
@@ -30,6 +30,12 @@ from app.dialogs.connect_prompt import ask_secret, confirm_host_key, confirm_hos
from app.services.ssh_target import detect_ssh_target
from app.services.reconnect import compute_backoff_delay
from app.snippets import Snippet, SnippetStore
+from app.ai.service import KIND_EXPLAIN, KIND_DIAGNOSE, KIND_SOLUTION, KIND_GENERATE, AIRequestWorker
+from app.ai.danger import check_dangerous
+from app.ai.redaction import redact
+from app.ui.ai_panel import AIPanel
+from app.dialogs.ai_command_dialog import AICommandDialog
+from app.dialogs.ai_prompts import ensure_ai_notice_acknowledged, confirm_sensitive_content, MASK, CANCEL
# sequences d'echappement ANSI, retirees des journaux pour qu'ils restent lisibles
ANSI_ESCAPE_RE = re.compile(
@@ -48,11 +54,15 @@ class BaseTerminalTab(QWidget):
def __init__(
self, initial_status: str, snippet_store: SnippetStore | None = None,
- broadcast=None, parent=None,
+ broadcast=None, ai_provider_factory=None, credential_store=None, parent=None,
):
super().__init__(parent)
self.snippet_store = snippet_store
self.broadcast = broadcast
+ self.ai_provider_factory = ai_provider_factory
+ self.credential_store = credential_store
+ self.ai_panel: AIPanel | None = None
+ self._ai_command_worker: AIRequestWorker | None = None
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
@@ -82,6 +92,13 @@ class BaseTerminalTab(QWidget):
self.snippets_btn.clicked.connect(self._show_snippets_menu)
self.bar.addWidget(self.snippets_btn)
+ self.ai_btn = QToolButton()
+ self.ai_btn.setText("🤖 Assistant IA")
+ self.ai_btn.setToolTip("Analyser la sortie selectionnee ou generer une commande (optionnel)")
+ self.ai_btn.setPopupMode(QToolButton.InstantPopup)
+ self.ai_btn.clicked.connect(self._show_ai_menu)
+ self.bar.addWidget(self.ai_btn)
+
self.log_btn = QToolButton()
self.log_btn.setText("⏺ Journal")
self.log_btn.setToolTip("Enregistrer la session dans un fichier")
@@ -269,6 +286,141 @@ class BaseTerminalTab(QWidget):
elif self.connection:
self.connection.send(data)
+ # ---------- assistant IA (optionnel) ----------
+ def _show_ai_menu(self) -> None:
+ menu = self._build_ai_menu()
+ menu.exec(self.ai_btn.mapToGlobal(self.ai_btn.rect().bottomLeft()))
+
+ def _build_ai_menu(self) -> QMenu:
+ menu = QMenu(self)
+ has_selection = self.terminal.has_selection()
+
+ explain_action = menu.addAction("Expliquer cette sortie")
+ explain_action.setEnabled(has_selection)
+ explain_action.triggered.connect(self._ai_explain)
+
+ diagnose_action = menu.addAction("Diagnostiquer cette erreur")
+ diagnose_action.setEnabled(has_selection)
+ diagnose_action.triggered.connect(self._ai_diagnose)
+
+ solution_action = menu.addAction("Proposer une solution")
+ solution_action.setEnabled(has_selection)
+ solution_action.triggered.connect(self._ai_solution)
+
+ ask_action = menu.addAction("Poser une question...")
+ ask_action.setEnabled(has_selection)
+ ask_action.triggered.connect(self._ai_ask)
+
+ if not has_selection:
+ for action in (explain_action, diagnose_action, solution_action, ask_action):
+ action.setToolTip("Selectionnez d'abord du texte dans le terminal.")
+
+ menu.addSeparator()
+ generate_action = menu.addAction("Generer une commande...")
+ generate_action.triggered.connect(self._ai_generate_command)
+
+ menu.addSeparator()
+ panel_visible = self.ai_panel is not None and self.ai_panel.isVisible()
+ toggle_action = menu.addAction("Masquer le panneau IA" if panel_visible else "Afficher le panneau IA")
+ toggle_action.triggered.connect(self._toggle_ai_panel)
+
+ menu.addSeparator()
+ settings_action = menu.addAction("Parametres de l'assistant IA...")
+ settings_action.triggered.connect(self._open_ai_settings)
+ return menu
+
+ def _ensure_ai_panel(self) -> AIPanel:
+ if self.ai_panel is None:
+ self.ai_panel = AIPanel(self.ai_provider_factory)
+ self.splitter.addWidget(self.ai_panel)
+ # cree masque : sans cela, Qt affiche automatiquement un widget
+ # fraichement ajoute a un splitter deja visible, ce qui romprait
+ # l'etat "ferme par defaut" attendu du panneau.
+ self.ai_panel.hide()
+ return self.ai_panel
+
+ def _toggle_ai_panel(self) -> None:
+ panel = self._ensure_ai_panel()
+ panel.setVisible(not panel.isVisible())
+
+ def _open_ai_settings(self) -> None:
+ from app.dialogs.ai_settings_dialog import AISettingsDialog
+ AISettingsDialog(credential_store=self.credential_store, parent=self).exec()
+
+ def _ai_explain(self) -> None:
+ self._run_ai_selection_action(KIND_EXPLAIN, "Expliquer cette sortie")
+
+ def _ai_diagnose(self) -> None:
+ self._run_ai_selection_action(KIND_DIAGNOSE, "Diagnostiquer cette erreur")
+
+ def _ai_solution(self) -> None:
+ self._run_ai_selection_action(KIND_SOLUTION, "Proposer une solution")
+
+ def _run_ai_selection_action(self, kind: str, label: str) -> None:
+ selection = self.terminal.selected_text()
+ if not selection:
+ return
+ panel = self._ensure_ai_panel()
+ panel.run_request(kind, selection, label=label)
+
+ def _ai_ask(self) -> None:
+ selection = self.terminal.selected_text()
+ if not selection:
+ return
+ panel = self._ensure_ai_panel()
+ panel.prompt_question(context=selection)
+
+ def _ai_generate_command(self) -> None:
+ if self._ai_command_worker is not None and self._ai_command_worker.isRunning():
+ return
+ provider = self.ai_provider_factory() if self.ai_provider_factory else None
+ if provider is None:
+ QMessageBox.information(
+ self, "Assistant IA",
+ "L'assistant IA n'est pas configure (menu Assistant IA > Parametres de l'assistant IA...).",
+ )
+ return
+ request, ok = QInputDialog.getMultiLineText(
+ self, "Generer une commande", "Decrivez ce que vous voulez accomplir :",
+ )
+ if not ok or not request.strip():
+ return
+ request = request.strip()
+ if not ensure_ai_notice_acknowledged(self):
+ return
+ decision = confirm_sensitive_content(self, request)
+ if decision == CANCEL:
+ return
+ if decision == MASK:
+ request = redact(request)
+
+ self.ai_btn.setEnabled(False)
+ self.ai_btn.setText("🤖 Generation...")
+ self._ai_command_worker = AIRequestWorker(provider, KIND_GENERATE, request)
+ self._ai_command_worker.finished_ok.connect(self._on_ai_command_ok)
+ self._ai_command_worker.finished_error.connect(self._on_ai_command_error)
+ self._ai_command_worker.start()
+
+ def _reset_ai_generate_button(self) -> None:
+ self.ai_btn.setEnabled(True)
+ self.ai_btn.setText("🤖 Assistant IA")
+
+ def _on_ai_command_ok(self, response: str) -> None:
+ self._reset_ai_generate_button()
+ dangerous = [m.label for m in check_dangerous(response)]
+ dialog = AICommandDialog(response, dangerous_labels=dangerous, parent=self)
+ if dialog.exec() and dialog.insert_requested and dialog.command:
+ self.insert_snippet_text(dialog.command)
+
+ def _on_ai_command_error(self, message: str) -> None:
+ self._reset_ai_generate_button()
+ QMessageBox.critical(self, "Assistant IA", message)
+
+ def insert_snippet_text(self, text: str) -> None:
+ """N'envoie jamais Entree : le texte est ecrit, a l'utilisateur de valider."""
+ if self.connection:
+ self.connection.send(text.encode("utf-8"))
+
# ---------- recherche dans l'historique ----------
def open_search(self) -> None:
self.search_bar.show()
@@ -290,6 +442,10 @@ class BaseTerminalTab(QWidget):
def close_connection(self) -> None:
self._closing = True # ne jamais reconnecter automatiquement une fermeture voulue
self.stop_logging()
+ if self.ai_panel is not None:
+ self.ai_panel.shutdown()
+ if self._ai_command_worker is not None and self._ai_command_worker.isRunning():
+ self._ai_command_worker.wait(2000)
if self.connection:
self.connection.close_connection()
self.connection.wait(2000)
@@ -300,11 +456,12 @@ class SessionTabWidget(BaseTerminalTab):
def __init__(
self, session: Session, secret: str, bastion_target=None, bastion_password: str = "",
- snippet_store=None, broadcast=None, parent=None,
+ snippet_store=None, broadcast=None, ai_provider_factory=None, credential_store=None, parent=None,
):
super().__init__(
f"○ Connexion a {session.username}@{session.host}...",
- snippet_store=snippet_store, broadcast=broadcast, parent=parent,
+ snippet_store=snippet_store, broadcast=broadcast,
+ ai_provider_factory=ai_provider_factory, credential_store=credential_store, parent=parent,
)
self.session = session
self._bastion_target = bastion_target
@@ -594,10 +751,14 @@ class LocalTerminalTabWidget(BaseTerminalTab):
sftp_requested = Signal(object) # cible ssh detectee (ou None)
- def __init__(self, shell: str = SHELL_POWERSHELL, snippet_store=None, broadcast=None, parent=None):
+ def __init__(
+ self, shell: str = SHELL_POWERSHELL, snippet_store=None, broadcast=None,
+ ai_provider_factory=None, credential_store=None, parent=None,
+ ):
super().__init__(
f"○ Ouverture de {SHELL_LABELS.get(shell, shell)}...",
- snippet_store=snippet_store, broadcast=broadcast, parent=parent,
+ snippet_store=snippet_store, broadcast=broadcast,
+ ai_provider_factory=ai_provider_factory, credential_store=credential_store, parent=parent,
)
self.shell = shell
diff --git a/tests/test_ai_command_dialog.py b/tests/test_ai_command_dialog.py
new file mode 100644
index 0000000..654a14e
--- /dev/null
+++ b/tests/test_ai_command_dialog.py
@@ -0,0 +1,65 @@
+"""AICommandDialog : extraction de la commande depuis la reponse de l'IA
+(bloc de code s'il y en a un), et les 3 actions possibles (Copier / Inserer
+dans le terminal / Annuler) - jamais d'execution automatique."""
+import sys, pathlib
+sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
+
+from PySide6.QtWidgets import QApplication
+
+from app.dialogs.ai_command_dialog import AICommandDialog, extract_command
+
+app = QApplication.instance() or QApplication(sys.argv)
+
+print("=== 1. extract_command() : bloc de code present ===")
+response = "Pour lister les gros fichiers :\n\n```bash\nfind / -size +100M\n```\n\nCela peut prendre du temps."
+assert extract_command(response) == "find / -size +100M"
+print(" OK")
+
+print()
+print("=== 2. extract_command() : bloc de code sans langage indique ===")
+response = "```\ndf -h\n```"
+assert extract_command(response) == "df -h"
+print(" OK")
+
+print()
+print("=== 3. extract_command() : pas de bloc de code -> reponse entiere ===")
+response = "Utilisez simplement : uptime"
+assert extract_command(response) == "Utilisez simplement : uptime"
+print(" OK")
+
+print()
+print("=== 4. Avertissement affiche seulement si la commande est dangereuse ===")
+dialog_safe = AICommandDialog("```\ndf -h\n```")
+assert dialog_safe.command == "df -h"
+dialog_dangerous = AICommandDialog("```\nrm -rf /tmp/x\n```", dangerous_labels=["rm -rf (suppression recursive)"])
+assert dialog_dangerous.command == "rm -rf /tmp/x"
+print(" OK (pas de verification visuelle automatisee, juste la construction sans exception)")
+
+print()
+print("=== 5. 'Copier' place la commande (eventuellement corrigee) dans le presse-papiers ===")
+dialog = AICommandDialog("```\necho bonjour\n```")
+dialog.command_view.setPlainText("echo modifie")
+dialog._copy()
+assert QApplication.clipboard().text() == "echo modifie"
+print(" OK")
+
+print()
+print("=== 6. 'Inserer dans le terminal' : accepte le dialogue, ne l'execute jamais ===")
+dialog = AICommandDialog("```\nls -la\n```")
+assert dialog.insert_requested is False
+dialog._insert()
+assert dialog.insert_requested is True
+assert dialog.command == "ls -la"
+assert dialog.result() == 1 # QDialog.Accepted
+print(" OK")
+
+print()
+print("=== 7. 'Annuler' : rejette le dialogue, insert_requested reste False ===")
+dialog = AICommandDialog("```\nls -la\n```")
+dialog.cancel_btn.click()
+assert dialog.insert_requested is False
+assert dialog.result() == 0 # QDialog.Rejected
+print(" OK")
+
+print()
+print("AI COMMAND DIALOG : TOUS LES TESTS OK")
diff --git a/tests/test_ai_danger.py b/tests/test_ai_danger.py
new file mode 100644
index 0000000..af465cb
--- /dev/null
+++ b/tests/test_ai_danger.py
@@ -0,0 +1,57 @@
+"""Detection locale de commandes potentiellement destructives proposees par
+l'assistant IA (jamais un blocage, seulement un avertissement)."""
+import sys, pathlib
+sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
+
+from app.ai.danger import is_dangerous, check_dangerous
+
+print("=== 1. Commandes destructrices detectees ===")
+dangerous_examples = [
+ "rm -rf /var/log/*",
+ "rm -fr /tmp/build",
+ "mkfs.ext4 /dev/sdb1",
+ "dd if=/dev/zero of=/dev/sda bs=1M",
+ "wipefs -a /dev/sdb",
+ "sudo shutdown -h now",
+ "sudo reboot",
+ "iptables -F",
+ "nft flush ruleset",
+ "DROP DATABASE production;",
+ "DELETE FROM users;",
+ "docker system prune -a --volumes",
+ "zpool destroy tank",
+ "lvremove /dev/vg0/lv0",
+ "vgremove vg0",
+]
+for cmd in dangerous_examples:
+ result = is_dangerous(cmd)
+ print(f" {'[X]' if result else '[ ]'} {cmd}")
+ assert result, f"aurait du etre detectee comme dangereuse : {cmd}"
+print(" OK")
+
+print()
+print("=== 2. Commandes normales non signalees ===")
+safe_examples = [
+ "df -h",
+ "systemctl status nginx",
+ "docker ps -a",
+ "ls -la /var/log",
+ "find /var -type f -printf '%s %p\\n' | sort -nr | head -20",
+ "SELECT * FROM users WHERE id = 1;",
+ "git rm --cached fichier.txt", # 'rm' present mais pas -rf : ne doit pas matcher
+]
+for cmd in safe_examples:
+ result = is_dangerous(cmd)
+ print(f" {'[X]' if result else '[ ]'} {cmd}")
+ assert not result, f"n'aurait pas du etre signalee : {cmd}"
+print(" OK")
+
+print()
+print("=== 3. check_dangerous() renvoie une explication exploitable ===")
+matches = check_dangerous("rm -rf /")
+print(" correspondances:", [m.label for m in matches])
+assert matches and "rm" in matches[0].label.lower()
+print(" OK")
+
+print()
+print("AI DANGER : TOUS LES TESTS OK")
diff --git a/tests/test_ai_panel.py b/tests/test_ai_panel.py
new file mode 100644
index 0000000..75d1ab1
--- /dev/null
+++ b/tests/test_ai_panel.py
@@ -0,0 +1,168 @@
+"""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")
diff --git a/tests/test_ai_prompts.py b/tests/test_ai_prompts.py
new file mode 100644
index 0000000..93e83a1
--- /dev/null
+++ b/tests/test_ai_prompts.py
@@ -0,0 +1,96 @@
+"""Confirmations de securite avant envoi a l'assistant IA : avis de principe
+affiche une seule fois (persist dans app_settings, redirige vers un fichier
+temporaire), et verification des secrets a chaque envoi. QMessageBox est
+simule (exec()/clickedButton() patches) : aucune boite modale reelle ne doit
+bloquer les tests."""
+import sys, pathlib
+sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
+
+import tempfile
+from pathlib import Path
+from unittest.mock import patch
+
+from PySide6.QtWidgets import QApplication, QMessageBox
+
+import app.app_settings as app_settings
+from app.dialogs.ai_prompts import ensure_ai_notice_acknowledged, confirm_sensitive_content, MASK, SEND, CANCEL
+
+app = QApplication.instance() or QApplication(sys.argv)
+
+tmp = Path(tempfile.mkdtemp())
+settings_redirect = patch.object(app_settings, "SETTINGS_FILE", tmp / "settings.json")
+settings_redirect.start()
+
+
+def click(button_text):
+ """Simule un clic sur le bouton `button_text` d'une QMessageBox construite
+ avec addButton(), sans faire tourner de boucle d'evenements reelle."""
+ def fake_clicked(self):
+ for b in self.buttons():
+ if b.text() == button_text:
+ return b
+ return None
+ return fake_clicked
+
+
+print("=== 1. Avis de principe : affiche la premiere fois, refus -> False ===")
+with patch("app.dialogs.ai_prompts.QMessageBox.exec", return_value=QMessageBox.Cancel) as mocked_exec:
+ result = ensure_ai_notice_acknowledged(None)
+ assert mocked_exec.called
+ assert result is False
+assert app_settings.get_setting("ai_notice_acknowledged", False) is False
+print(" OK")
+
+print()
+print("=== 2. Avis de principe : acceptation -> True, persiste ===")
+with patch("app.dialogs.ai_prompts.QMessageBox.exec", return_value=QMessageBox.Ok) as mocked_exec:
+ result = ensure_ai_notice_acknowledged(None)
+ assert mocked_exec.called
+ assert result is True
+assert app_settings.get_setting("ai_notice_acknowledged", False) is True
+print(" OK")
+
+print()
+print("=== 3. Avis de principe : deja accepte -> plus jamais affiche ===")
+with patch("app.dialogs.ai_prompts.QMessageBox.exec") as mocked_exec:
+ result = ensure_ai_notice_acknowledged(None)
+ assert not mocked_exec.called, "la boite ne doit plus s'afficher une fois l'accord persiste"
+ assert result is True
+print(" OK")
+
+print()
+print("=== 4. Aucun secret detecte : SEND direct, aucune boite affichee ===")
+with patch("app.dialogs.ai_prompts.QMessageBox.exec") as mocked_exec:
+ result = confirm_sensitive_content(None, "systemctl status nginx")
+ assert not mocked_exec.called
+ assert result == SEND
+print(" OK")
+
+print()
+print("=== 5. Secret detecte + 'Masquer automatiquement' -> MASK ===")
+with patch("app.dialogs.ai_prompts.QMessageBox.exec", return_value=0), \
+ patch("app.dialogs.ai_prompts.QMessageBox.clickedButton", click("Masquer automatiquement")):
+ result = confirm_sensitive_content(None, "password=Sup3rSecret!")
+ assert result == MASK
+print(" OK")
+
+print()
+print("=== 6. Secret detecte + 'Voir le contenu' -> SEND (envoi tel quel) ===")
+with patch("app.dialogs.ai_prompts.QMessageBox.exec", return_value=0), \
+ patch("app.dialogs.ai_prompts.QMessageBox.clickedButton", click("Voir le contenu")):
+ result = confirm_sensitive_content(None, "password=Sup3rSecret!")
+ assert result == SEND
+print(" OK")
+
+print()
+print("=== 7. Secret detecte + 'Annuler' -> CANCEL ===")
+with patch("app.dialogs.ai_prompts.QMessageBox.exec", return_value=0), \
+ patch("app.dialogs.ai_prompts.QMessageBox.clickedButton", click("Annuler")):
+ result = confirm_sensitive_content(None, "password=Sup3rSecret!")
+ assert result == CANCEL
+print(" OK")
+
+settings_redirect.stop()
+
+print()
+print("AI PROMPTS : TOUS LES TESTS OK")
diff --git a/tests/test_ai_provider.py b/tests/test_ai_provider.py
new file mode 100644
index 0000000..26ba0d7
--- /dev/null
+++ b/tests/test_ai_provider.py
@@ -0,0 +1,157 @@
+"""OpenAICompatibleProvider : requete HTTP construite correctement, et gestion
+des erreurs (401, 429, timeout, reseau, reponse invalide) - urllib.request.urlopen
+est entierement mocke, aucun appel reseau reel."""
+import sys, pathlib
+sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
+
+import io
+import json
+import urllib.error
+from unittest.mock import patch, MagicMock
+
+from app.ai.provider import AIProvider, AIError, OpenAICompatibleProvider
+
+
+def fake_response(payload: dict):
+ m = MagicMock()
+ m.read.return_value = json.dumps(payload).encode("utf-8")
+ m.__enter__.return_value = m
+ m.__exit__.return_value = False
+ return m
+
+
+def make_http_error(code: int, body: bytes = b"") -> urllib.error.HTTPError:
+ return urllib.error.HTTPError(url="http://x/chat/completions", code=code, msg="err",
+ hdrs=None, fp=io.BytesIO(body))
+
+
+provider = OpenAICompatibleProvider(
+ base_url="https://api.example.com/v1", api_key="sk-test", model="gpt-test", timeout=5, temperature=0.2,
+)
+
+print("=== 1. Requete construite correctement (URL, methode, en-tetes, corps) ===")
+with patch("app.ai.provider.urllib.request.urlopen") as mocked:
+ mocked.return_value = fake_response({"choices": [{"message": {"content": "bonjour"}}]})
+ result = provider.chat([{"role": "user", "content": "salut"}])
+
+request = mocked.call_args.args[0]
+print(" URL:", request.full_url)
+print(" methode:", request.get_method())
+print(" Authorization:", request.get_header("Authorization"))
+body = json.loads(request.data.decode("utf-8"))
+print(" corps:", body)
+assert request.full_url == "https://api.example.com/v1/chat/completions"
+assert request.get_method() == "POST"
+assert request.get_header("Authorization") == "Bearer sk-test"
+assert body["model"] == "gpt-test"
+assert body["temperature"] == 0.2
+assert body["messages"] == [{"role": "user", "content": "salut"}]
+assert result == "bonjour"
+print(" OK")
+
+print()
+print("=== 2. Erreur 401 -> message clair mentionnant la cle API ===")
+with patch("app.ai.provider.urllib.request.urlopen", side_effect=make_http_error(401)):
+ try:
+ provider.chat([{"role": "user", "content": "x"}])
+ ok = False
+ except AIError as exc:
+ ok = True
+ print(" message:", exc)
+ assert "401" in str(exc) and "cle" in str(exc).lower()
+assert ok
+print(" OK")
+
+print()
+print("=== 3. Erreur 429 -> message clair mentionnant la limitation ===")
+with patch("app.ai.provider.urllib.request.urlopen", side_effect=make_http_error(429)):
+ try:
+ provider.chat([{"role": "user", "content": "x"}])
+ ok = False
+ except AIError as exc:
+ ok = True
+ print(" message:", exc)
+ assert "429" in str(exc)
+assert ok
+print(" OK")
+
+print()
+print("=== 4. Timeout -> message clair, pas une exception brute ===")
+timeout_error = urllib.error.URLError(reason=TimeoutError("timed out"))
+with patch("app.ai.provider.urllib.request.urlopen", side_effect=timeout_error):
+ try:
+ provider.chat([{"role": "user", "content": "x"}])
+ ok = False
+ except AIError as exc:
+ ok = True
+ print(" message:", exc)
+ assert "delai" in str(exc).lower() or "temps" in str(exc).lower()
+assert ok
+print(" OK")
+
+print()
+print("=== 5. Erreur reseau (DNS, connexion refusee) -> message clair ===")
+network_error = urllib.error.URLError(reason="nom ou service inconnu")
+with patch("app.ai.provider.urllib.request.urlopen", side_effect=network_error):
+ try:
+ provider.chat([{"role": "user", "content": "x"}])
+ ok = False
+ except AIError as exc:
+ ok = True
+ print(" message:", exc)
+assert ok
+print(" OK")
+
+print()
+print("=== 6. Reponse invalide (JSON incoherent) -> message clair, pas de KeyError brut ===")
+with patch("app.ai.provider.urllib.request.urlopen") as mocked:
+ mocked.return_value = fake_response({"unexpected": "shape"})
+ try:
+ provider.chat([{"role": "user", "content": "x"}])
+ ok = False
+ except AIError as exc:
+ ok = True
+ print(" message:", exc)
+assert ok
+print(" OK")
+
+print()
+print("=== 7. Sans cle API : pas d'en-tete Authorization ===")
+no_key_provider = OpenAICompatibleProvider(base_url="http://localhost:8080/v1", api_key="", model="local-model")
+with patch("app.ai.provider.urllib.request.urlopen") as mocked:
+ mocked.return_value = fake_response({"choices": [{"message": {"content": "ok"}}]})
+ no_key_provider.chat([{"role": "user", "content": "x"}])
+request2 = mocked.call_args.args[0]
+assert request2.get_header("Authorization") is None
+print(" OK (utile pour un LLM local sans authentification)")
+
+print()
+print("=== 8. Les 4 usages metier construisent les bons messages (via chat()) ===")
+captured = []
+
+
+class SpyProvider(AIProvider):
+ def chat(self, messages, *, temperature=None):
+ captured.append(messages)
+ return "reponse"
+
+
+spy = SpyProvider()
+spy.explain_output("erreur X")
+spy.diagnose("erreur Y")
+spy.propose_solution("erreur Z")
+spy.generate_command("trouver les gros fichiers")
+spy.ask("pourquoi ?", context="ligne de contexte")
+
+print(" nombre d'appels a chat():", len(captured))
+assert len(captured) == 5
+assert "erreur X" in captured[0][-1]["content"]
+assert "Cause probable" in captured[1][0]["content"] # structure imposee au system prompt de diagnose
+assert "erreur Z" in captured[2][-1]["content"]
+assert captured[3][-1]["content"] == "trouver les gros fichiers"
+assert "ligne de contexte" in captured[4][1]["content"]
+assert captured[4][-1]["content"] == "pourquoi ?"
+print(" OK")
+
+print()
+print("AI PROVIDER : TOUS LES TESTS OK")
diff --git a/tests/test_ai_redaction.py b/tests/test_ai_redaction.py
new file mode 100644
index 0000000..7d66d81
--- /dev/null
+++ b/tests/test_ai_redaction.py
@@ -0,0 +1,60 @@
+"""Detection/redaction de secrets avant envoi a l'assistant IA."""
+import sys, pathlib
+sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
+
+from app.ai.redaction import detect_secrets, redact
+
+print("=== 1. Cle privee PEM detectee ===")
+text = "-----BEGIN RSA PRIVATE KEY-----\nMIIEowIBAAKCAQEA...\n-----END RSA PRIVATE KEY-----"
+found = detect_secrets(text)
+print(" detecte:", [f.label for f in found])
+assert any("privee" in f.label.lower() for f in found)
+print(" OK")
+
+print()
+print("=== 2. Jeton Bearer detecte ===")
+found = detect_secrets("Authorization: Bearer sk-abcdefghijklmnopqrstuvwxyz123456")
+labels = [f.label for f in found]
+print(" detecte:", labels)
+assert any("bearer" in l.lower() for l in labels)
+assert any("authorization" in l.lower() for l in labels)
+print(" OK")
+
+print()
+print("=== 3. Cle AWS detectee ===")
+found = detect_secrets("export AWS_ACCESS_KEY_ID=AKIAIOSFODNN7EXAMPLE")
+assert any("aws" in f.label.lower() for f in found)
+print(" OK")
+
+print()
+print("=== 4. password= / api_key= / token= detectes ===")
+assert any("mot de passe" in f.label.lower() for f in detect_secrets("password=Sup3rSecret!"))
+assert any("api" in f.label.lower() for f in detect_secrets("api_key=abcd1234"))
+assert any("jeton" in f.label.lower() for f in detect_secrets("token: eyJhbGciOiJIUzI1NiJ9"))
+print(" OK")
+
+print()
+print("=== 5. Chaine longue ressemblant a un secret ===")
+found = detect_secrets("valeur=" + "a1b2c3d4e5f6g7h8i9j0" * 3)
+assert any("longue" in f.label.lower() for f in found)
+print(" OK")
+
+print()
+print("=== 6. Texte anodin : rien detecte ===")
+assert detect_secrets("systemd[1]: nginx.service: Failed with result 'exit-code'") == []
+assert detect_secrets("nginx: [emerg] bind() to 0.0.0.0:443 failed (98: Address already in use)") == []
+print(" OK")
+
+print()
+print("=== 7. redact() masque sans lever, texte anodin inchange ===")
+redacted = redact("password=abc123 et Bearer xyzxyzxyzxyzxyzxyz")
+print(" resultat:", redacted)
+assert "abc123" not in redacted
+assert "xyzxyzxyzxyzxyzxyz" not in redacted
+assert "[MASQUE]" in redacted
+clean = "systemctl status nginx"
+assert redact(clean) == clean
+print(" OK")
+
+print()
+print("AI REDACTION : TOUS LES TESTS OK")
diff --git a/tests/test_ai_service.py b/tests/test_ai_service.py
new file mode 100644
index 0000000..a4fed32
--- /dev/null
+++ b/tests/test_ai_service.py
@@ -0,0 +1,168 @@
+"""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 sys, pathlib
+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 AIProvider, AIError
+from app.ai.service import (
+ AIRequestWorker, KIND_EXPLAIN, KIND_DIAGNOSE, KIND_SOLUTION, KIND_GENERATE, KIND_ASK, KIND_TEST,
+)
+
+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")
diff --git a/tests/test_ai_settings.py b/tests/test_ai_settings.py
new file mode 100644
index 0000000..91cf604
--- /dev/null
+++ b/tests/test_ai_settings.py
@@ -0,0 +1,99 @@
+"""Configuration de l'assistant IA : parametres non sensibles persistes via
+app_settings (redirige vers un fichier temporaire pour ne jamais toucher les
+vrais reglages), cle API exclusivement via CredentialStore, et construction du
+fournisseur a partir des deux."""
+import sys, pathlib
+sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
+
+import tempfile
+from pathlib import Path
+from unittest.mock import patch
+
+from app.credentials import CredentialStore
+from app.ai.settings import (
+ AISettings, load_ai_settings, save_ai_settings, load_api_key, save_api_key,
+ build_provider_from_settings, AI_API_KEY_CREDENTIAL, DEFAULT_BASE_URL,
+)
+from app.ai.provider import OpenAICompatibleProvider
+import app.app_settings as app_settings
+
+
+class FakeCredentialStore(CredentialStore):
+ def __init__(self):
+ self._data = {}
+
+ def is_available(self):
+ return True
+
+ def set_secret(self, key, secret):
+ self._data[key] = secret
+ return True
+
+ def get_secret(self, key):
+ return self._data.get(key)
+
+ def delete_secret(self, key):
+ self._data.pop(key, None)
+
+
+tmp = Path(tempfile.mkdtemp())
+settings_redirect = patch.object(app_settings, "SETTINGS_FILE", tmp / "settings.json")
+settings_redirect.start()
+
+print("=== 1. Valeurs par defaut : desactive, pas de secret expose ===")
+default = load_ai_settings()
+print(" ", default)
+assert default.enabled is False
+assert default.base_url == DEFAULT_BASE_URL
+print(" OK")
+
+print()
+print("=== 2. Sauvegarde/rechargement des parametres non sensibles ===")
+settings = AISettings(enabled=True, provider_name="OpenRouter", base_url="https://openrouter.ai/api/v1",
+ model="anthropic/claude", timeout=45.0, temperature=0.7)
+save_ai_settings(settings)
+reloaded = load_ai_settings()
+print(" ", reloaded)
+assert reloaded == settings
+print(" OK")
+
+print()
+print("=== 3. La cle API ne transite jamais par app_settings (verification du fichier disque) ===")
+raw = (tmp / "settings.json").read_text(encoding="utf-8")
+assert "sk-" not in raw
+assert "api_key" not in raw.lower() or "ai_api" not in raw.lower() # aucune cle presente, seul le nom du champ pourrait
+print(" contenu persiste:", raw)
+print(" OK")
+
+print()
+print("=== 4. Cle API : lecture/ecriture via CredentialStore uniquement ===")
+store = FakeCredentialStore()
+assert load_api_key(store) == ""
+save_api_key(store, "sk-test-secret")
+assert load_api_key(store) == "sk-test-secret"
+assert store.get_secret(AI_API_KEY_CREDENTIAL) == "sk-test-secret"
+save_api_key(store, "") # vider la cle -> supprime l'entree
+assert load_api_key(store) == ""
+print(" OK")
+
+print()
+print("=== 5. build_provider_from_settings() : None si desactive ===")
+save_ai_settings(AISettings(enabled=False))
+assert build_provider_from_settings(store) is None
+print(" OK")
+
+print()
+print("=== 6. build_provider_from_settings() : fournisseur pret si active ===")
+save_ai_settings(AISettings(enabled=True, base_url="https://api.example.com/v1", model="gpt-x"))
+save_api_key(store, "sk-live")
+provider = build_provider_from_settings(store)
+assert isinstance(provider, OpenAICompatibleProvider)
+assert provider.base_url == "https://api.example.com/v1"
+assert provider.model == "gpt-x"
+assert provider.api_key == "sk-live"
+print(" OK")
+
+settings_redirect.stop()
+
+print()
+print("AI SETTINGS : TOUS LES TESTS OK")
diff --git a/tests/test_ai_tab_integration.py b/tests/test_ai_tab_integration.py
new file mode 100644
index 0000000..ecff5ff
--- /dev/null
+++ b/tests/test_ai_tab_integration.py
@@ -0,0 +1,160 @@
+"""Integration de l'assistant IA dans un onglet terminal : menu (active/desactive
+selon la selection), panneau lateral cree a la demande dans le splitter (meme
+mecanisme que le panneau SFTP), et generation de commande (jamais executee
+automatiquement, avec avertissement si dangereuse).
+
+LocalTerminalConnection.start() est neutralisee comme dans test_snippets.py :
+aucun sous-processus reel, send() depose seulement les octets dans une file."""
+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.local_terminal import LocalTerminalConnection
+from app.ai.provider import AIProvider
+from app.dialogs.ai_prompts import SEND
+from app.ui.ai_panel import AIPanel
+from app.ui.tabs import LocalTerminalTabWidget
+
+app = QApplication.instance() or QApplication(sys.argv)
+
+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)
+PROMPTS_OK_IN_TABS = patch("app.ui.tabs.ensure_ai_notice_acknowledged", return_value=True)
+SEND_ASIS_IN_TABS = patch("app.ui.tabs.confirm_sensitive_content", return_value=SEND)
+
+
+class FakeProvider(AIProvider):
+ def __init__(self, reply="reponse"):
+ self.reply = reply
+
+ def chat(self, messages, *, temperature=None):
+ return self.reply
+
+
+def pump(ms=3000):
+ import time
+ deadline = time.time() + ms / 1000
+ while time.time() < deadline:
+ QCoreApplication.processEvents()
+ time.sleep(0.01)
+
+
+def make_tab(ai_provider_factory=None):
+ return LocalTerminalTabWidget(ai_provider_factory=ai_provider_factory)
+
+
+print("=== 1. Menu IA : actions de selection desactivees sans selection ===")
+with patch.object(LocalTerminalConnection, "start", lambda self: None):
+ tab = make_tab()
+ assert not tab.terminal.has_selection()
+ menu = tab._build_ai_menu()
+ labels = [a.text() for a in menu.actions() if a.text()]
+ print(" actions:", labels)
+ assert "Expliquer cette sortie" in labels
+ assert "Generer une commande..." in labels
+ for text in ("Expliquer cette sortie", "Diagnostiquer cette erreur", "Proposer une solution", "Poser une question..."):
+ action = next(a for a in menu.actions() if a.text() == text)
+ assert not action.isEnabled(), f"{text} devrait etre desactive sans selection"
+ generate_action = next(a for a in menu.actions() if a.text() == "Generer une commande...")
+ assert generate_action.isEnabled(), "Generer une commande ne necessite pas de selection"
+ tab.close_connection()
+print(" OK")
+
+print()
+print("=== 2. Menu IA : actions de selection activees des qu'il y a une selection ===")
+with patch.object(LocalTerminalConnection, "start", lambda self: None):
+ tab = make_tab()
+ tab.terminal.feed(b"erreur : connection refused\r\n")
+ tab.terminal._sel_anchor, tab.terminal._sel_focus = (0, 0), (10, 0)
+ assert tab.terminal.has_selection()
+ menu = tab._build_ai_menu()
+ for text in ("Expliquer cette sortie", "Diagnostiquer cette erreur", "Proposer une solution", "Poser une question..."):
+ action = next(a for a in menu.actions() if a.text() == text)
+ assert action.isEnabled(), f"{text} devrait etre active avec une selection"
+ tab.close_connection()
+print(" OK")
+
+print()
+print("=== 3. 'Expliquer cette sortie' : cree le panneau IA dans le splitter et lance la requete ===")
+with patch.object(LocalTerminalConnection, "start", lambda self: None):
+ provider = FakeProvider(reply="Ceci indique une connexion refusee.")
+ tab = make_tab(ai_provider_factory=lambda: provider)
+ tab.terminal.feed(b"erreur : connection refused\r\n")
+ tab.terminal._sel_anchor, tab.terminal._sel_focus = (0, 0), (10, 0)
+ assert tab.ai_panel is None
+ with NOTICE_OK, SEND_ASIS:
+ tab._ai_explain()
+ pump()
+ assert isinstance(tab.ai_panel, AIPanel)
+ assert tab.splitter.indexOf(tab.ai_panel) >= 0
+ assert "connexion refusee" in tab.ai_panel.conversation.toPlainText().lower()
+ tab.close_connection()
+print(" OK")
+
+print()
+print("=== 4. Basculer l'affichage du panneau IA (menu) ===")
+with patch.object(LocalTerminalConnection, "start", lambda self: None):
+ tab = make_tab(ai_provider_factory=lambda: FakeProvider())
+ tab.show() # isVisible() se compose avec les ancetres : il en faut un reellement affiche
+ tab._toggle_ai_panel() # premiere fois : cree + affiche
+ assert tab.ai_panel is not None and tab.ai_panel.isVisible()
+ tab._toggle_ai_panel() # deuxieme fois : masque
+ assert not tab.ai_panel.isVisible()
+ tab.close_connection()
+print(" OK")
+
+print()
+print("=== 5. Generer une commande : jamais auto-execute, insertion sans Entree ===")
+with patch.object(LocalTerminalConnection, "start", lambda self: None):
+ provider = FakeProvider(reply="Pour lister les gros fichiers :\n\n```bash\nfind / -size +100M\n```")
+ tab = make_tab(ai_provider_factory=lambda: provider)
+ with PROMPTS_OK_IN_TABS, SEND_ASIS_IN_TABS, \
+ patch("app.ui.tabs.QInputDialog.getMultiLineText", return_value=("trouver les gros fichiers", True)), \
+ patch("app.ui.tabs.AICommandDialog.exec", lambda self: (setattr(self, "insert_requested", True), 1)[1]):
+ tab._ai_generate_command()
+ pump()
+ sent = tab.connection._out_queue.get_nowait()
+ print(" commande inseree:", sent)
+ assert sent == b"find / -size +100M"
+ assert not sent.endswith(b"\r") and not sent.endswith(b"\n")
+ assert tab.ai_btn.isEnabled() and tab.ai_btn.text() == "🤖 Assistant IA"
+ tab.close_connection()
+print(" OK")
+
+print()
+print("=== 6. Generer une commande : annulee par l'utilisateur -> rien envoye ===")
+with patch.object(LocalTerminalConnection, "start", lambda self: None):
+ provider = FakeProvider(reply="```bash\nrm -rf /\n```")
+ tab = make_tab(ai_provider_factory=lambda: provider)
+ with patch("app.ui.tabs.QInputDialog.getMultiLineText", return_value=("", False)):
+ tab._ai_generate_command()
+ assert tab.connection._out_queue.empty()
+ tab.close_connection()
+print(" OK")
+
+print()
+print("=== 7. Sans fournisseur configure : message clair, pas de boite de saisie ===")
+with patch.object(LocalTerminalConnection, "start", lambda self: None):
+ tab = make_tab(ai_provider_factory=lambda: None)
+ with patch("app.ui.tabs.QInputDialog.getMultiLineText") as mocked_input, \
+ patch("app.ui.tabs.QMessageBox.information") as mocked_info:
+ tab._ai_generate_command()
+ assert not mocked_input.called
+ assert mocked_info.called
+ tab.close_connection()
+print(" OK")
+
+print()
+print("=== 8. Fermeture de l'onglet : le panneau IA est arrete proprement ===")
+with patch.object(LocalTerminalConnection, "start", lambda self: None):
+ tab = make_tab(ai_provider_factory=lambda: FakeProvider())
+ tab._toggle_ai_panel()
+ tab.close_connection() # ne doit pas lever, ni bloquer indefiniment
+print(" OK")
+
+print()
+print("AI TAB INTEGRATION : TOUS LES TESTS OK")