feature/integracion-panel-restore-targets

tests/test_sftp_copy.py

@@ -0,0 +1,135 @@
+"""
+Pruebas de la transferencia SFTP al servidor remoto. Se mockea paramiko para no
+requerir un servidor SSH real; se valida la conversión de rutas y el flujo de subida.
+"""
+import pytest
+
+from app.transfer import sftp_copy
+from app.transfer.sftp_copy import SFTPCopyError
+
+CFG = {
+    "ssh_host": "192.168.1.100",
+    "ssh_port": 22,
+    "ssh_username": "Administrator",
+    "ssh_password": "ssh-secreto",
+    "remote_inbox_path": "C:\\RestoreInbox",
+}
+
+
+def test_windows_restore_path():
+    assert sftp_copy.windows_restore_path("C:\\RestoreInbox", "empresa.bak") == "C:\\RestoreInbox\\empresa.bak"
+
+
+def test_sftp_path_convierte_windows_a_posix():
+    # ruta destino en formato Windows → SFTP usa forward slashes
+    assert sftp_copy._sftp_path("C:\\RestoreInbox", "empresa.bak") == "C:/RestoreInbox/empresa.bak"
+    # ya en posix se respeta (sin doble slash)
+    assert sftp_copy._sftp_path("/srv/inbox/", "x.bak") == "/srv/inbox/x.bak"
+
+
+def test_upload_origen_inexistente(tmp_path):
+    with pytest.raises(SFTPCopyError, match="no existe"):
+        sftp_copy.upload_to_remote(str(tmp_path / "noexiste.bak"), CFG)
+
+
+class FakeSFTP:
+    def __init__(self, store):
+        self.store = store
+
+    def put(self, local, remote):
+        self.store["put"] = (local, remote)
+
+    def remove(self, remote):
+        self.store["removed"] = remote
+
+    def close(self):
+        self.store["sftp_closed"] = True
+
+
+class FakeClient:
+    def __init__(self, store):
+        self.store = store
+
+    def set_missing_host_key_policy(self, policy):
+        self.store["policy_set"] = True
+
+    def connect(self, **kwargs):
+        self.store["connect"] = kwargs
+
+    def open_sftp(self):
+        return FakeSFTP(self.store)
+
+    def close(self):
+        self.store["client_closed"] = True
+
+
+def _fake_paramiko(store):
+    class FakeParamiko:
+        AutoAddPolicy = object
+        SSHClient = lambda self=None: FakeClient(store)
+    fp = FakeParamiko()
+    # SSHClient() debe devolver FakeClient
+    fp.SSHClient = lambda: FakeClient(store)
+    return fp
+
+
+def test_upload_ok(tmp_path, monkeypatch):
+    bak = tmp_path / "empresa.bak"
+    bak.write_bytes(b"data")
+    store: dict = {}
+    monkeypatch.setattr(sftp_copy, "paramiko", _fake_paramiko(store))
+
+    remote = sftp_copy.upload_to_remote(str(bak), CFG)
+    assert remote == "C:/RestoreInbox/empresa.bak"
+    assert store["put"][1] == "C:/RestoreInbox/empresa.bak"
+    assert store["connect"]["hostname"] == "192.168.1.100"
+    assert store["connect"]["port"] == 22
+    assert store["client_closed"] is True
+
+
+def test_cleanup_remote_borra(monkeypatch):
+    store: dict = {}
+    monkeypatch.setattr(sftp_copy, "paramiko", _fake_paramiko(store))
+    sftp_copy.cleanup_remote(CFG, "C:/RestoreInbox/empresa.bak")
+    assert store["removed"] == "C:/RestoreInbox/empresa.bak"
+    assert store["client_closed"] is True
+
+
+def test_cleanup_remote_none_no_falla():
+    sftp_copy.cleanup_remote(CFG, None)  # no debe conectar ni lanzar
+
+
+def test_upload_file_to_folder_ok(tmp_path, monkeypatch):
+    zf = tmp_path / "backup.zip"
+    zf.write_bytes(b"zipdata")
+    store: dict = {}
+    monkeypatch.setattr(sftp_copy, "paramiko", _fake_paramiko(store))
+
+    remote = sftp_copy.upload_file_to_folder(
+        str(zf), CFG, "D:\\Restore\\Alfa\\Entrada"
+    )
+    assert remote == "D:/Restore/Alfa/Entrada/backup.zip"
+    assert store["put"][1] == "D:/Restore/Alfa/Entrada/backup.zip"
+
+
+def test_upload_zip_parts_multipart(tmp_path, monkeypatch):
+    p1 = tmp_path / "big.zip.001"
+    p2 = tmp_path / "big.zip.002"
+    p1.write_bytes(b"a")
+    p2.write_bytes(b"b")
+    store: dict = {}
+    monkeypatch.setattr(sftp_copy, "paramiko", _fake_paramiko(store))
+
+    uploaded = sftp_copy.upload_zip_parts(
+        [str(p1), str(p2)], CFG, "D:\\In"
+    )
+    assert len(uploaded) == 2
+    assert uploaded[0].endswith("/big.zip.001")
+    assert uploaded[1].endswith("/big.zip.002")
+
+
+def test_upload_file_to_folder_vacio_falla(tmp_path):
+    f = tmp_path / "x.zip"
+    f.write_bytes(b"x")
+    with pytest.raises(SFTPCopyError, match="carpeta remota"):
+        sftp_copy.upload_file_to_folder(str(f), CFG, "  ")