diff --git a/src/portkeydrop/app.py b/src/portkeydrop/app.py index a7f2219..aa94877 100644 --- a/src/portkeydrop/app.py +++ b/src/portkeydrop/app.py @@ -14,8 +14,11 @@ from portkeydrop.dialogs.settings import SettingsDialog from portkeydrop.dialogs.site_manager import SiteManagerDialog from portkeydrop.dialogs.transfer import ( + TransferDirection, + TransferStatus, TransferManager, create_transfer_dialog, + get_transfer_event_binder, ) from portkeydrop.local_files import ( delete_local, @@ -75,6 +78,7 @@ def __init__(self) -> None: self._settings = load_settings() self._site_manager = SiteManager() self._transfer_manager = TransferManager(notify_window=self) + self._transfer_state_by_id: dict[int, str] = {} self._remote_filter_text = "" self._local_filter_text = "" self._local_cwd = resolve_startup_local_folder(self._settings) @@ -327,6 +331,7 @@ def _bind_events(self) -> None: self.Bind(wx.EVT_MENU, self._on_transfer_queue, id=ID_TRANSFER_QUEUE) self.Bind(wx.EVT_MENU, self._on_settings, id=ID_SETTINGS) self.Bind(wx.EVT_MENU, self._on_about, id=wx.ID_ABOUT) + self.Bind(get_transfer_event_binder(), self._on_transfer_update) # Toolbar connect button self.tb_connect_btn.Bind(wx.EVT_BUTTON, self._on_connect_toolbar) @@ -962,10 +967,12 @@ def _on_upload(self, event) -> None: return self._transfer_manager.add_recursive_upload(self._client, local_path, remote_path) self._announce(f"Uploading folder {filename}") + self._update_status(f"Uploading folder {filename}...", self._client.cwd) else: total = os.path.getsize(local_path) self._transfer_manager.add_upload(self._client, local_path, remote_path, total) self._announce(f"Uploading {filename}") + self._update_status(f"Uploading {filename}...", self._client.cwd) self._show_transfer_queue() def _get_clipboard_files(self) -> list[str]: @@ -1004,6 +1011,9 @@ def _paste_upload(self) -> None: count += 1 if count: self._announce(f"Uploading {count} item{'s' if count != 1 else ''} from clipboard") + self._update_status( + f"Uploading {count} item{'s' if count != 1 else ''}...", self._client.cwd + ) self._show_transfer_queue() def _paste_local(self) -> None: @@ -1060,13 +1070,16 @@ def _delete_remote(self) -> None: ) if result == wx.YES: try: + self._update_status(f"Deleting {f.name}...", self._client.cwd) if f.is_dir: self._client.rmdir(f.path) else: self._client.delete(f.path) self._announce(f"Deleted {f.name}") + self._update_status("Delete complete.", self._client.cwd) self._refresh_remote_files() except Exception as e: + self._update_status("Delete failed.", self._client.cwd) wx.MessageBox(f"Delete failed: {e}", "Error", wx.OK | wx.ICON_ERROR, self) def _delete_local(self) -> None: @@ -1102,10 +1115,13 @@ def _rename_remote(self) -> None: parent = str(PurePosixPath(f.path).parent) new_path = f"{parent.rstrip('/')}/{new_name}" try: + self._update_status(f"Renaming {f.name}...", self._client.cwd) self._client.rename(f.path, new_path) self._announce(f"Renamed to {new_name}") + self._update_status("Rename complete.", self._client.cwd) self._refresh_remote_files() except Exception as e: + self._update_status("Rename failed.", self._client.cwd) wx.MessageBox(f"Rename failed: {e}", "Error", wx.OK | wx.ICON_ERROR, self) dlg.Destroy() @@ -1142,10 +1158,13 @@ def _mkdir_remote(self) -> None: if name: path = f"{self._client.cwd.rstrip('/')}/{name}" try: + self._update_status(f"Creating directory {name}...", self._client.cwd) self._client.mkdir(path) self._announce(f"Created directory {name}") + self._update_status("Directory created.", self._client.cwd) self._refresh_remote_files() except Exception as e: + self._update_status("Create directory failed.", self._client.cwd) wx.MessageBox( f"Failed to create directory: {e}", "Error", wx.OK | wx.ICON_ERROR, self ) @@ -1182,7 +1201,30 @@ def _on_transfer_queue(self, event: wx.CommandEvent) -> None: self._show_transfer_queue() def _on_transfer_update(self, event) -> None: - pass + latest_status_message = None + for transfer in self._transfer_manager.transfers: + current_state = transfer.status.value + previous_state = self._transfer_state_by_id.get(transfer.id) + if current_state == previous_state: + continue + + self._transfer_state_by_id[transfer.id] = current_state + direction_label = ( + "Upload" if transfer.direction == TransferDirection.UPLOAD else "Download" + ) + + if transfer.status == TransferStatus.IN_PROGRESS: + latest_status_message = f"{direction_label} in progress..." + elif transfer.status == TransferStatus.COMPLETED: + latest_status_message = f"{direction_label} complete." + elif transfer.status == TransferStatus.FAILED: + latest_status_message = f"{direction_label} failed." + elif transfer.status == TransferStatus.CANCELLED: + latest_status_message = f"{direction_label} cancelled." + + if latest_status_message: + current_path = self._client.cwd if self._client and self._client.connected else "" + self._update_status(latest_status_message, current_path) def _on_settings(self, event: wx.CommandEvent) -> None: dlg = SettingsDialog(self, self._settings) diff --git a/src/portkeydrop/dialogs/transfer.py b/src/portkeydrop/dialogs/transfer.py index 2fead45..82bcc44 100644 --- a/src/portkeydrop/dialogs/transfer.py +++ b/src/portkeydrop/dialogs/transfer.py @@ -14,6 +14,9 @@ logger = logging.getLogger(__name__) +_TRANSFER_EVENT_BINDER = None +_TRANSFER_EVENT_TYPE = None + class TransferDirection(Enum): UPLOAD = "upload" @@ -318,7 +321,8 @@ def _notify(self) -> None: try: import wx - evt = wx.PyCommandEvent(wx.NewEventType(), -1) + _binder, evt_type = _get_wx_event_binder() + evt = wx.PyCommandEvent(evt_type, -1) wx.PostEvent(self._notify_window, evt) except Exception: pass @@ -326,10 +330,19 @@ def _notify(self) -> None: def _get_wx_event_binder(): """Lazy creation of wx event type and binder.""" + global _TRANSFER_EVENT_BINDER, _TRANSFER_EVENT_TYPE + if _TRANSFER_EVENT_BINDER is not None and _TRANSFER_EVENT_TYPE is not None: + return _TRANSFER_EVENT_BINDER, _TRANSFER_EVENT_TYPE import wx - evt_type = wx.NewEventType() - return wx.PyEventBinder(evt_type, 1), evt_type + _TRANSFER_EVENT_TYPE = wx.NewEventType() + _TRANSFER_EVENT_BINDER = wx.PyEventBinder(_TRANSFER_EVENT_TYPE, 1) + return _TRANSFER_EVENT_BINDER, _TRANSFER_EVENT_TYPE + + +def get_transfer_event_binder(): + binder, _event_type = _get_wx_event_binder() + return binder def create_transfer_dialog(parent, transfer_manager: TransferManager): diff --git a/src/portkeydrop/protocols.py b/src/portkeydrop/protocols.py index b09cadf..bdbcb12 100644 --- a/src/portkeydrop/protocols.py +++ b/src/portkeydrop/protocols.py @@ -209,6 +209,22 @@ def _ensure_connected(self) -> ftplib.FTP: raise ConnectionError("Not connected") return self._ftp + def _path_exists(self, path: str) -> bool: + ftp = self._ensure_connected() + try: + ftp.sendcmd(f"MLST {path}") + return True + except Exception: + return False + + def _is_directory(self, path: str) -> bool: + ftp = self._ensure_connected() + try: + response = ftp.sendcmd(f"MLST {path}") + return "type=dir" in response.lower() + except Exception: + return False + def list_dir(self, path: str = ".") -> list[RemoteFile]: ftp = self._ensure_connected() files: list[RemoteFile] = [] @@ -287,22 +303,36 @@ def read_callback(data: bytes) -> None: callback(transferred, total) ftp.storbinary(f"STOR {remote_path}", local_file, block_size, read_callback) + remote_size = ftp.size(remote_path) + if remote_size is None or remote_size != total: + raise RuntimeError( + f"Remote upload verification failed for {remote_path}: expected {total} bytes, " + f"got {remote_size if remote_size is not None else 'unknown'}." + ) def delete(self, path: str) -> None: ftp = self._ensure_connected() ftp.delete(path) + if self._path_exists(path): + raise RuntimeError(f"Remote delete verification failed for {path}.") def rmdir(self, path: str) -> None: ftp = self._ensure_connected() ftp.rmd(path) + if self._path_exists(path): + raise RuntimeError(f"Remote directory delete verification failed for {path}.") def mkdir(self, path: str) -> None: ftp = self._ensure_connected() ftp.mkd(path) + if not self._is_directory(path): + raise RuntimeError(f"Remote mkdir verification failed for {path}.") def rename(self, old_path: str, new_path: str) -> None: ftp = self._ensure_connected() ftp.rename(old_path, new_path) + if not self._path_exists(new_path): + raise RuntimeError(f"Remote rename verification failed for {new_path}.") def stat(self, path: str) -> RemoteFile: ftp = self._ensure_connected() @@ -602,22 +632,50 @@ def progress(transferred: int, total_bytes: int) -> None: callback(transferred, total_bytes) sftp.putfo(local_file, remote_path, file_size=total, callback=progress) + attr = sftp.stat(remote_path) + if (attr.st_size or 0) != total: + raise RuntimeError( + f"Remote upload verification failed for {remote_path}: expected {total} bytes, " + f"got {attr.st_size if attr.st_size is not None else 'unknown'}." + ) def delete(self, path: str) -> None: sftp = self._ensure_connected() sftp.remove(path) + try: + sftp.stat(path) + except FileNotFoundError: + return + except OSError as exc: + if getattr(exc, "errno", None) == 2: + return + raise + raise RuntimeError(f"Remote delete verification failed for {path}.") def rmdir(self, path: str) -> None: sftp = self._ensure_connected() sftp.rmdir(path) + try: + sftp.stat(path) + except FileNotFoundError: + return + except OSError as exc: + if getattr(exc, "errno", None) == 2: + return + raise + raise RuntimeError(f"Remote directory delete verification failed for {path}.") def mkdir(self, path: str) -> None: sftp = self._ensure_connected() sftp.mkdir(path) + attr = sftp.stat(path) + if not attr.st_mode or not stat.S_ISDIR(attr.st_mode): + raise RuntimeError(f"Remote mkdir verification failed for {path}.") def rename(self, old_path: str, new_path: str) -> None: sftp = self._ensure_connected() sftp.rename(old_path, new_path) + sftp.stat(new_path) def stat(self, path: str) -> RemoteFile: sftp = self._ensure_connected() diff --git a/tests/_wx_stub.py b/tests/_wx_stub.py new file mode 100644 index 0000000..69deb01 --- /dev/null +++ b/tests/_wx_stub.py @@ -0,0 +1,194 @@ +"""Helpers for injecting a fake wx module into tests.""" + +from __future__ import annotations + +import importlib +import sys +import types +from unittest.mock import MagicMock + + +class _FakeFrame: + def __init__(self, *args, **kwargs): + self._bindings: list[tuple] = [] + self.title = "" + self.status_bar = MagicMock(SetStatusText=MagicMock()) + + def Bind(self, *args, **kwargs) -> None: + self._bindings.append((args, kwargs)) + + def SetName(self, *args, **kwargs) -> None: + pass + + def SetSizer(self, *_args, **_kwargs) -> None: + pass + + def CreateStatusBar(self, *args, **kwargs): + status = MagicMock(SetStatusText=MagicMock()) + self.status_bar = status + return status + + def SetTitle(self, title: str) -> None: + self.title = title + + def Show(self) -> None: + pass + + def Raise(self) -> None: + pass + + def FindFocus(self): + return None + + def Focus(self) -> None: + pass + + def Select(self, *args, **kwargs) -> None: + pass + + +class _FakeApp(_FakeFrame): + def MainLoop(self) -> None: + pass + + +class _SimpleWidget(MagicMock): + def __init__(self, *args, **kwargs): + super().__init__(bind=MagicMock()) + self.Bind = MagicMock() + self.SetSizer = MagicMock() + self.InsertItem = MagicMock(return_value=0) + self.SetItem = MagicMock() + self.GetItemCount = MagicMock(return_value=0) + self.Select = MagicMock() + self.Focus = MagicMock() + self.SetValue = MagicMock() + self.GetValue = MagicMock() + + +def _create_fake_wx() -> tuple[types.ModuleType, types.ModuleType]: + fake_wx = types.ModuleType("wx") + counter = 0 + + def _new_id_ref(*_args, **_kwargs): + nonlocal counter + counter += 1 + return counter + + fake_wx.NewIdRef = _new_id_ref + fake_wx.Frame = _FakeFrame + fake_wx.App = _FakeApp + fake_wx.MenuBar = lambda *args, **kwargs: _SimpleWidget() + fake_wx.Menu = lambda *args, **kwargs: _SimpleWidget() + fake_wx.ToolBar = lambda *args, **kwargs: _SimpleWidget() + fake_wx.Panel = lambda *args, **kwargs: _SimpleWidget() + fake_wx.BoxSizer = lambda *args, **kwargs: _SimpleWidget() + fake_wx.StaticText = lambda *args, **kwargs: _SimpleWidget() + fake_wx.Choice = lambda *args, **kwargs: _SimpleWidget() + fake_wx.TextCtrl = lambda *args, **kwargs: _SimpleWidget() + fake_wx.Button = lambda *args, **kwargs: _SimpleWidget() + fake_wx.ListCtrl = lambda *args, **kwargs: _SimpleWidget() + fake_wx.Timer = lambda *args, **kwargs: _SimpleWidget() + fake_wx.FileDataObject = MagicMock() + + class _Clipboard: + @staticmethod + def Open() -> bool: + return False + + @staticmethod + def GetData(_data) -> bool: + return False + + @staticmethod + def Close() -> None: + pass + + fake_wx.TheClipboard = _Clipboard + + # Basic constants + fake_wx.ALIGN_CENTER_VERTICAL = 1 + fake_wx.LEFT = 2 + fake_wx.EXPAND = 4 + fake_wx.ALL = 8 + fake_wx.VERTICAL = 16 + fake_wx.HORIZONTAL = 32 + fake_wx.LC_REPORT = 64 + fake_wx.LC_SINGLE_SEL = 128 + fake_wx.TE_PASSWORD = 256 + fake_wx.TE_PROCESS_ENTER = 512 + fake_wx.WXK_BACK = 513 + fake_wx.WXK_DELETE = 514 + fake_wx.WXK_F2 = 515 + fake_wx.ID_OK = 100 + fake_wx.OK = 100 + fake_wx.YES = 101 + fake_wx.YES_NO = 102 + fake_wx.ICON_WARNING = 103 + fake_wx.ICON_ERROR = 104 + fake_wx.ID_EXIT = 200 + fake_wx.ID_ABOUT = 201 + + fake_wx.EVT_MENU = object() + fake_wx.EVT_BUTTON = object() + fake_wx.EVT_LIST_ITEM_ACTIVATED = object() + fake_wx.EVT_KEY_DOWN = object() + fake_wx.EVT_CONTEXT_MENU = object() + fake_wx.EVT_TEXT_ENTER = object() + fake_wx.EVT_TIMER = object() + fake_wx.EVT_CHAR_HOOK = object() + fake_wx.EVT_CLOSE = object() + fake_wx.FILECTRL_ACTIVATED = object() + + fake_wx.CallAfter = lambda callback, *args, **kwargs: callback(*args, **kwargs) + fake_wx.Yield = lambda *args, **kwargs: None + fake_wx.NotFound = -1 + fake_wx.NOT_FOUND = -1 + + def _new_event_type() -> str: + return f"event-{_new_id_ref()}" + + fake_wx.NewEventType = MagicMock(side_effect=_new_event_type) + fake_wx.PyEventBinder = MagicMock(side_effect=lambda event_type, flag: f"binder-{event_type}") + fake_wx.PyCommandEvent = MagicMock( + side_effect=lambda event_type, id: MagicMock(event_type=event_type, id=id) + ) + fake_wx.PostEvent = MagicMock() + fake_wx.MessageBox = MagicMock(return_value=fake_wx.OK) + + fake_wx.Dialog = _SimpleWidget + fake_wx.TextEntryDialog = lambda *args, **kwargs: _SimpleWidget() + + fake_adv = types.ModuleType("wx.adv") + + class _AboutDialogInfo: + def __init__(self) -> None: + self.name = "" + self.version = "" + self.description = "" + + def SetName(self, value: str) -> None: + self.name = value + + def SetVersion(self, value: str) -> None: + self.version = value + + def SetDescription(self, value: str) -> None: + self.description = value + + fake_adv.AboutDialogInfo = _AboutDialogInfo + fake_adv.AboutBox = lambda info: None + fake_wx.adv = fake_adv + + return fake_wx, fake_adv + + +def load_module_with_fake_wx( + module_name: str, monkeypatch +) -> tuple[types.ModuleType, types.ModuleType]: + fake_wx, fake_adv = _create_fake_wx() + monkeypatch.setitem(sys.modules, "wx", fake_wx) + monkeypatch.setitem(sys.modules, "wx.adv", fake_adv) + sys.modules.pop(module_name, None) + module = importlib.import_module(module_name) + return module, fake_wx diff --git a/tests/test_app.py b/tests/test_app.py new file mode 100644 index 0000000..ca90bfb --- /dev/null +++ b/tests/test_app.py @@ -0,0 +1,283 @@ +"""Tests covering MainFrame helpers around uploads, deletes, and transfer updates.""" + +from contextlib import ExitStack +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from tests._wx_stub import load_module_with_fake_wx + + +@pytest.fixture +def app_module(monkeypatch): + module, fake_wx = load_module_with_fake_wx("portkeydrop.app", monkeypatch) + return module, fake_wx + + +def _build_frame(module, tmp_path): + app, _ = module + display = SimpleNamespace( + show_hidden_files=True, + announce_file_count=False, + sort_by="name", + sort_ascending=True, + ) + settings = SimpleNamespace(display=display) + fake_manager = MagicMock(transfers=[]) + fake_site_manager = MagicMock() + + with ExitStack() as stack: + stack.enter_context(patch.object(app, "load_settings", return_value=settings)) + stack.enter_context( + patch.object(app, "resolve_startup_local_folder", return_value=str(tmp_path)) + ) + stack.enter_context(patch.object(app, "SiteManager", return_value=fake_site_manager)) + transfer_manager_patch = stack.enter_context(patch.object(app, "TransferManager")) + transfer_manager_patch.return_value = fake_manager + for method in ( + "_build_menu", + "_build_toolbar", + "_build_dual_pane", + "_build_status_bar", + "_bind_events", + "_update_title", + "_refresh_local_files", + "_persist_local_folder_setting", + ): + stack.enter_context(patch.object(app.MainFrame, method, lambda self: None)) + frame = app.MainFrame() + return frame, fake_manager, transfer_manager_patch + + +def _hydrate_frame(module): + app, _ = module + frame = object.__new__(app.MainFrame) + frame._announce = MagicMock() + frame._update_status = MagicMock() + frame._show_transfer_queue = MagicMock() + frame._refresh_remote_files = MagicMock() + frame._get_selected_local_file = MagicMock() + frame._get_selected_remote_file = MagicMock() + frame._transfer_manager = MagicMock() + frame.status_bar = MagicMock(SetStatusText=MagicMock()) + return frame + + +def test_main_frame_init_sets_transfer_state(tmp_path, app_module): + frame, _, transfer_manager_cls = _build_frame(app_module, tmp_path) + assert frame._transfer_state_by_id == {} + transfer_manager_cls.assert_called_once_with(notify_window=frame) + + +def test_bind_events_hooks_transfer_update(app_module): + app, _ = app_module + frame = object.__new__(app.MainFrame) + frame.Bind = MagicMock() + frame.tb_connect_btn = MagicMock(Bind=MagicMock()) + frame.tb_protocol = MagicMock(Bind=MagicMock()) + frame.remote_file_list = MagicMock(Bind=MagicMock()) + frame.local_file_list = MagicMock(Bind=MagicMock()) + frame.local_path_bar = MagicMock(Bind=MagicMock()) + frame.remote_path_bar = MagicMock(Bind=MagicMock()) + + binder = object() + with patch.object(app, "get_transfer_event_binder", return_value=binder): + frame._bind_events() + + assert any( + call.args[0] == binder and call.args[1] == frame._on_transfer_update + for call in frame.Bind.call_args_list + ) + + +def test_on_upload_directory_updates_status(app_module): + app, _ = app_module + frame = _hydrate_frame(app_module) + frame._client = MagicMock(connected=True, cwd="/remote") + selected = MagicMock() + selected.name = "docs" + selected.path = "/tmp/docs" + selected.is_dir = True + frame._get_selected_local_file.return_value = selected + frame._transfer_manager.add_recursive_upload = MagicMock() + + frame._on_upload(None) + + frame._transfer_manager.add_recursive_upload.assert_called_once() + frame._update_status.assert_called_with("Uploading folder docs...", "/remote") + + +def test_on_upload_file_reports_progress(app_module): + app, _ = app_module + frame = _hydrate_frame(app_module) + frame._client = MagicMock(connected=True, cwd="/remote") + selected = MagicMock(name="file.txt") + selected.is_dir = False + selected.name = "file.txt" + selected.path = "/tmp/file.txt" + frame._get_selected_local_file.return_value = selected + frame._transfer_manager.add_upload = MagicMock() + + with patch.object(app.os.path, "getsize", return_value=123): + frame._on_upload(None) + + frame._transfer_manager.add_upload.assert_called_once() + frame._update_status.assert_called_with("Uploading file.txt...", "/remote") + + +def test_paste_upload_shows_queue(tmp_path, app_module): + app, _ = app_module + frame = _hydrate_frame(app_module) + frame._client = MagicMock(connected=True, cwd="/remote") + frame._transfer_manager.add_upload = MagicMock() + file_path = tmp_path / "clip.txt" + file_path.write_text("clip") + frame._get_clipboard_files = MagicMock(return_value=[str(file_path)]) + + frame._paste_upload() + + frame._transfer_manager.add_upload.assert_called_once() + frame._update_status.assert_called() + frame._show_transfer_queue.assert_called_once() + + +def test_delete_remote_updates_status_on_success(app_module): + app, fake_wx = app_module + frame = _hydrate_frame(app_module) + frame._client = MagicMock(connected=True, cwd="/remote") + remote = MagicMock(name="doc.txt") + remote.name = "doc.txt" + remote.path = "/remote/doc.txt" + remote.is_dir = False + frame._get_selected_remote_file.return_value = remote + frame._client.delete = MagicMock() + frame._update_status.reset_mock() + fake_wx.MessageBox.return_value = fake_wx.YES + + frame._delete_remote() + + frame._update_status.assert_any_call("Deleting doc.txt...", "/remote") + frame._update_status.assert_any_call("Delete complete.", "/remote") + frame._refresh_remote_files.assert_called_once() + + +def test_delete_remote_reports_failure(app_module): + app, fake_wx = app_module + frame = _hydrate_frame(app_module) + frame._client = MagicMock(connected=True, cwd="/remote") + remote = MagicMock(name="doc.txt") + remote.name = "doc.txt" + remote.path = "/remote/doc.txt" + remote.is_dir = False + frame._get_selected_remote_file.return_value = remote + frame._client.delete.side_effect = RuntimeError("boom") + fake_wx.MessageBox.return_value = fake_wx.YES + + frame._delete_remote() + + frame._update_status.assert_any_call("Delete failed.", "/remote") + fake_wx.MessageBox.assert_called() + + +def test_rename_remote_updates_status(app_module): + app, fake_wx = app_module + frame = _hydrate_frame(app_module) + frame._client = MagicMock(connected=True, cwd="/remote") + remote = MagicMock(name="old.txt") + remote.name = "old.txt" + remote.path = "/remote/old.txt" + frame._get_selected_remote_file.return_value = remote + frame._client.rename = MagicMock() + dialog = MagicMock( + ShowModal=MagicMock(return_value=fake_wx.ID_OK), + GetValue=MagicMock(return_value="new.txt"), + Destroy=MagicMock(), + ) + with patch.object(fake_wx, "TextEntryDialog", return_value=dialog): + frame._rename_remote() + + frame._update_status.assert_any_call("Renaming old.txt...", "/remote") + frame._update_status.assert_any_call("Rename complete.", "/remote") + + +def test_rename_remote_handles_error(app_module): + app, fake_wx = app_module + frame = _hydrate_frame(app_module) + frame._client = MagicMock(connected=True, cwd="/remote") + remote = MagicMock(name="old.txt") + remote.name = "old.txt" + remote.path = "/remote/old.txt" + frame._get_selected_remote_file.return_value = remote + frame._client.rename.side_effect = RuntimeError("boom") + dialog = MagicMock( + ShowModal=MagicMock(return_value=fake_wx.ID_OK), + GetValue=MagicMock(return_value="new.txt"), + Destroy=MagicMock(), + ) + fake_wx.MessageBox.reset_mock() + + with patch.object(fake_wx, "TextEntryDialog", return_value=dialog): + frame._rename_remote() + + frame._update_status.assert_any_call("Rename failed.", "/remote") + fake_wx.MessageBox.assert_called() + + +def test_mkdir_remote_updates_status(app_module): + app, fake_wx = app_module + frame = _hydrate_frame(app_module) + frame._client = MagicMock(connected=True, cwd="/remote") + frame._client.mkdir = MagicMock() + dialog = MagicMock( + ShowModal=MagicMock(return_value=fake_wx.ID_OK), + GetValue=MagicMock(return_value="new-dir"), + Destroy=MagicMock(), + ) + + with patch.object(fake_wx, "TextEntryDialog", return_value=dialog): + frame._mkdir_remote() + + frame._update_status.assert_any_call("Creating directory new-dir...", "/remote") + frame._update_status.assert_any_call("Directory created.", "/remote") + + +def test_mkdir_remote_reports_error(app_module): + app, fake_wx = app_module + frame = _hydrate_frame(app_module) + frame._client = MagicMock(connected=True, cwd="/remote") + frame._client.mkdir.side_effect = RuntimeError("boom") + dialog = MagicMock( + ShowModal=MagicMock(return_value=fake_wx.ID_OK), + GetValue=MagicMock(return_value="new-dir"), + Destroy=MagicMock(), + ) + + with patch.object(fake_wx, "TextEntryDialog", return_value=dialog): + frame._mkdir_remote() + + frame._update_status.assert_any_call("Create directory failed.", "/remote") + fake_wx.MessageBox.assert_called() + + +def test_on_transfer_update_reports_latest_status(app_module): + import importlib + + transfer_module = importlib.import_module("portkeydrop.dialogs.transfer") + + app, _ = app_module + frame = _hydrate_frame(app_module) + frame._client = MagicMock(connected=True, cwd="/remote") + frame._transfer_manager = MagicMock() + upload = transfer_module.TransferItem( + id=1, direction=app.TransferDirection.UPLOAD, status=app.TransferStatus.IN_PROGRESS + ) + download = transfer_module.TransferItem( + id=2, direction=app.TransferDirection.DOWNLOAD, status=app.TransferStatus.COMPLETED + ) + frame._transfer_manager.transfers = [upload, download] + frame._transfer_state_by_id = {} + + frame._on_transfer_update(None) + + frame._update_status.assert_called_once_with("Download complete.", "/remote") diff --git a/tests/test_protocols.py b/tests/test_protocols.py index 9cf9570..dafbf50 100644 --- a/tests/test_protocols.py +++ b/tests/test_protocols.py @@ -224,6 +224,7 @@ def test_chdir(self, mock_ftp_class): def test_mkdir(self, mock_ftp_class): mock_ftp = MagicMock() mock_ftp.pwd.return_value = "/" + mock_ftp.sendcmd.return_value = "250-Listing\r\n type=dir; /new_dir\r\n250 End" mock_ftp_class.return_value = mock_ftp info = ConnectionInfo(protocol=Protocol.FTP, host="example.com") @@ -237,6 +238,7 @@ def test_mkdir(self, mock_ftp_class): def test_delete(self, mock_ftp_class): mock_ftp = MagicMock() mock_ftp.pwd.return_value = "/" + mock_ftp.sendcmd.side_effect = Exception("not found") mock_ftp_class.return_value = mock_ftp info = ConnectionInfo(protocol=Protocol.FTP, host="example.com") @@ -250,6 +252,7 @@ def test_delete(self, mock_ftp_class): def test_rename(self, mock_ftp_class): mock_ftp = MagicMock() mock_ftp.pwd.return_value = "/" + mock_ftp.sendcmd.return_value = "250-Listing\r\n type=file; /new.txt\r\n250 End" mock_ftp_class.return_value = mock_ftp info = ConnectionInfo(protocol=Protocol.FTP, host="example.com") @@ -259,6 +262,69 @@ def test_rename(self, mock_ftp_class): mock_ftp.rename.assert_called_with("/old.txt", "/new.txt") + @patch("ftplib.FTP") + def test_is_directory_returns_false_on_exception(self, mock_ftp_class): + mock_ftp = MagicMock() + mock_ftp.pwd.return_value = "/" + mock_ftp.sendcmd.side_effect = Exception("failure") + mock_ftp_class.return_value = mock_ftp + + info = ConnectionInfo(protocol=Protocol.FTP, host="example.com") + client = FTPClient(info) + client.connect() + + assert not client._is_directory("/remote") + + @patch("ftplib.FTP") + def test_upload_raises_when_remote_size_mismatch(self, mock_ftp_class): + mock_ftp = MagicMock() + mock_ftp.size.return_value = 5 + + def fake_storbinary(cmd, file_obj, block_size, callback): + callback(file_obj.read()) + + mock_ftp.storbinary.side_effect = fake_storbinary + mock_ftp_class.return_value = mock_ftp + + info = ConnectionInfo(protocol=Protocol.FTP, host="example.com") + client = FTPClient(info) + client.connect() + + import io + + with pytest.raises(RuntimeError, match="Remote upload verification failed"): + client.upload(io.BytesIO(b"data"), "/remote.bin") + + def test_delete_raises_when_verification_fails(self): + info = ConnectionInfo(protocol=Protocol.FTP, host="example.com") + client = FTPClient(info) + client._ftp = MagicMock() + client._connected = True + client._path_exists = MagicMock(return_value=True) + + with pytest.raises(RuntimeError, match="verification failed"): + client.delete("/file.txt") + + def test_rmdir_raises_when_verification_fails(self): + info = ConnectionInfo(protocol=Protocol.FTP, host="example.com") + client = FTPClient(info) + client._ftp = MagicMock() + client._connected = True + client._path_exists = MagicMock(return_value=True) + + with pytest.raises(RuntimeError, match="verification failed"): + client.rmdir("/dir") + + def test_mkdir_raises_when_verification_fails(self): + info = ConnectionInfo(protocol=Protocol.FTP, host="example.com") + client = FTPClient(info) + client._ftp = MagicMock() + client._connected = True + client._is_directory = MagicMock(return_value=False) + + with pytest.raises(RuntimeError, match="verification failed"): + client.mkdir("/dir") + class TestSFTPClient: def test_not_connected_initially(self): @@ -381,6 +447,7 @@ def test_list_dir_maps_file_attributes(self, mock_ssh_class): @patch("paramiko.SSHClient") def test_chdir_download_upload_and_file_ops(self, mock_ssh_class): import io + import stat as stat_mod mock_ssh = MagicMock() mock_sftp = MagicMock() @@ -412,6 +479,23 @@ def fake_putfo(_file, _path, file_size, callback): callback(4, file_size) mock_sftp.putfo.side_effect = fake_putfo + file_attr = MagicMock(st_mode=stat_mod.S_IFREG | 0o644, st_size=4) + dir_attr = MagicMock(st_mode=stat_mod.S_IFDIR | 0o755, st_size=0) + + def stat_side_effect(path: str): + if path == "/remote.txt": + return file_attr + if path == "/a": + raise FileNotFoundError(path) + if path == "/b": + if mock_sftp.rmdir.called: + raise FileNotFoundError(path) + return dir_attr + if path == "/new": + return file_attr + raise FileNotFoundError(path) + + mock_sftp.stat.side_effect = stat_side_effect client.upload( io.BytesIO(b"data"), "/remote.txt", callback=lambda t, n: upload_calls.append((t, n)) ) @@ -426,6 +510,20 @@ def fake_putfo(_file, _path, file_size, callback): mock_sftp.rmdir.assert_called_once_with("/b") mock_sftp.rename.assert_called_once_with("/old", "/new") + @patch("ftplib.FTP") + def test_ftp_rename_raises_when_target_not_found_after_rename(self, mock_ftp_class): + mock_ftp = MagicMock() + mock_ftp.pwd.return_value = "/" + mock_ftp.sendcmd.side_effect = Exception("550 not found") + mock_ftp_class.return_value = mock_ftp + + info = ConnectionInfo(protocol=Protocol.FTP, host="example.com") + client = FTPClient(info) + client.connect() + + with pytest.raises(RuntimeError, match="verification failed"): + client.rename("/old.txt", "/new.txt") + @patch("paramiko.SSHClient") @patch("os.path.exists", return_value=True) def test_connect_with_key_and_stat(self, _mock_exists, mock_ssh_class): @@ -459,6 +557,72 @@ def test_connect_with_key_and_stat(self, _mock_exists, mock_ssh_class): assert remote.size == 42 assert remote.is_dir is False + @patch("paramiko.SSHClient") + def test_upload_raises_when_remote_size_mismatch(self, mock_ssh_class): + import io + import stat as stat_mod + + mock_ssh = MagicMock() + mock_sftp = MagicMock() + mock_sftp.normalize.return_value = "/" + mock_ssh.open_sftp.return_value = mock_sftp + mock_ssh_class.return_value = mock_ssh + + mock_sftp.stat.return_value = MagicMock(st_mode=stat_mod.S_IFREG | 0o644, st_size=3) + + client = SFTPClient(ConnectionInfo(protocol=Protocol.SFTP, host="example.com")) + client.connect() + + with pytest.raises(RuntimeError, match="verification failed"): + client.upload(io.BytesIO(b"data"), "/remote.txt") + + @patch("paramiko.SSHClient") + def test_mkdir_raises_when_created_path_is_not_directory(self, mock_ssh_class): + import stat as stat_mod + + mock_ssh = MagicMock() + mock_sftp = MagicMock() + mock_sftp.normalize.return_value = "/" + mock_ssh.open_sftp.return_value = mock_sftp + mock_ssh_class.return_value = mock_ssh + mock_sftp.stat.return_value = MagicMock(st_mode=stat_mod.S_IFREG | 0o644, st_size=10) + + client = SFTPClient(ConnectionInfo(protocol=Protocol.SFTP, host="example.com")) + client.connect() + + with pytest.raises(RuntimeError, match="verification failed"): + client.mkdir("/not-a-dir") + + @patch("paramiko.SSHClient") + def test_delete_raises_when_remote_stat_succeeds(self, mock_ssh_class): + mock_ssh = MagicMock() + mock_sftp = MagicMock() + mock_sftp.normalize.return_value = "/" + mock_ssh.open_sftp.return_value = mock_sftp + mock_ssh_class.return_value = mock_ssh + mock_sftp.stat.return_value = MagicMock() + + client = SFTPClient(ConnectionInfo(protocol=Protocol.SFTP, host="example.com")) + client.connect() + + with pytest.raises(RuntimeError, match="verification failed"): + client.delete("/file") + + @patch("paramiko.SSHClient") + def test_rmdir_raises_when_remote_stat_succeeds(self, mock_ssh_class): + mock_ssh = MagicMock() + mock_sftp = MagicMock() + mock_sftp.normalize.return_value = "/" + mock_ssh.open_sftp.return_value = mock_sftp + mock_ssh_class.return_value = mock_ssh + mock_sftp.stat.return_value = MagicMock() + + client = SFTPClient(ConnectionInfo(protocol=Protocol.SFTP, host="example.com")) + client.connect() + + with pytest.raises(RuntimeError, match="verification failed"): + client.rmdir("/dir") + class TestParentDir: def test_parent_from_subdir(self): diff --git a/tests/test_transfer_dialog.py b/tests/test_transfer_dialog.py new file mode 100644 index 0000000..9e84973 --- /dev/null +++ b/tests/test_transfer_dialog.py @@ -0,0 +1,48 @@ +"""Tests for transfer dialog event binder helpers.""" + +from unittest.mock import MagicMock + +import pytest + +from tests._wx_stub import load_module_with_fake_wx + + +@pytest.fixture +def transfer_module(monkeypatch): + module, fake_wx = load_module_with_fake_wx("portkeydrop.dialogs.transfer", monkeypatch) + return module, fake_wx + + +def test_notify_posts_event(transfer_module): + module, fake_wx = transfer_module + fake_wx.PostEvent.reset_mock() + module._TRANSFER_EVENT_BINDER = None + module._TRANSFER_EVENT_TYPE = None + window = MagicMock() + manager = module.TransferManager(notify_window=window) + + manager._notify() + + fake_wx.PostEvent.assert_called_once() + + +def test_wx_event_binder_is_cached(transfer_module): + module, fake_wx = transfer_module + module._TRANSFER_EVENT_BINDER = None + module._TRANSFER_EVENT_TYPE = None + binder1, event_type1 = module._get_wx_event_binder() + binder2, event_type2 = module._get_wx_event_binder() + + assert binder1 == binder2 + assert event_type1 == event_type2 + assert fake_wx.NewEventType.call_count == 1 + + +def test_get_transfer_event_binder_returns_shared_value(transfer_module): + module, _ = transfer_module + module._TRANSFER_EVENT_BINDER = None + module._TRANSFER_EVENT_TYPE = None + + binder = module.get_transfer_event_binder() + + assert binder is module._TRANSFER_EVENT_BINDER