pyssh-client/tests/test_sftp.py

"""Test de la logique SFTP avec un client SFTP simule (adosse a un dossier temporaire)."""
import sys, pathlib
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
import os
import posixpath
import shutil
import stat as statmod
import tempfile
import time
from pathlib import Path

from PySide6.QtWidgets import QApplication
from PySide6.QtCore import Qt

from app.sftp_panel import SFTPPanel, TransferJob, NAME_ROLE, ISDIR_ROLE, PARENT_ROLE


class FakeAttr:
    def __init__(self, path: Path):
        st = path.stat()
        self.filename = path.name
        self.st_mode = st.st_mode
        self.st_size = st.st_size
        self.st_mtime = st.st_mtime


class FakeSFTP:
    """Implemente juste ce que le panneau utilise, adosse a un vrai dossier local."""

    def __init__(self, root: Path):
        self.root = root

    def _real(self, path: str) -> Path:
        return self.root / path.lstrip("/")

    def normalize(self, path):
        return "/"

    def listdir_attr(self, path):
        return [FakeAttr(p) for p in sorted(self._real(path).iterdir())]

    def stat(self, path):
        return FakeAttr(self._real(path))

    def mkdir(self, path):
        self._real(path).mkdir()

    def rename(self, old, new):
        self._real(old).rename(self._real(new))

    def remove(self, path):
        self._real(path).unlink()

    def rmdir(self, path):
        self._real(path).rmdir()

    def chmod(self, path, mode):
        self._real(path).chmod(mode)

    def put(self, local, remote, callback=None):
        shutil.copyfile(local, self._real(remote))
        if callback:
            callback(os.path.getsize(local), os.path.getsize(local))

    def get(self, remote, local, callback=None):
        shutil.copyfile(self._real(remote), local)
        if callback:
            callback(os.path.getsize(local), os.path.getsize(local))


app = QApplication([])


def wait_idle(panel_side, timeout=5.0):
    """Pompe la boucle d'evenements Qt jusqu'a ce que toutes les operations SFTP
    soumises par ce panneau (navigate/refresh/etc, desormais asynchrones) soient
    revenues."""
    deadline = time.time() + timeout
    app.processEvents()
    while panel_side._callbacks and time.time() < deadline:
        app.processEvents()
        time.sleep(0.01)
    app.processEvents()


remote_root = Path(tempfile.mkdtemp())
local_root = Path(tempfile.mkdtemp())

# contenu distant simule
(remote_root / "dossier_distant").mkdir()
(remote_root / "config.yml").write_text("cle: valeur\n", encoding="utf-8")
(remote_root / "gros_fichier.log").write_text("x" * 5000, encoding="utf-8")
# contenu local
(local_root / "a_envoyer.txt").write_text("contenu local\n", encoding="utf-8")
(local_root / "dossier_local").mkdir()
(local_root / "dossier_local" / "interne.txt").write_text("imbrique\n", encoding="utf-8")

panel = SFTPPanel(FakeSFTP(remote_root))
panel.local.navigate(str(local_root))
panel.resize(1000, 600)
wait_idle(panel.remote)  # laisse le temps a la navigation initiale (asynchrone) d'arriver

print("=== 1. Listing distant avec colonnes ===")
tree = panel.remote.tree
for i in range(tree.topLevelItemCount()):
    it = tree.topLevelItem(i)
    print(f"  {it.text(0):<28} | {it.text(1):>9} | {it.text(2):<17} | {it.text(3)}")
assert tree.topLevelItemCount() == 4, "attendu .. + 3 entrees"
print("  -> colonnes Nom/Taille/Modifie/Droits presentes:", all(tree.headerItem().text(c) for c in range(4)))

print()
print("=== 2. Envoi par glisser-deposer (local -> distant) ===")
payload = {"side": "local", "entries": [{"name": "a_envoyer.txt", "is_dir": False}]}
panel._on_transfer_payload(payload)
deadline = time.time() + 5
while time.time() < deadline and panel._worker and panel._worker.isRunning():
    app.processEvents()
    time.sleep(0.02)
app.processEvents()
sent = (remote_root / "a_envoyer.txt")
print("  fichier arrive sur le serveur:", sent.exists())
assert sent.exists()

print()
print("=== 3. Envoi recursif d'un dossier ===")
panel._on_transfer_payload({"side": "local", "entries": [{"name": "dossier_local", "is_dir": True}]})
deadline = time.time() + 5
while time.time() < deadline and panel._worker and panel._worker.isRunning():
    app.processEvents()
    time.sleep(0.02)
app.processEvents()
nested = remote_root / "dossier_local" / "interne.txt"
print("  dossier + fichier imbrique transferes:", nested.exists())
assert nested.exists()

print()
print("=== 4. Reception (distant -> local) ===")
panel._on_transfer_payload({"side": "remote", "entries": [{"name": "config.yml", "is_dir": False}]})
deadline = time.time() + 5
while time.time() < deadline and panel._worker and panel._worker.isRunning():
    app.processEvents()
    time.sleep(0.02)
app.processEvents()
got = local_root / "config.yml"
print("  fichier recupere en local:", got.exists(), "| contenu identique:",
      got.read_text(encoding="utf-8") == "cle: valeur\n")
assert got.exists()

print()
print("=== 5. Renommage distant ===")
panel.remote.navigate("/")
wait_idle(panel.remote)
item = next(panel.remote.tree.topLevelItem(i) for i in range(panel.remote.tree.topLevelItemCount())
            if panel.remote.tree.topLevelItem(i).data(0, NAME_ROLE) == "config.yml")
item.setSelected(True)
panel.remote.sftp.rename("/config.yml", "/config_renomme.yml")
panel.remote.refresh()
wait_idle(panel.remote)
names = [panel.remote.tree.topLevelItem(i).data(0, NAME_ROLE)
         for i in range(panel.remote.tree.topLevelItemCount())]
print("  entrees apres renommage:", [n for n in names if n])
assert "config_renomme.yml" in names

print()
print("=== 6. Permissions (chmod) ===")
panel.remote.sftp.chmod("/config_renomme.yml", 0o600)
mode = statmod.S_IMODE((remote_root / "config_renomme.yml").stat().st_mode)
print(f"  droits appliques: {mode:03o}")

print()
print("=== 7. Suppression recursive distante ===")
panel.remote._rmdir_recursive("/dossier_local")
panel.remote.refresh()
wait_idle(panel.remote)
print("  dossier supprime du serveur:", not (remote_root / "dossier_local").exists())
assert not (remote_root / "dossier_local").exists()

print()
print("=== 8. Entree '..' presente pour remonter ===")
has_parent = any(panel.remote.tree.topLevelItem(i).data(0, PARENT_ROLE)
                 for i in range(panel.remote.tree.topLevelItemCount()))
print("  ligne '..' presente:", has_parent)
assert has_parent

panel.shutdown()

print()
print("TOUS LES TESTS SFTP OK")