From e88aab64d1b384a22d0fd29d0b4532b6554e5441 Mon Sep 17 00:00:00 2001 From: Jason Ross Date: Mon, 22 Sep 2025 15:23:20 -0500 Subject: [PATCH] added initial test suite and fixed pytest env problems --- poetry.lock | 25 +- pyproject.toml | 1 - tests/__init__.py | 1 + tests/conftest.py | 62 +++++ tests/core/test_adb_manager.py | 336 +++++++++++++++++++++++++ tests/managers/test_device_manager.py | 273 ++++++++++++++++++++ tests/test_basic.py | 20 ++ tests/utils/test_file_deduplication.py | 292 +++++++++++++++++++++ 8 files changed, 986 insertions(+), 24 deletions(-) create mode 100644 tests/__init__.py create mode 100644 tests/conftest.py create mode 100644 tests/core/test_adb_manager.py create mode 100644 tests/managers/test_device_manager.py create mode 100644 tests/test_basic.py create mode 100644 tests/utils/test_file_deduplication.py diff --git a/poetry.lock b/poetry.lock index d365585..ba45178 100644 --- a/poetry.lock +++ b/poetry.lock @@ -715,27 +715,6 @@ pytest = ">=6.2.5" [package.extras] dev = ["pre-commit", "pytest-asyncio", "tox"] -[[package]] -name = "pytest-qt" -version = "4.5.0" -description = "pytest support for PyQt and PySide applications" -optional = false -python-versions = ">=3.9" -groups = ["test"] -files = [ - {file = "pytest_qt-4.5.0-py3-none-any.whl", hash = "sha256:ed21ea9b861247f7d18090a26bfbda8fb51d7a8a7b6f776157426ff2ccf26eff"}, - {file = "pytest_qt-4.5.0.tar.gz", hash = "sha256:51620e01c488f065d2036425cbc1cbcf8a6972295105fd285321eb47e66a319f"}, -] - -[package.dependencies] -pluggy = ">=1.1" -pytest = "*" -typing_extensions = "*" - -[package.extras] -dev = ["pre-commit", "tox"] -doc = ["sphinx", "sphinx_rtd_theme"] - [[package]] name = "pywin32-ctypes" version = "0.2.3" @@ -861,7 +840,7 @@ version = "4.14.1" description = "Backported and Experimental Type Hints for Python 3.9+" optional = false python-versions = ">=3.9" -groups = ["dev", "test"] +groups = ["dev"] files = [ {file = "typing_extensions-4.14.1-py3-none-any.whl", hash = "sha256:d1e1e3b58374dc93031d6eda2420a48ea44a36c2b4766a4fdeb3710755731d76"}, {file = "typing_extensions-4.14.1.tar.gz", hash = "sha256:38b39f4aeeab64884ce9f74c94263ef78f3c22467c8724005483154c26648d36"}, @@ -909,4 +888,4 @@ test = ["covdefaults (>=2.3)", "coverage (>=7.2.7)", "coverage-enable-subprocess [metadata] lock-version = "2.1" python-versions = "<3.13,>=3.12" -content-hash = "2e5b4bc609ab9efe7fe2f83c7c475b1189133759463452d0ed664870cea7ed08" +content-hash = "a0025fb4372c9ab2d17f9f426c5862673c37628f8ba9425215152a0b7ecef738" diff --git a/pyproject.toml b/pyproject.toml index 325dbff..0d78b7b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,7 +32,6 @@ pre-commit = "^3.4.0" [tool.poetry.group.test.dependencies] pytest = "^7.4.0" -pytest-qt = "^4.2.0" pytest-mock = "^3.11.0" pytest-cov = "^4.1.0" diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..7b3aa58 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Test package initialization.""" \ No newline at end of file diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..28740db --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,62 @@ +"""Pytest configuration and fixtures.""" + +import pytest +import sys +import os + +# Add src directory to path for imports +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src')) + + +@pytest.fixture +def mock_tkinter_root(): + """Create a mock Tkinter root window for GUI tests.""" + import unittest.mock + return unittest.mock.MagicMock() + + +@pytest.fixture(autouse=True) +def mock_tkinter_imports(): + """Mock tkinter imports to avoid GUI dependencies in tests.""" + import unittest.mock + + # Mock tkinter modules + mock_tk = unittest.mock.MagicMock() + mock_messagebox = unittest.mock.MagicMock() + mock_filedialog = unittest.mock.MagicMock() + + modules_to_mock = { + 'tkinter': mock_tk, + 'tkinter.messagebox': mock_messagebox, + 'tkinter.filedialog': mock_filedialog, + 'tkinter.ttk': unittest.mock.MagicMock(), + } + + with unittest.mock.patch.dict('sys.modules', modules_to_mock): + yield + + +@pytest.fixture +def temp_directory(): + """Create a temporary directory for testing.""" + import tempfile + import shutil + + temp_dir = tempfile.mkdtemp() + yield temp_dir + shutil.rmtree(temp_dir) + + +@pytest.fixture +def temp_file(): + """Create a temporary file for testing.""" + import tempfile + import os + + fd, temp_path = tempfile.mkstemp() + os.close(fd) + yield temp_path + try: + os.unlink(temp_path) + except FileNotFoundError: + pass \ No newline at end of file diff --git a/tests/core/test_adb_manager.py b/tests/core/test_adb_manager.py new file mode 100644 index 0000000..9f33b6e --- /dev/null +++ b/tests/core/test_adb_manager.py @@ -0,0 +1,336 @@ +"""Tests for core ADB manager functionality.""" + +import pytest +import unittest.mock as mock +import os +import tempfile +import subprocess +import sys +from unittest.mock import MagicMock, patch, mock_open + +from src.core.adb_manager import ( + ADBManager, + get_executable_directory, + get_platform_tools_directory, + get_adb_binary_path, + is_adb_available, + get_platform_type, + ensure_platform_tools_in_user_dir +) + + +class TestADBManagerUtilityFunctions: + """Test utility functions in adb_manager module.""" + + def test_get_executable_directory_frozen(self): + """Test get_executable_directory when running as frozen executable.""" + with patch.object(sys, 'frozen', True, create=True): + with patch('sys.executable', '/path/to/executable'): + result = get_executable_directory() + assert result == '/path/to' + + def test_get_executable_directory_script(self): + """Test get_executable_directory when running as script.""" + with patch.object(sys, 'frozen', False, create=True): + with patch('os.path.abspath') as mock_abspath: + with patch('os.path.dirname') as mock_dirname: + mock_abspath.return_value = '/path/to/script.py' + mock_dirname.return_value = '/path/to' + result = get_executable_directory() + assert result == '/path/to' + + def test_get_platform_type(self): + """Test platform type detection.""" + # Test current platform (should be linux based on test output) + result = get_platform_type() + assert result in ['linux', 'windows', 'darwin'] + + # Test mocked platforms by patching the OS_TYPE variable + with patch('src.core.adb_manager.OS_TYPE', 'win32'): + result = get_platform_type() + assert result == 'win32' + + with patch('src.core.adb_manager.OS_TYPE', 'darwin'): + result = get_platform_type() + assert result == 'darwin' + + @patch('src.core.adb_manager.ensure_platform_tools_in_user_dir') + @patch('os.path.isfile') + def test_is_adb_available_system_path(self, mock_isfile, mock_ensure): + """Test ADB availability check when ADB binary is found.""" + # Mock ensure_platform_tools_in_user_dir to return a path + mock_ensure.return_value = '/path/to/adb' + mock_isfile.return_value = True # File exists + + result = is_adb_available() + assert result is True + mock_ensure.assert_called_once() + # isfile might be called multiple times (once inside ensure_platform_tools_in_user_dir) + assert mock_isfile.called + mock_isfile.assert_any_call('/path/to/adb') + + @patch('shutil.which') + @patch('os.path.isfile') + def test_is_adb_available_platform_tools(self, mock_isfile, mock_which): + """Test ADB availability check from platform-tools directory.""" + mock_which.return_value = None # Not in PATH + mock_isfile.return_value = True # But exists in platform-tools + assert is_adb_available() is True + + @patch('shutil.which') + @patch('os.path.isfile') + def test_is_adb_available_not_found(self, mock_isfile, mock_which): + """Test ADB availability check when ADB is not found.""" + mock_which.return_value = None + mock_isfile.return_value = False + assert is_adb_available() is False + + +class TestADBManager: + """Test ADBManager class functionality.""" + + @pytest.fixture + def adb_manager(self): + """Create an ADBManager instance for testing.""" + with patch('src.core.adb_manager.is_adb_available', return_value=True): + return ADBManager() + + def test_init(self, adb_manager): + """Test ADBManager initialization.""" + assert adb_manager.progress_callback is None + assert adb_manager.status_callback is None + assert adb_manager.current_process is None + assert adb_manager.deduplicator is not None + assert 'current_file' in adb_manager.transfer_progress + assert 'total_files' in adb_manager.transfer_progress + + def test_set_progress_callback(self, adb_manager): + """Test setting progress callback.""" + callback = MagicMock() + adb_manager.set_progress_callback(callback) + assert adb_manager.progress_callback == callback + + def test_set_status_callback(self, adb_manager): + """Test setting status callback.""" + callback = MagicMock() + adb_manager.set_status_callback(callback) + assert adb_manager.status_callback == callback + + def test_update_progress(self, adb_manager): + """Test progress update with callback.""" + callback = MagicMock() + adb_manager.set_progress_callback(callback) + adb_manager._update_progress(50) + callback.assert_called_once_with(50) + + def test_update_progress_no_callback(self, adb_manager): + """Test progress update without callback.""" + # Should not raise an exception + adb_manager._update_progress(50) + + def test_update_status(self, adb_manager): + """Test status update with callback.""" + callback = MagicMock() + adb_manager.set_status_callback(callback) + adb_manager._update_status("Test status") + callback.assert_called_once_with("Test status") + + def test_update_status_no_callback(self, adb_manager): + """Test status update without callback.""" + # Should not raise an exception + adb_manager._update_status("Test status") + + def test_update_transfer_progress(self, adb_manager): + """Test transfer progress calculation.""" + status_callback = MagicMock() + adb_manager.set_status_callback(status_callback) + + adb_manager._update_transfer_progress(3, 10) + + # Should update internal progress tracking + assert adb_manager.transfer_progress['current_file'] == 3 + assert adb_manager.transfer_progress['total_files'] == 10 + + # Should call status callback with special format + status_callback.assert_called_once_with("TRANSFER_PROGRESS:3:10") + + def test_reset_transfer_progress(self, adb_manager): + """Test transfer progress reset.""" + # Set some progress values + adb_manager.transfer_progress['current_file'] = 5 + adb_manager.transfer_progress['total_files'] = 10 + adb_manager.transfer_progress['files_to_transfer'] = 10 + + adb_manager._reset_transfer_progress() + + assert adb_manager.transfer_progress['current_file'] == 0 + assert adb_manager.transfer_progress['total_files'] == 0 + assert adb_manager.transfer_progress['files_to_transfer'] == 0 + + @patch('shutil.disk_usage') + @patch('src.core.adb_manager.get_platform_tools_directory') + def test_check_local_disk_space_sufficient(self, mock_get_dir, mock_disk_usage, adb_manager): + """Test disk space check with sufficient space.""" + mock_get_dir.return_value = '/fake/path' + mock_disk_usage.return_value = (1000*1024*1024, 500*1024*1024, 200*1024*1024) # total, used, free (200MB) + assert adb_manager.check_local_disk_space() is True + + @patch('shutil.disk_usage') + @patch('src.core.adb_manager.get_platform_tools_directory') + def test_check_local_disk_space_insufficient(self, mock_get_dir, mock_disk_usage, adb_manager): + """Test disk space check with insufficient space.""" + mock_get_dir.return_value = '/fake/path' + mock_disk_usage.return_value = (1000*1024*1024, 980*1024*1024, 20*1024*1024) # total, used, free (20MB < 50MB) + + with pytest.raises(Exception, match="Insufficient disk space"): + adb_manager.check_local_disk_space() + + @patch('src.core.adb_manager.get_adb_binary_path') + @patch('subprocess.run') + def test_run_adb_command_success(self, mock_run, mock_get_path, adb_manager): + """Test successful ADB command execution.""" + mock_get_path.return_value = '/fake/adb' + mock_result = MagicMock() + mock_result.stdout = "success output" + mock_result.stderr = "" + mock_result.returncode = 0 + mock_run.return_value = mock_result + + stdout, stderr, returncode = adb_manager.run_adb_command(['devices']) + + assert returncode == 0 + assert stdout == "success output" + assert stderr == "" + + @patch('src.core.adb_manager.get_adb_binary_path') + @patch('subprocess.run') + def test_run_adb_command_failure(self, mock_run, mock_get_path, adb_manager): + """Test failed ADB command execution.""" + mock_get_path.return_value = '/fake/adb' + mock_run.side_effect = Exception("Command failed") + + stdout, stderr, returncode = adb_manager.run_adb_command(['devices']) + + assert stdout is None + assert "Command failed" in stderr + assert returncode == -1 + + @patch('src.core.adb_manager.ADBManager.run_adb_command') + def test_check_device_connected(self, mock_run_command, adb_manager): + """Test device detection when device is connected.""" + mock_run_command.return_value = ("ABC123\tdevice\n", "", 0) + + device_id = adb_manager.check_device() + + assert device_id == "ABC123" + mock_run_command.assert_called_once_with(['devices'], capture_output=True) + + @patch('src.core.adb_manager.ADBManager.run_adb_command') + def test_check_device_not_connected(self, mock_run_command, adb_manager): + """Test device detection when no device is connected.""" + mock_run_command.return_value = ("List of devices attached\n\n", "", 0) + + device_id = adb_manager.check_device() + + assert device_id is None + + def test_parse_progress_valid(self, adb_manager): + """Test progress parsing with valid input.""" + test_line = "/sdcard/test.txt: (100%)" + result = adb_manager.parse_progress(test_line) + assert result == 100 + + test_line = "/sdcard/folder/file.jpg: (45%)" + result = adb_manager.parse_progress(test_line) + assert result == 45 + + # Test other patterns + test_line = "75% complete" + result = adb_manager.parse_progress(test_line) + assert result == 75 + + test_line = "transferred 50%" + result = adb_manager.parse_progress(test_line) + assert result == 50 + + def test_parse_progress_invalid(self, adb_manager): + """Test progress parsing with invalid input.""" + test_line = "Some random text" + result = adb_manager.parse_progress(test_line) + assert result is None + + test_line = "[ 50%] invalid format" + result = adb_manager.parse_progress(test_line) + assert result is None + + def test_cancel_transfer(self, adb_manager): + """Test transfer cancellation.""" + # Set up a mock process + mock_process = MagicMock() + adb_manager.current_process = mock_process + + result = adb_manager.cancel_transfer() + + assert result is True + mock_process.terminate.assert_called_once() + mock_process.wait.assert_called() + assert adb_manager.current_process is None + + +class TestEnsurePlatformToolsInUserDir: + """Test the ensure_platform_tools_in_user_dir function.""" + + @patch('src.core.adb_manager.get_platform_type') + @patch('requests.get') + @patch('tempfile.mkdtemp') + @patch('os.path.exists') + @patch('os.makedirs') + @patch('shutil.move') + @patch('os.symlink') + @patch('os.path.islink') + @patch('os.unlink') + def test_ensure_platform_tools_download_success( + self, mock_unlink, mock_islink, mock_symlink, mock_move, + mock_makedirs, mock_exists, mock_mkdtemp, mock_get, mock_platform_type + ): + """Test successful platform tools download and installation.""" + # Setup mocks + mock_platform_type.return_value = 'linux' + mock_exists.return_value = False + mock_mkdtemp.return_value = '/tmp/test' + mock_islink.return_value = False + + # Mock HTTP response + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.content = b'fake zip content' + mock_get.return_value = mock_response + + # Mock zipfile extraction + with patch('zipfile.ZipFile') as mock_zip: + mock_zip_instance = MagicMock() + mock_zip.return_value.__enter__.return_value = mock_zip_instance + + # Mock file operations + with patch('builtins.open', mock_open()) as mock_file: + with patch('os.listdir') as mock_listdir: + mock_listdir.return_value = ['platform-tools'] # Simulate extracted folder + with patch('os.path.isdir') as mock_isdir: + # Return True for the candidate directory check + mock_isdir.return_value = True + with patch('os.path.expanduser') as mock_expanduser: + mock_expanduser.return_value = '/home/user/.local/share/android-file-handler' + + + result = ensure_platform_tools_in_user_dir() + + # Verify the result is a path to adb + assert 'adb' in result + + # Verify download was attempted + mock_get.assert_called_once() + + # Verify extraction was attempted + mock_zip_instance.extractall.assert_called_once() +if __name__ == '__main__': + pytest.main([__file__]) \ No newline at end of file diff --git a/tests/managers/test_device_manager.py b/tests/managers/test_device_manager.py new file mode 100644 index 0000000..7593f53 --- /dev/null +++ b/tests/managers/test_device_manager.py @@ -0,0 +1,273 @@ +"""Tests for device manager functionality.""" + +import pytest +import tkinter as tk +from unittest.mock import MagicMock, patch, call + +from src.managers.device_manager import DeviceManager + + +class TestDeviceManager: + """Test DeviceManager class functionality.""" + + @pytest.fixture + def mock_parent_window(self): + """Create a mock parent window for testing.""" + return MagicMock(spec=tk.Tk) + + @pytest.fixture + def mock_status_callback(self): + """Create a mock status callback for testing.""" + return MagicMock() + + @pytest.fixture + def device_manager(self, mock_parent_window, mock_status_callback): + """Create a DeviceManager instance for testing.""" + with patch('src.managers.device_manager.ADBManager') as mock_adb: + manager = DeviceManager(mock_parent_window, mock_status_callback) + manager.adb_manager = mock_adb.return_value + return manager + + def test_init(self, mock_parent_window, mock_status_callback): + """Test DeviceManager initialization.""" + with patch('src.managers.device_manager.ADBManager') as mock_adb_class: + mock_adb_instance = MagicMock() + mock_adb_class.return_value = mock_adb_instance + + device_manager = DeviceManager(mock_parent_window, mock_status_callback) + + assert device_manager.parent == mock_parent_window + assert device_manager.status_callback == mock_status_callback + assert device_manager.device_connected is False + + # Verify ADB callbacks are set + mock_adb_instance.set_status_callback.assert_called_once() + mock_adb_instance.set_progress_callback.assert_called_once() + + def test_init_without_status_callback(self, mock_parent_window): + """Test DeviceManager initialization without status callback.""" + with patch('src.managers.device_manager.ADBManager'): + device_manager = DeviceManager(mock_parent_window) + assert device_manager.status_callback is None + + @patch('src.managers.device_manager.is_adb_available', return_value=True) + def test_initialize_adb_already_available(self, mock_is_available, device_manager): + """Test ADB initialization when ADB is already available.""" + result = device_manager.initialize_adb() + assert result is True + + @patch('src.managers.device_manager.is_adb_available') + @patch('src.managers.device_manager.messagebox.showinfo') + def test_initialize_adb_needs_download_success(self, mock_showinfo, mock_is_available, device_manager): + """Test ADB initialization when download is needed and succeeds.""" + # Mock is_adb_available to return False first (triggering download), then True (after download) + mock_is_available.side_effect = [False, True] + device_manager.adb_manager.download_and_extract_adb.return_value = True + + result = device_manager.initialize_adb() + + # Should show welcome message + mock_showinfo.assert_called_once() + assert "Welcome to Android File Transfer!" in mock_showinfo.call_args[0][0] + + # Should attempt download + device_manager.adb_manager.download_and_extract_adb.assert_called_once() + + assert result is True + + @patch('src.managers.device_manager.is_adb_available', return_value=False) + @patch('src.managers.device_manager.messagebox.showinfo') + @patch('src.managers.device_manager.messagebox.showerror') + def test_initialize_adb_needs_download_failure(self, mock_showerror, mock_showinfo, mock_is_available, device_manager): + """Test ADB initialization when download fails.""" + device_manager.adb_manager.download_and_extract_adb.return_value = False + + result = device_manager.initialize_adb() + + # Should show welcome message + mock_showinfo.assert_called_once() + + # Should attempt download + device_manager.adb_manager.download_and_extract_adb.assert_called_once() + + # Should show error message + mock_showerror.assert_called_once() + + assert result is False + + def test_check_device_connection_connected(self, device_manager): + """Test device connection check when device is connected.""" + device_manager.adb_manager.check_device.return_value = "ABC123" + + result = device_manager.check_device_connection() + + assert result == "ABC123" # Returns the device ID, not a boolean + assert device_manager.device_connected is True + device_manager.adb_manager.check_device.assert_called_once() + + def test_check_device_connection_not_connected(self, device_manager): + """Test device connection check when device is not connected.""" + device_manager.adb_manager.check_device.return_value = None + + result = device_manager.check_device_connection() + + assert result is None # Returns None, not False + assert device_manager.device_connected is False + device_manager.adb_manager.check_device.assert_called_once() + + def test_check_device_connection_exception(self, device_manager): + """Test device connection check when exception occurs.""" + device_manager.adb_manager.check_device.side_effect = Exception("Connection error") + + # The method doesn't catch exceptions, so it should raise + with pytest.raises(Exception, match="Connection error"): + device_manager.check_device_connection() + + def test_device_connected_property_after_connection(self, device_manager): + """Test device_connected property after successful connection.""" + device_manager.adb_manager.check_device.return_value = "TEST123" + + device_id = device_manager.check_device_connection() + + assert device_id == "TEST123" + assert device_manager.device_connected is True + + def test_device_connected_property_after_failed_connection(self, device_manager): + """Test device_connected property after failed connection.""" + device_manager.adb_manager.check_device.return_value = None + + device_id = device_manager.check_device_connection() + + assert device_id is None + assert device_manager.device_connected is False + + def test_on_adb_status_update_with_callback(self, device_manager): + """Test ADB status update with callback.""" + device_manager._on_adb_status_update("Test status") + + device_manager.status_callback.assert_called_once_with("Test status") + + def test_on_adb_status_update_without_callback(self, device_manager): + """Test ADB status update without callback.""" + device_manager.status_callback = None + + # Should not raise an exception + device_manager._on_adb_status_update("Test status") + + def test_on_adb_progress_update(self, device_manager): + """Test ADB progress update.""" + # This method currently just passes through, so we test it doesn't crash + device_manager._on_adb_progress_update(50) + # No assertions needed as the method doesn't do anything currently + + def test_is_remote_file_detects_file(self, device_manager): + """Test is_remote_file correctly identifies a file.""" + # Mock adb command to return file listing (starts with '-' for files) + device_manager.adb_manager.run_adb_command.return_value = ("-rw-r--r-- 1 root root 1234 test.txt", "", 0) + + result = device_manager.is_remote_file("/sdcard/test.txt") + assert result is True + + def test_is_remote_file_detects_directory(self, device_manager): + """Test is_remote_file correctly identifies a directory.""" + # Mock adb command to return directory listing (starts with 'd' for directories) + device_manager.adb_manager.run_adb_command.return_value = ("drwxr-xr-x 1 root root 4096 testdir", "", 0) + + result = device_manager.is_remote_file("/sdcard/testdir") + assert result is False + + def test_is_remote_file_command_fails(self, device_manager): + """Test is_remote_file when ADB command fails.""" + device_manager.adb_manager.run_adb_command.return_value = ("", "No such file", 1) + + result = device_manager.is_remote_file("/sdcard/nonexistent") + assert result is False + + def test_get_file_transfer_methods_push_file(self, device_manager): + """Test getting file transfer methods for pushing a file.""" + method_func, transfer_type = device_manager.get_file_transfer_methods("push", True) + assert transfer_type == "file" + assert method_func == device_manager.adb_manager.push_file + + def test_get_file_transfer_methods_pull_folder(self, device_manager): + """Test getting file transfer methods for pulling a folder.""" + method_func, transfer_type = device_manager.get_file_transfer_methods("pull", False) + assert transfer_type == "folder" + assert method_func == device_manager.adb_manager.pull_folder_with_dedup + + def test_get_file_transfer_methods_pull_file(self, device_manager): + """Test getting file transfer methods for pulling a file.""" + method_func, transfer_type = device_manager.get_file_transfer_methods("pull", True) + assert transfer_type == "file" + assert method_func == device_manager.adb_manager.pull_file + + def test_get_file_transfer_methods_push_folder(self, device_manager): + """Test getting file transfer methods for pushing a folder.""" + method_func, transfer_type = device_manager.get_file_transfer_methods("push", False) + assert transfer_type == "folder" + assert method_func == device_manager.adb_manager.push_folder_with_dedup + + def test_cancel_current_operation(self, device_manager): + """Test canceling current ADB operation.""" + # This method returns None, so just test that it calls the right method + device_manager.cancel_current_operation() + + device_manager.adb_manager.cancel_current_operation.assert_called_once() + + def test_device_connected_property_access(self, device_manager): + """Test device_connected property access.""" + # Test initial state + assert hasattr(device_manager, 'device_connected') + + # Test setting the property + device_manager.device_connected = True + assert device_manager.device_connected is True + + device_manager.device_connected = False + assert device_manager.device_connected is False + + def test_adb_manager_property_access(self, device_manager): + """Test accessing ADB manager property.""" + assert hasattr(device_manager, 'adb_manager') + assert device_manager.adb_manager is not None + + +class TestDeviceManagerIntegration: + """Integration tests for DeviceManager.""" + + def test_full_initialization_flow(self): + """Test the complete initialization flow.""" + mock_parent = MagicMock(spec=tk.Tk) + mock_callback = MagicMock() + + with patch('src.managers.device_manager.ADBManager') as mock_adb_class: + with patch('src.managers.device_manager.is_adb_available', return_value=True): + device_manager = DeviceManager(mock_parent, mock_callback) + result = device_manager.initialize_adb() + + assert result is True + assert device_manager.device_connected is False + + def test_device_connection_workflow(self): + """Test the device connection workflow.""" + mock_parent = MagicMock(spec=tk.Tk) + + with patch('src.managers.device_manager.ADBManager') as mock_adb_class: + mock_adb_instance = mock_adb_class.return_value + device_manager = DeviceManager(mock_parent) + + # Test connection success - should return device ID, not boolean + mock_adb_instance.check_device.return_value = "TEST123" + result = device_manager.check_device_connection() + assert result == "TEST123" # Returns device ID + assert device_manager.device_connected is True + + # Test connection failure - should return None, not False + mock_adb_instance.check_device.return_value = None + result = device_manager.check_device_connection() + assert result is None # Returns None + assert device_manager.device_connected is False + + +if __name__ == '__main__': + pytest.main([__file__]) \ No newline at end of file diff --git a/tests/test_basic.py b/tests/test_basic.py new file mode 100644 index 0000000..2118ce8 --- /dev/null +++ b/tests/test_basic.py @@ -0,0 +1,20 @@ +"""Simple test to verify pytest setup.""" + +def test_basic(): + """Basic test to verify pytest is working.""" + assert 1 + 1 == 2 + + +def test_imports(): + """Test that we can import our modules.""" + try: + from src.utils.file_deduplication import FileDeduplicator + deduplicator = FileDeduplicator() + assert deduplicator is not None + except ImportError as e: + pytest.fail(f"Failed to import FileDeduplicator: {e}") + + +if __name__ == '__main__': + import pytest + pytest.main([__file__]) \ No newline at end of file diff --git a/tests/utils/test_file_deduplication.py b/tests/utils/test_file_deduplication.py new file mode 100644 index 0000000..3363f2b --- /dev/null +++ b/tests/utils/test_file_deduplication.py @@ -0,0 +1,292 @@ +"""Tests for file deduplication utility.""" + +import pytest +import hashlib +import os +import tempfile +from unittest.mock import MagicMock, patch, mock_open + +from src.utils.file_deduplication import FileDeduplicator + + +class TestFileDeduplicator: + """Test FileDeduplicator class functionality.""" + + @pytest.fixture + def deduplicator(self): + """Create a FileDeduplicator instance for testing.""" + return FileDeduplicator() + + @pytest.fixture + def deduplicator_with_callbacks(self): + """Create a FileDeduplicator instance with callbacks for testing.""" + status_callback = MagicMock() + progress_callback = MagicMock() + return FileDeduplicator(status_callback, progress_callback), status_callback, progress_callback + + def test_init_without_callbacks(self, deduplicator): + """Test FileDeduplicator initialization without callbacks.""" + assert deduplicator.status_callback is None + assert deduplicator.progress_callback is None + + def test_init_with_callbacks(self): + """Test FileDeduplicator initialization with callbacks.""" + status_callback = MagicMock() + progress_callback = MagicMock() + deduplicator = FileDeduplicator(status_callback, progress_callback) + + assert deduplicator.status_callback == status_callback + assert deduplicator.progress_callback == progress_callback + + def test_update_status_with_callback(self, deduplicator_with_callbacks): + """Test status update with callback.""" + deduplicator, status_callback, _ = deduplicator_with_callbacks + deduplicator._update_status("Test message") + status_callback.assert_called_once_with("Test message") + + def test_update_status_without_callback(self, deduplicator): + """Test status update without callback.""" + # Should not raise an exception + deduplicator._update_status("Test message") + + def test_update_progress_with_callback(self, deduplicator_with_callbacks): + """Test progress update with callback.""" + deduplicator, _, progress_callback = deduplicator_with_callbacks + deduplicator._update_progress(75) + progress_callback.assert_called_once_with(75) + + def test_update_progress_without_callback(self, deduplicator): + """Test progress update without callback.""" + # Should not raise an exception + deduplicator._update_progress(75) + + @patch('builtins.open', new_callable=mock_open, read_data=b'test file content') + @patch('os.path.isfile', return_value=True) + def test_compute_local_file_hash_small_file(self, mock_isfile, mock_file, deduplicator): + """Test file hash computation for small file.""" + expected_hash = hashlib.sha256(b'test file content').hexdigest() + result = deduplicator.compute_local_file_hash('/fake/path/file.txt') + + assert result == expected_hash + mock_file.assert_called_once_with('/fake/path/file.txt', 'rb') + mock_isfile.assert_called_once_with('/fake/path/file.txt') + + @patch('builtins.open', new_callable=mock_open) + @patch('os.path.isfile', return_value=True) + def test_compute_local_file_hash_large_file(self, mock_isfile, mock_file, deduplicator_with_callbacks): + """Test file hash computation for large file with progress updates.""" + deduplicator, status_callback, progress_callback = deduplicator_with_callbacks + + # Mock reading chunks + mock_file.return_value.__enter__.return_value.read.side_effect = [ + b'chunk1' * 1000, # First chunk + b'chunk2' * 1000, # Second chunk + b'', # EOF + ] + + result = deduplicator.compute_local_file_hash('/fake/path/largefile.txt') + + # Should return a valid hash + assert isinstance(result, str) + assert len(result) == 64 # SHA256 hex digest length + + @patch('os.path.isfile', return_value=False) + def test_compute_local_file_hash_nonexistent_file(self, mock_isfile, deduplicator): + """Test file hash computation for nonexistent file.""" + result = deduplicator.compute_local_file_hash('/fake/nonexistent/file.txt') + assert result is None + mock_isfile.assert_called_once_with('/fake/nonexistent/file.txt') + + @patch('builtins.open', side_effect=PermissionError("Permission denied")) + @patch('os.path.isfile', return_value=True) + def test_compute_local_file_hash_permission_error(self, mock_isfile, mock_file, deduplicator): + """Test file hash computation with permission error.""" + result = deduplicator.compute_local_file_hash('/fake/protected/file.txt') + assert result is None + + def test_check_files_identical_same_hash(self, deduplicator): + """Test file comparison with identical hashes.""" + with patch.object(deduplicator, 'compute_local_file_hash') as mock_hash: + mock_hash.return_value = 'abc123' + with patch.object(deduplicator, 'compute_remote_file_hash') as mock_remote_hash: + mock_remote_hash.return_value = 'abc123' + result = deduplicator.check_files_identical('/file1.txt', '/file2.txt') + assert result is True + + def test_check_files_identical_different_hash(self, deduplicator): + """Test file comparison with different hashes.""" + with patch.object(deduplicator, 'compute_local_file_hash') as mock_hash: + mock_hash.return_value = 'abc123' + with patch.object(deduplicator, 'compute_remote_file_hash') as mock_remote_hash: + mock_remote_hash.return_value = 'def456' + result = deduplicator.check_files_identical('/file1.txt', '/file2.txt') + assert result is False + + def test_check_files_identical_one_hash_none(self, deduplicator): + """Test file comparison with one hash being None.""" + with patch.object(deduplicator, 'compute_local_file_hash') as mock_hash: + mock_hash.return_value = 'abc123' + with patch.object(deduplicator, 'compute_remote_file_hash') as mock_remote_hash: + mock_remote_hash.return_value = None + result = deduplicator.check_files_identical('/file1.txt', '/file2.txt') + assert result is False + + def test_check_files_identical_both_hashes_none(self, deduplicator): + """Test file comparison with both hashes being None.""" + with patch.object(deduplicator, 'compute_local_file_hash') as mock_hash: + mock_hash.return_value = None + with patch.object(deduplicator, 'compute_remote_file_hash') as mock_remote_hash: + mock_remote_hash.return_value = None + result = deduplicator.check_files_identical('/file1.txt', '/file2.txt') + assert result is False + + @patch('os.path.exists') + @patch('os.listdir') + @patch('os.path.isfile') + def test_build_local_file_hash_map(self, mock_isfile, mock_listdir, mock_exists, deduplicator): + """Test getting local file hashes from directory.""" + mock_exists.return_value = True + mock_listdir.return_value = ['file1.txt', 'file2.jpg', 'subdir'] + mock_isfile.side_effect = lambda x: not x.endswith('subdir') + + file_paths = ['/fake/directory/file1.txt', '/fake/directory/file2.jpg'] + + with patch.object(deduplicator, 'compute_local_file_hash') as mock_hash: + mock_hash.side_effect = ['hash1', 'hash2'] + + result = deduplicator.build_local_file_hash_map(file_paths) + + expected = { + '/fake/directory/file1.txt': 'hash1', + '/fake/directory/file2.jpg': 'hash2' + } + assert result == expected + + @patch('os.path.exists', return_value=False) + def test_build_local_file_hash_map_nonexistent_files(self, mock_exists, deduplicator): + """Test getting local file hashes from nonexistent files.""" + result = deduplicator.build_local_file_hash_map(['/fake/nonexistent']) + assert result == {} + + def test_find_duplicate_files_with_duplicates(self, deduplicator): + """Test finding duplicates when duplicates exist.""" + local_files = ['/local/file1.txt', '/local/file2.jpg', '/local/file3.txt'] + remote_files = ['/remote/remote1.txt', '/remote/remote2.jpg', '/remote/remote3.txt'] + + with patch.object(deduplicator, 'build_local_file_hash_map') as mock_local: + with patch.object(deduplicator, 'build_remote_file_hash_map') as mock_remote: + mock_local.return_value = { + '/local/file1.txt': 'hash1', + '/local/file2.jpg': 'hash2', + '/local/file3.txt': 'hash3' + } + mock_remote.return_value = { + '/remote/remote1.txt': 'hash1', # Duplicate of file1.txt + '/remote/remote2.jpg': 'hash4', # Unique + '/remote/remote3.txt': 'hash3' # Duplicate of file3.txt + } + + files_to_transfer, duplicates = deduplicator.find_duplicate_files( + local_files, remote_files, is_remote_target=True) + + # Should find 2 duplicates: file1.txt and file3.txt + assert len(duplicates) == 2 + assert '/local/file1.txt' in duplicates + assert '/local/file3.txt' in duplicates + # file2.jpg should be transferred since it's not a duplicate + assert len(files_to_transfer) == 1 + assert '/local/file2.jpg' in files_to_transfer + + def test_find_duplicate_files_no_duplicates(self, deduplicator): + """Test finding duplicates when no duplicates exist.""" + local_files = ['/local/file1.txt', '/local/file2.jpg'] + remote_files = ['/remote/remote1.txt', '/remote/remote2.jpg'] + + with patch.object(deduplicator, 'build_local_file_hash_map') as mock_local: + with patch.object(deduplicator, 'build_remote_file_hash_map') as mock_remote: + mock_local.return_value = { + '/local/file1.txt': 'hash1', + '/local/file2.jpg': 'hash2' + } + mock_remote.return_value = { + '/remote/remote1.txt': 'hash3', + '/remote/remote2.jpg': 'hash4' + } + + files_to_transfer, duplicates = deduplicator.find_duplicate_files( + local_files, remote_files, is_remote_target=True) + assert duplicates == [] + assert files_to_transfer == local_files + + def test_find_duplicate_files_empty_collections(self, deduplicator): + """Test finding duplicates with empty collections.""" + files_to_transfer, duplicates = deduplicator.find_duplicate_files([], []) + assert duplicates == [] + assert files_to_transfer == [] + + +class TestFileDeduplicatorIntegration: + """Integration tests for FileDeduplicator with real files.""" + + def test_real_file_hash_computation(self): + """Test hash computation with real temporary files.""" + deduplicator = FileDeduplicator() + + # Create temporary files with known content + with tempfile.NamedTemporaryFile(mode='w', delete=False) as tf1: + tf1.write("test content 1") + temp_file1 = tf1.name + + with tempfile.NamedTemporaryFile(mode='w', delete=False) as tf2: + tf2.write("test content 1") # Same content + temp_file2 = tf2.name + + with tempfile.NamedTemporaryFile(mode='w', delete=False) as tf3: + tf3.write("test content 2") # Different content + temp_file3 = tf3.name + + try: + hash1 = deduplicator.compute_local_file_hash(temp_file1) + hash2 = deduplicator.compute_local_file_hash(temp_file2) + hash3 = deduplicator.compute_local_file_hash(temp_file3) + + # Same content should have same hash + assert hash1 == hash2 + # Different content should have different hash + assert hash1 != hash3 + + # Test file comparison - since check_files_identical expects local/remote, + # we'll just compare hashes directly for local files + assert hash1 == hash2 # Same files should have same hash + assert hash1 != hash3 # Different files should have different hash + + finally: + # Clean up + os.unlink(temp_file1) + os.unlink(temp_file2) + os.unlink(temp_file3) + + def test_real_directory_hash_computation(self): + """Test directory hash computation with real temporary directory.""" + deduplicator = FileDeduplicator() + + # Create temporary directory with files + with tempfile.TemporaryDirectory() as temp_dir: + file1_path = os.path.join(temp_dir, 'file1.txt') + file2_path = os.path.join(temp_dir, 'file2.txt') + + with open(file1_path, 'w') as f1: + f1.write("content 1") + with open(file2_path, 'w') as f2: + f2.write("content 2") + + hashes = deduplicator.build_local_file_hash_map([file1_path, file2_path]) + + assert file1_path in hashes + assert file2_path in hashes + assert len(hashes) == 2 + assert hashes[file1_path] != hashes[file2_path] + + +if __name__ == '__main__': + pytest.main([__file__]) \ No newline at end of file