pyssh-client/tests/test_snippets.py

"""Snippets : stockage (CRUD, recherche, categories) et integration terminal
(inserer sans executer, executer, executer avec Multi-Send).

LocalTerminalConnection.start() est neutralisee : send() ne fait que deposer
les octets dans une file interne (aucun sous-processus reel), ce qui suffit a
verifier ce qui a ete envoye sans avoir besoin d'un vrai terminal."""
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, QWidget

from app.snippets import Snippet, SnippetStore, DEFAULT_SNIPPETS
from app.local_terminal import LocalTerminalConnection
from app.controllers.broadcast_manager import BroadcastManager
from app.ui.tabs import LocalTerminalTabWidget

app = QApplication.instance() or QApplication([])

tmp = Path(tempfile.mkdtemp())

print("=== 1. Premier lancement : les exemples fournis sont charges ===")
store = SnippetStore(path=tmp / "snippets.json")
print(f"  {len(store.snippets)} snippet(s) au premier lancement")
assert len(store.snippets) == len(DEFAULT_SNIPPETS)
assert any(s.name == "Disk Usage" and s.category == "Linux" for s in store.snippets)
print("  OK")

print()
print("=== 2. CRUD ===")
store2 = SnippetStore(path=tmp / "snippets2.json")
initial_count = len(store2.snippets)
snippet = Snippet(name="Mon snippet", category="Perso", description="test", command="echo bonjour")
store2.add(snippet)
assert len(store2.snippets) == initial_count + 1
snippet.command = "echo modifie"
store2.update(snippet)
assert store2.get(snippet.id).command == "echo modifie"
store2.delete(snippet.id)
assert store2.get(snippet.id) is None
assert len(store2.snippets) == initial_count
print("  OK")

print()
print("=== 3. Categories et regroupement ===")
store3 = SnippetStore(path=tmp / "snippets3.json")
store3.snippets = [
    Snippet(name="b", category="Docker", command="x"),
    Snippet(name="a", category="Docker", command="y"),
    Snippet(name="z", category="Linux", command="z"),
]
grouped = store3.by_category()
print("  categories:", list(grouped.keys()))
assert list(grouped.keys()) == ["Docker", "Linux"]
assert [s.name for s in grouped["Docker"]] == ["a", "b"]  # tries par nom
print("  OK")

print()
print("=== 4. Recherche ===")
results = store3.search("docker")
assert {s.category for s in results} == {"Docker"}
results = store3.search("")
assert len(results) == len(store3.snippets)
print("  OK")

print()
print("=== 5. Persistance (relecture depuis le disque) ===")
path4 = tmp / "snippets4.json"
store4 = SnippetStore(path=path4)
store4.add(Snippet(name="persist", category="Test", command="echo test"))
reloaded = SnippetStore(path=path4)
assert any(s.name == "persist" for s in reloaded.snippets)
print("  OK")

print()
print("=== 6. Terminal : Inserer n'envoie jamais Entree ===")
with patch.object(LocalTerminalConnection, "start", lambda self: None):
    tab = LocalTerminalTabWidget()
    snippet = Snippet(name="test", command="echo bonjour")
    tab._insert_snippet(snippet)
    sent = tab.connection._out_queue.get_nowait()
    print("  envoye:", sent)
    assert sent == b"echo bonjour"
    assert not sent.endswith(b"\r") and not sent.endswith(b"\n")
    tab.close_connection()
print("  OK")

print()
print("=== 7. Terminal : Executer envoie la commande suivie d'Entree ===")
with patch.object(LocalTerminalConnection, "start", lambda self: None):
    tab = LocalTerminalTabWidget()
    tab._execute_snippet(Snippet(name="test", command="echo bonjour"))
    sent = tab.connection._out_queue.get_nowait()
    print("  envoye:", sent)
    assert sent == b"echo bonjour\r"
    tab.close_connection()
print("  OK")

print()
print("=== 8. Terminal : Executer avec Multi-Send atteint tous les terminaux ===")
with patch.object(LocalTerminalConnection, "start", lambda self: None):
    tabs = [LocalTerminalTabWidget() for _ in range(3)]
    broadcast = BroadcastManager(lambda: tabs)
    for t in tabs:
        broadcast.register(t)
        t.broadcast = broadcast  # comme le ferait ConnectionController.new_local_terminal()

    tabs[0]._execute_snippet(Snippet(name="test", command="uptime"), multi=True)
    received = [t.connection._out_queue.get_nowait() for t in tabs]
    print("  recu par chaque terminal:", received)
    assert all(r == b"uptime\r" for r in received), received
    for t in tabs:
        t.close_connection()
print("  OK")

print()
print("=== 9. Menu Snippets : hierarchie categorie > snippet > actions, et declenchement ===")
with patch.object(LocalTerminalConnection, "start", lambda self: None):
    tab2 = LocalTerminalTabWidget()
    tab2.snippet_store = SnippetStore(path=tmp / "snippets_menu.json")
    tab2.snippet_store.snippets = [Snippet(name="Ping", category="Reseau", command="ping 1.1.1.1")]
    tab2.broadcast = BroadcastManager(lambda: [tab2])

    menu = tab2._build_snippets_menu()  # ne fait qu'assembler le menu, ne l'affiche pas (pas de exec())
    categories = [a.text() for a in menu.actions() if a.menu() is not None]
    print("  categories dans le menu:", categories)
    assert "Reseau" in categories

    reseau_menu = next(a.menu() for a in menu.actions() if a.text() == "Reseau")
    ping_menu = next(a.menu() for a in reseau_menu.actions() if a.text() == "Ping")
    action_labels = [a.text() for a in ping_menu.actions()]
    print("  actions du snippet:", action_labels)
    assert action_labels == ["Inserer dans le terminal", "Executer", "Executer avec Multi-Send"]

    insert_action = next(a for a in ping_menu.actions() if a.text() == "Inserer dans le terminal")
    insert_action.trigger()
    sent = tab2.connection._out_queue.get_nowait()
    assert sent == b"ping 1.1.1.1", sent
    print("  declenchement 'Inserer' -> commande envoyee sans Entree : OK")

    tab2.close_connection()
print("  OK")

print()
print("SNIPPETS : TOUS LES TESTS OK")