136 lines
4.0 KiB
Python
136 lines
4.0 KiB
Python
"""
|
|
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, " ")
|