pyssh-client/tests/test_keepalive.py

"""Keepalive de SSHConnection : verifie l'envoi periodique et l'abandon apres
`keepalive_count_max` echecs consecutifs, avec un transport paramiko mocke
(aucune connexion reseau reelle)."""
import pathlib
import sys

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

import threading
import time
from unittest.mock import MagicMock

from PySide6.QtWidgets import QApplication

from app.ssh_connection import SSHConnection

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


def make_connection(interval=0.05, count_max=3):
    conn = SSHConnection(
        host="10.0.0.1", port=22, username="root", auth_type="auto",
        keepalive_enabled=True, keepalive_interval=1, keepalive_count_max=count_max,
    )
    # keepalive_interval est borne a >= 1 dans __init__ (secondes reelles) : pour un
    # test rapide, on contourne ensuite l'attribut directement (le code ne relit
    # l'intervalle qu'a chaque iteration, une valeur fractionnaire est acceptee par
    # threading.Event.wait()).
    conn.keepalive_interval = interval
    return conn


print("=== 1. Requetes reussies : le compteur d'echecs ne progresse jamais ===")
conn = make_connection()
transport = MagicMock()
transport.is_active.return_value = True
conn.client = MagicMock()
conn.client.get_transport.return_value = transport

thread = threading.Thread(target=conn._keepalive_loop, daemon=True)
thread.start()
time.sleep(0.35)
conn._stop.set()
thread.join(timeout=2)

calls = transport.global_request.call_count
print(f"  requetes envoyees en ~0.35s (intervalle 0.05s): {calls}")
assert calls >= 3, calls
assert conn._disconnect_reason == "Connexion fermee"  # jamais modifie : aucun echec
print("  OK")

print()
print("=== 2. N echecs consecutifs -> abandon, canal ferme, raison mise a jour ===")
conn2 = make_connection(count_max=3)
transport2 = MagicMock()
transport2.is_active.return_value = True
transport2.global_request.side_effect = OSError("reseau injoignable")
conn2.client = MagicMock()
conn2.client.get_transport.return_value = transport2
fake_channel = MagicMock()
conn2.channel = fake_channel

thread2 = threading.Thread(target=conn2._keepalive_loop, daemon=True)
thread2.start()
thread2.join(timeout=3)

print(f"  requetes tentees avant abandon: {transport2.global_request.call_count}")
assert transport2.global_request.call_count == 3, transport2.global_request.call_count
assert conn2._stop.is_set(), "la connexion doit etre marquee pour arret"
assert fake_channel.close.called, "le canal doit etre ferme pour reveiller la boucle de lecture"
print("  raison de deconnexion:", conn2._disconnect_reason)
assert "keepalive" in conn2._disconnect_reason.lower()
print("  OK")

print()
print("=== 3. Une reponse reussie apres des echecs repart de zero ===")
conn3 = make_connection(count_max=3)
transport3 = MagicMock()
transport3.is_active.return_value = True
_calls = {"n": 0}


def _flaky_then_ok(*_a, **_k):
    _calls["n"] += 1
    if _calls["n"] <= 2:
        raise OSError("timeout")
    return None


transport3.global_request.side_effect = _flaky_then_ok
conn3.client = MagicMock()
conn3.client.get_transport.return_value = transport3

thread3 = threading.Thread(target=conn3._keepalive_loop, daemon=True)
thread3.start()
time.sleep(0.35)
conn3._stop.set()
thread3.join(timeout=2)

print(f"  requetes envoyees: {transport3.global_request.call_count} (jamais 3 echecs consecutifs)")
# la connexion ne doit pas avoir ete abandonnee malgre 2 echecs, puisqu'un succes
# est intervenu avant le 3e (le compteur d'echecs consecutifs est remis a zero) :
assert conn3._disconnect_reason == "Connexion fermee"
print("  OK")

print()
print("=== 4. Keepalive desactive : aucun thread demarre ===")
conn4 = SSHConnection(host="10.0.0.1", port=22, username="root", auth_type="auto", keepalive_enabled=False)
conn4._start_keepalive()
assert conn4._keepalive_thread is None
print("  OK")

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