""" 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 os 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 FakeStat: def __init__(self, st_size): self.st_size = st_size class FakeSFTP: def __init__(self, store): self.store = store def put(self, local, remote, confirm=True): self.store["put"] = (local, remote) self.store["confirm"] = confirm # Registra el tamaño para que stat() (verificación de subida) lo confirme. self.store.setdefault("sizes", {})[remote] = os.path.getsize(local) def stat(self, remote): sizes = self.store.get("sizes", {}) if remote not in sizes: raise FileNotFoundError(remote) return FakeStat(sizes[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, " ") def test_upload_usa_confirm_false(tmp_path, monkeypatch): """La subida no debe delegar la verificación al confirm inmediato de paramiko.""" zf = tmp_path / "backup.zip" zf.write_bytes(b"zipdata") store: dict = {} monkeypatch.setattr(sftp_copy, "paramiko", _fake_paramiko(store)) sftp_copy.upload_file_to_folder(str(zf), CFG, "D:\\In") assert store["confirm"] is False def test_verify_remote_size_reintenta_stat_flaky(monkeypatch): """Un stat transitoriamente fallido se reintenta y NO produce falso fallo.""" monkeypatch.setattr(sftp_copy, "VERIFY_DELAY_SECONDS", 0) calls = {"n": 0} class Flaky: def stat(self, remote): calls["n"] += 1 if calls["n"] < 2: raise OSError("stat flaky") return FakeStat(100) sftp_copy._verify_remote_size(Flaky(), "C:/In/x.zip", 100) # no debe lanzar assert calls["n"] == 2 def test_verify_remote_size_tamano_incorrecto_falla(monkeypatch): """Si el tamaño remoto nunca coincide, es un fallo genuino de entrega.""" monkeypatch.setattr(sftp_copy, "VERIFY_DELAY_SECONDS", 0) class Wrong: def stat(self, remote): return FakeStat(50) with pytest.raises(SFTPCopyError, match="verificar"): sftp_copy._verify_remote_size(Wrong(), "C:/In/x.zip", 100) def test_upload_zip_parts_adjunta_uploaded_en_fallo(tmp_path, monkeypatch): """Ante un fallo parcial, la excepción lleva las partes ya subidas para limpieza.""" p1 = tmp_path / "big.zip.001" p2 = tmp_path / "big.zip.002" p1.write_bytes(b"a") p2.write_bytes(b"b") calls = {"n": 0} def fake_upload(local, cfg, remote_folder): calls["n"] += 1 if calls["n"] == 1: return "D:/In/big.zip.001" raise SFTPCopyError("boom en la parte 2") monkeypatch.setattr(sftp_copy, "upload_file_to_folder", fake_upload) with pytest.raises(SFTPCopyError) as exc_info: sftp_copy.upload_zip_parts([str(p1), str(p2)], CFG, "D:\\In") assert getattr(exc_info.value, "uploaded", None) == ["D:/In/big.zip.001"]