modularized test suite to match modular ADB manager

This commit is contained in:
2025-09-23 06:01:53 +08:00
committed by Jason Ross
parent 4e6688bfef
commit 835c041fc6
13 changed files with 2565 additions and 1276 deletions
+188
View File
@@ -0,0 +1,188 @@
"""Tests for ADB command execution module."""
import pytest
import unittest.mock as mock
import subprocess
from unittest.mock import patch, MagicMock
from src.core.adb_command import ADBCommandRunner
class TestADBCommandRunner:
"""Test ADB command runner functionality."""
def test_init(self):
"""Test ADBCommandRunner initialization."""
runner = ADBCommandRunner()
assert runner.current_process is None
@patch('src.core.adb_command.get_adb_binary_path')
@patch('subprocess.run')
def test_run_adb_command_success(self, mock_subprocess, mock_get_path):
"""Test successful ADB command execution."""
mock_get_path.return_value = '/path/to/adb'
mock_result = MagicMock()
mock_result.returncode = 0
mock_result.stdout = "device_list"
mock_result.stderr = ""
mock_subprocess.return_value = mock_result
runner = ADBCommandRunner()
stdout, stderr, returncode = runner.run_adb_command(['devices'])
assert returncode == 0
assert stdout == "device_list"
assert stderr == ""
mock_subprocess.assert_called_once_with(
['/path/to/adb', 'devices'],
capture_output=True,
text=True,
timeout=15
)
@patch('src.core.adb_command.get_adb_binary_path')
@patch('subprocess.run')
def test_run_adb_command_failure(self, mock_subprocess, mock_get_path):
"""Test ADB command execution failure."""
mock_get_path.return_value = '/path/to/adb'
mock_result = MagicMock()
mock_result.returncode = 1
mock_result.stdout = ""
mock_result.stderr = "error message"
mock_subprocess.return_value = mock_result
runner = ADBCommandRunner()
stdout, stderr, returncode = runner.run_adb_command(['invalid'])
assert returncode == 1
assert stdout == ""
assert stderr == "error message"
@patch('src.core.adb_command.get_adb_binary_path')
@patch('subprocess.run')
def test_run_adb_command_exception(self, mock_subprocess, mock_get_path):
"""Test ADB command execution with exception."""
mock_get_path.return_value = '/path/to/adb'
mock_subprocess.side_effect = FileNotFoundError("ADB not found")
runner = ADBCommandRunner()
stdout, stderr, returncode = runner.run_adb_command(['devices'])
assert returncode == -1
assert stdout is None
assert "ADB not found" in stderr
@patch('src.core.adb_command.get_adb_binary_path')
@patch('subprocess.Popen')
def test_run_adb_command_no_capture(self, mock_popen, mock_get_path):
"""Test ADB command execution without output capture."""
mock_get_path.return_value = '/path/to/adb'
mock_process = MagicMock()
mock_popen.return_value = mock_process
runner = ADBCommandRunner()
result = runner.run_adb_command(['devices'], capture_output=False)
assert result == mock_process
mock_popen.assert_called_once_with(
['/path/to/adb', 'devices'],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True
)
def test_check_device_no_process(self):
"""Test device check with no current process."""
runner = ADBCommandRunner()
result = runner.check_device()
assert result is None
@patch('src.core.adb_command.get_adb_binary_path')
@patch('subprocess.run')
def test_check_device_with_devices(self, mock_subprocess, mock_get_path):
"""Test device check with connected devices."""
mock_get_path.return_value = '/path/to/adb'
mock_result = MagicMock()
mock_result.returncode = 0
mock_result.stdout = "List of devices attached\ndevice1\tdevice\ndevice2\tdevice"
mock_subprocess.return_value = mock_result
runner = ADBCommandRunner()
result = runner.check_device()
assert result == "device1"
@patch('src.core.adb_command.get_adb_binary_path')
@patch('subprocess.run')
def test_check_device_no_devices(self, mock_subprocess, mock_get_path):
"""Test device check with no connected devices."""
mock_get_path.return_value = '/path/to/adb'
mock_result = MagicMock()
mock_result.returncode = 0
mock_result.stdout = "List of devices attached\n"
mock_subprocess.return_value = mock_result
runner = ADBCommandRunner()
result = runner.check_device()
assert result is None
def test_parse_progress_valid_percentage(self):
"""Test parsing valid progress percentage."""
runner = ADBCommandRunner()
result = runner.parse_progress("Transferring: 45%")
assert result == 45
def test_parse_progress_valid_fraction(self):
"""Test parsing valid progress fraction."""
runner = ADBCommandRunner()
result = runner.parse_progress("1024/2048 KB transferred")
assert result == 50
def test_parse_progress_valid_bytes(self):
"""Test parsing valid progress bytes."""
runner = ADBCommandRunner()
result = runner.parse_progress("2048 KB/s (1048576 bytes in 2.5s)")
assert result == 100
def test_parse_progress_invalid(self):
"""Test parsing invalid progress string."""
runner = ADBCommandRunner()
result = runner.parse_progress("No progress info here")
assert result is None
def test_parse_progress_empty(self):
"""Test parsing empty progress string."""
runner = ADBCommandRunner()
result = runner.parse_progress("")
assert result is None
def test_cancel_current_operation_no_process(self):
"""Test canceling operation with no current process."""
runner = ADBCommandRunner()
result = runner.cancel_current_operation()
assert result is False
def test_cancel_current_operation_with_process(self):
"""Test canceling operation with active process."""
runner = ADBCommandRunner()
mock_process = MagicMock()
mock_process.poll.return_value = None # Process is running
runner.current_process = mock_process
result = runner.cancel_current_operation()
assert result is True
mock_process.terminate.assert_called_once()
def test_cancel_current_operation_finished_process(self):
"""Test canceling operation with finished process."""
runner = ADBCommandRunner()
mock_process = MagicMock()
mock_process.poll.return_value = 0 # Process finished
runner.current_process = mock_process
result = runner.cancel_current_operation()
assert result is False
mock_process.terminate.assert_not_called()
+374 -302
View File
@@ -1,331 +1,403 @@
"""Tests for core ADB manager functionality."""
"""Tests for the main ADB manager module."""
import pytest
import unittest.mock as mock
import sys
from unittest.mock import MagicMock, patch, mock_open
import os
from unittest.mock import patch, MagicMock
from src.core.adb_manager import (
ADBManager,
get_executable_directory,
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
from src.core.adb_manager import ADBManager
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()
"""Test main ADB manager functionality."""
@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
def test_init_success(self, mock_get_path):
"""Test ADBManager initialization with successful ADB path."""
mock_get_path.return_value = '/path/to/adb'
stdout, stderr, returncode = adb_manager.run_adb_command(['devices'])
manager = ADBManager()
assert returncode == 0
assert stdout == "success output"
assert stderr == ""
assert manager.adb_path == '/path/to/adb'
assert manager.selected_device is None
assert manager.progress_callback is None
assert manager.status_callback is None
assert manager.transfer_progress['current_file'] == 0
@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")
def test_init_adb_path_failure(self, mock_get_path):
"""Test ADBManager initialization when ADB path fails."""
mock_get_path.side_effect = Exception("ADB not found")
stdout, stderr, returncode = adb_manager.run_adb_command(['devices'])
manager = ADBManager()
assert stdout is None
assert "Command failed" in stderr
assert returncode == -1
assert manager.adb_path is None
@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)
@patch('src.core.adb_manager.is_adb_available')
def test_is_available_true(self, mock_is_available):
"""Test ADB availability check returns True."""
mock_is_available.return_value = True
device_id = adb_manager.check_device()
assert device_id == "ABC123"
mock_run_command.assert_called_once_with(['devices'], capture_output=True)
manager = ADBManager()
assert manager.is_available() is 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)
@patch('src.core.adb_manager.is_adb_available')
def test_is_available_false(self, mock_is_available):
"""Test ADB availability check returns False."""
mock_is_available.return_value = False
device_id = adb_manager.check_device()
assert device_id is None
manager = ADBManager()
assert manager.is_available() is False
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
@patch('src.core.adb_manager.is_adb_available')
@patch('src.core.adb_manager.ensure_platform_tools_in_user_dir')
@patch('os.path.exists')
def test_ensure_adb_installed_already_available(self, mock_exists, mock_ensure, mock_is_available):
"""Test ensure_adb_installed when ADB is already available."""
mock_is_available.return_value = True
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()
manager = ADBManager()
result = manager.ensure_adb_installed()
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."""
mock_ensure.assert_not_called()
@patch('src.core.adb_manager.get_platform_type')
@patch('requests.get')
@patch('tempfile.mkdtemp')
@patch('src.core.adb_manager.is_adb_available')
@patch('src.core.adb_manager.ensure_platform_tools_in_user_dir')
@patch('os.path.exists')
def test_ensure_adb_installed_download_success(self, mock_exists, mock_ensure, mock_is_available):
"""Test ensure_adb_installed with successful download."""
mock_is_available.return_value = False
mock_ensure.return_value = '/user/data/adb'
mock_exists.return_value = True
manager = ADBManager()
result = manager.ensure_adb_installed()
assert result is True
assert manager.adb_path == '/user/data/adb'
@patch('src.core.adb_manager.is_adb_available')
@patch('src.core.adb_manager.ensure_platform_tools_in_user_dir')
def test_ensure_adb_installed_download_failure(self, mock_ensure, mock_is_available):
"""Test ensure_adb_installed with download failure."""
mock_is_available.return_value = False
mock_ensure.side_effect = Exception("Download failed")
manager = ADBManager()
result = manager.ensure_adb_installed()
assert result is False
def test_get_devices_success(self):
"""Test successful device enumeration."""
manager = ADBManager()
manager.command_runner.run_adb_command = MagicMock(
return_value=("List of devices attached\ndevice1\tdevice\ndevice2\tdevice", "", 0)
)
devices = manager.get_devices()
assert devices == ["device1", "device2"]
def test_get_devices_failure(self):
"""Test device enumeration failure."""
manager = ADBManager()
manager.command_runner.run_adb_command = MagicMock(
return_value=("", "Error", 1)
)
devices = manager.get_devices()
assert devices == []
def test_get_devices_no_devices(self):
"""Test device enumeration with no connected devices."""
manager = ADBManager()
manager.command_runner.run_adb_command = MagicMock(
return_value=("List of devices attached\n", "", 0)
)
devices = manager.get_devices()
assert devices == []
@patch.object(ADBManager, 'get_devices')
def test_is_device_connected_specific_device(self, mock_get_devices):
"""Test checking if specific device is connected."""
mock_get_devices.return_value = ["device1", "device2"]
manager = ADBManager()
assert manager.is_device_connected("device1") is True
assert manager.is_device_connected("device3") is False
@patch.object(ADBManager, 'get_devices')
def test_is_device_connected_any_device(self, mock_get_devices):
"""Test checking if any device is connected."""
mock_get_devices.return_value = ["device1"]
manager = ADBManager()
assert manager.is_device_connected() is True
mock_get_devices.return_value = []
assert manager.is_device_connected() is False
def test_select_device(self):
"""Test device selection."""
manager = ADBManager()
manager.select_device("test_device")
assert manager.selected_device == "test_device"
def test_get_selected_device(self):
"""Test getting selected device."""
manager = ADBManager()
manager.selected_device = "test_device"
assert manager.get_selected_device() == "test_device"
def test_get_selected_device_none(self):
"""Test getting selected device when none selected."""
manager = ADBManager()
assert manager.get_selected_device() is None
def test_list_files_success(self):
"""Test successful file listing."""
manager = ADBManager()
manager.command_runner.run_adb_command = MagicMock(
return_value=(
"total 12\n"
"drwxrwxr-x 2 user user 4096 Jan 1 12:00 Documents\n"
"-rw-rw-r-- 1 user user 100 Jan 1 12:00 test.txt",
"", 0
)
)
files = manager.list_files("/sdcard")
assert len(files) == 2
assert files[0]['name'] == 'Documents'
assert files[0]['type'] == 'folder'
assert files[1]['name'] == 'test.txt'
assert files[1]['type'] == 'file'
assert files[1]['size'] == 100
def test_list_files_failure(self):
"""Test file listing failure."""
manager = ADBManager()
manager.command_runner.run_adb_command = MagicMock(
return_value=("", "Permission denied", 1)
)
files = manager.list_files("/root")
assert files == []
@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'
def test_pull_file_success(self, mock_makedirs):
"""Test successful file pull."""
manager = ADBManager()
manager.file_transfer.pull_file = MagicMock(return_value=True)
success, message = manager.pull_file("/sdcard/test.txt", "/local/test.txt")
assert success is True
assert "successfully" in message
mock_makedirs.assert_called_once_with("/local", exist_ok=True)
def test_pull_file_failure(self):
"""Test file pull failure."""
manager = ADBManager()
manager.file_transfer.pull_file = MagicMock(return_value=False)
success, message = manager.pull_file("/sdcard/test.txt", "/local/test.txt")
assert success is False
assert "Failed" in message
@patch('os.makedirs')
def test_pull_folder_success(self, mock_makedirs):
"""Test successful folder pull."""
manager = ADBManager()
manager.file_transfer.pull_folder = MagicMock(return_value=True)
success, message = manager.pull_folder("/sdcard/Documents", "/local/Documents")
assert success is True
assert "successfully" in message
mock_makedirs.assert_called_once_with("/local/Documents", exist_ok=True)
def test_pull_folder_failure(self):
"""Test folder pull failure."""
manager = ADBManager()
manager.file_transfer.pull_folder = MagicMock(return_value=False)
success, message = manager.pull_folder("/sdcard/Documents", "/local/Documents")
assert success is False
assert "Failed" in message
@patch('os.path.exists')
def test_push_file_success(self, mock_exists):
"""Test successful file push."""
mock_exists.return_value = True
manager = ADBManager()
manager.file_transfer.push_file = MagicMock(return_value=True)
success, message = manager.push_file("/local/test.txt", "/sdcard/test.txt")
assert success is True
assert "successfully" in message
@patch('os.path.exists')
def test_push_file_not_found(self, mock_exists):
"""Test file push when local file doesn't exist."""
mock_exists.return_value = False
mock_mkdtemp.return_value = '/tmp/test'
mock_islink.return_value = False
manager = ADBManager()
# Mock HTTP response
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.content = b'fake zip content'
mock_get.return_value = mock_response
success, message = manager.push_file("/local/test.txt", "/sdcard/test.txt")
# 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__])
assert success is False
assert "not found" in message
@patch('os.path.exists')
def test_push_file_failure(self, mock_exists):
"""Test file push failure."""
mock_exists.return_value = True
manager = ADBManager()
manager.file_transfer.push_file = MagicMock(return_value=False)
success, message = manager.push_file("/local/test.txt", "/sdcard/test.txt")
assert success is False
assert "Failed" in message
@patch('os.path.exists')
def test_push_folder_success(self, mock_exists):
"""Test successful folder push."""
mock_exists.return_value = True
manager = ADBManager()
manager.file_transfer.push_folder = MagicMock(return_value=True)
success, message = manager.push_folder("/local/Documents", "/sdcard/Documents")
assert success is True
assert "successfully" in message
def test_delete_file_success(self):
"""Test successful file deletion."""
manager = ADBManager()
manager.command_runner.run_adb_command = MagicMock(
return_value=("", "", 0)
)
success, message = manager.delete_file("/sdcard/test.txt")
assert success is True
assert "deleted" in message
def test_delete_file_failure(self):
"""Test file deletion failure."""
manager = ADBManager()
manager.command_runner.run_adb_command = MagicMock(
return_value=("", "Permission denied", 1)
)
success, message = manager.delete_file("/sdcard/test.txt")
assert success is False
assert "Failed" in message
def test_create_folder_success(self):
"""Test successful folder creation."""
manager = ADBManager()
manager.command_runner.run_adb_command = MagicMock(
return_value=("", "", 0)
)
success, message = manager.create_folder("/sdcard/NewFolder")
assert success is True
assert "created" in message
def test_delete_folder_success(self):
"""Test successful folder deletion."""
manager = ADBManager()
manager.command_runner.run_adb_command = MagicMock(
return_value=("", "", 0)
)
success, message = manager.delete_folder("/sdcard/OldFolder")
assert success is True
assert "deleted" in message
def test_move_item_success(self):
"""Test successful item move/rename."""
manager = ADBManager()
manager.command_runner.run_adb_command = MagicMock(
return_value=("", "", 0)
)
success, message = manager.move_item("/sdcard/old.txt", "/sdcard/new.txt")
assert success is True
assert "moved" in message
def test_get_file_info_success(self):
"""Test successful file info retrieval."""
manager = ADBManager()
manager.command_runner.run_adb_command = MagicMock(
return_value=(
"-rw-rw-r-- 1 user user 1024 Jan 1 12:00 test.txt",
"", 0
)
)
info = manager.get_file_info("/sdcard/test.txt")
assert info is not None
assert info['name'] == 'test.txt'
assert info['type'] == 'file'
assert info['size'] == 1024
def test_get_file_info_failure(self):
"""Test file info retrieval failure."""
manager = ADBManager()
manager.command_runner.run_adb_command = MagicMock(
return_value=("", "File not found", 1)
)
info = manager.get_file_info("/sdcard/nonexistent.txt")
assert info is None
@patch('src.core.adb_manager.FileDeduplicator')
def test_deduplicate_files_success(self, mock_deduplicator_class):
"""Test successful file deduplication."""
mock_deduplicator = MagicMock()
mock_deduplicator.find_duplicates.return_value = [['file1.txt', 'file2.txt']]
mock_deduplicator.remove_duplicates.return_value = 1
mock_deduplicator_class.return_value = mock_deduplicator
manager = ADBManager()
removed_count, duplicates = manager.deduplicate_files("/test/folder")
assert removed_count == 1
assert len(duplicates) == 1
@patch('src.core.adb_manager.FileDeduplicator')
def test_deduplicate_files_no_duplicates(self, mock_deduplicator_class):
"""Test file deduplication with no duplicates found."""
mock_deduplicator = MagicMock()
mock_deduplicator.find_duplicates.return_value = []
mock_deduplicator_class.return_value = mock_deduplicator
manager = ADBManager()
removed_count, duplicates = manager.deduplicate_files("/test/folder")
assert removed_count == 0
assert duplicates == []
+218
View File
@@ -0,0 +1,218 @@
"""Tests for file transfer module."""
import pytest
import unittest.mock as mock
import os
from unittest.mock import patch, MagicMock
from src.core.file_transfer import ADBFileTransfer
class TestADBFileTransfer:
"""Test ADB file transfer functionality."""
def test_init(self):
"""Test ADBFileTransfer initialization."""
transfer = ADBFileTransfer()
assert transfer.progress_callback is None
assert transfer.current_process is None
def test_validate_windows_root_path_valid(self):
"""Test Windows root path validation with valid path."""
transfer = ADBFileTransfer()
# Should not raise exception
transfer._validate_windows_root_path("C:/Users/test", "push")
def test_validate_windows_root_path_invalid_push(self):
"""Test Windows root path validation with invalid path for push."""
transfer = ADBFileTransfer()
with pytest.raises(ValueError, match="Cannot push to Windows root"):
transfer._validate_windows_root_path("C:", "push")
def test_validate_windows_root_path_invalid_pull(self):
"""Test Windows root path validation with invalid path for pull."""
transfer = ADBFileTransfer()
with pytest.raises(ValueError, match="Cannot pull from Windows root"):
transfer._validate_windows_root_path("C:", "pull")
@patch('src.core.file_transfer.ADBCommandRunner')
@patch('os.path.exists')
def test_pull_file_success(self, mock_exists, mock_command_runner):
"""Test successful file pull operation."""
mock_exists.return_value = False # Remote file doesn't exist locally
mock_runner = MagicMock()
mock_runner.run_adb_command.return_value = ("", "", 0)
mock_command_runner.return_value = mock_runner
transfer = ADBFileTransfer()
result = transfer.pull_file("/sdcard/test.txt", "/local/test.txt")
assert result is True
mock_runner.run_adb_command.assert_called_with(['pull', '/sdcard/test.txt', '/local/test.txt'])
@patch('src.core.file_transfer.ADBCommandRunner')
@patch('os.path.exists')
def test_pull_file_already_exists(self, mock_exists, mock_command_runner):
"""Test file pull when local file already exists."""
mock_exists.return_value = True
mock_runner = MagicMock()
mock_command_runner.return_value = mock_runner
transfer = ADBFileTransfer()
result = transfer.pull_file("/sdcard/test.txt", "/local/test.txt")
assert result is False
mock_runner.run_adb_command.assert_not_called()
@patch('src.core.file_transfer.ADBCommandRunner')
@patch('os.path.exists')
def test_pull_file_command_failure(self, mock_exists, mock_command_runner):
"""Test file pull with ADB command failure."""
mock_exists.return_value = False
mock_runner = MagicMock()
mock_runner.run_adb_command.return_value = ("", "Error", 1)
mock_command_runner.return_value = mock_runner
transfer = ADBFileTransfer()
result = transfer.pull_file("/sdcard/test.txt", "/local/test.txt")
assert result is False
@patch('src.core.file_transfer.ADBCommandRunner')
@patch('os.path.exists')
@patch('os.path.isfile')
def test_push_file_success(self, mock_isfile, mock_exists, mock_command_runner):
"""Test successful file push operation."""
mock_exists.return_value = True # Local file exists
mock_isfile.return_value = True
mock_runner = MagicMock()
mock_runner.run_adb_command.return_value = ("", "", 0)
mock_command_runner.return_value = mock_runner
transfer = ADBFileTransfer()
result = transfer.push_file("/local/test.txt", "/sdcard/test.txt")
assert result is True
mock_runner.run_adb_command.assert_called_with(['push', '/local/test.txt', '/sdcard/test.txt'])
@patch('src.core.file_transfer.ADBCommandRunner')
@patch('os.path.exists')
def test_push_file_not_exists(self, mock_exists, mock_command_runner):
"""Test file push when local file doesn't exist."""
mock_exists.return_value = False
mock_runner = MagicMock()
mock_command_runner.return_value = mock_runner
transfer = ADBFileTransfer()
result = transfer.push_file("/local/test.txt", "/sdcard/test.txt")
assert result is False
mock_runner.run_adb_command.assert_not_called()
@patch('src.core.file_transfer.ADBCommandRunner')
@patch('os.path.exists')
@patch('os.path.isfile')
def test_push_file_not_a_file(self, mock_isfile, mock_exists, mock_command_runner):
"""Test file push when local path is not a file."""
mock_exists.return_value = True
mock_isfile.return_value = False # It's a directory, not a file
mock_runner = MagicMock()
mock_command_runner.return_value = mock_runner
transfer = ADBFileTransfer()
result = transfer.push_file("/local/test", "/sdcard/test.txt")
assert result is False
mock_runner.run_adb_command.assert_not_called()
@patch('src.core.file_transfer.ADBCommandRunner')
@patch('os.path.exists')
@patch('os.path.isdir')
def test_pull_folder_success(self, mock_isdir, mock_exists, mock_command_runner):
"""Test successful folder pull operation."""
mock_exists.return_value = False # Local folder doesn't exist
mock_isdir.return_value = True
mock_runner = MagicMock()
mock_runner.run_adb_command.return_value = ("", "", 0)
mock_command_runner.return_value = mock_runner
transfer = ADBFileTransfer()
result = transfer.pull_folder("/sdcard/Documents", "/local/Documents")
assert result is True
mock_runner.run_adb_command.assert_called_with(['pull', '/sdcard/Documents', '/local/Documents'])
@patch('src.core.file_transfer.ADBCommandRunner')
@patch('os.path.exists')
def test_pull_folder_already_exists(self, mock_exists, mock_command_runner):
"""Test folder pull when local folder already exists."""
mock_exists.return_value = True
mock_runner = MagicMock()
mock_command_runner.return_value = mock_runner
transfer = ADBFileTransfer()
result = transfer.pull_folder("/sdcard/Documents", "/local/Documents")
assert result is False
mock_runner.run_adb_command.assert_not_called()
@patch('src.core.file_transfer.ADBCommandRunner')
@patch('os.path.exists')
@patch('os.path.isdir')
def test_push_folder_success(self, mock_isdir, mock_exists, mock_command_runner):
"""Test successful folder push operation."""
mock_exists.return_value = True # Local folder exists
mock_isdir.return_value = True
mock_runner = MagicMock()
mock_runner.run_adb_command.return_value = ("", "", 0)
mock_command_runner.return_value = mock_runner
transfer = ADBFileTransfer()
result = transfer.push_folder("/local/Documents", "/sdcard/Documents")
assert result is True
mock_runner.run_adb_command.assert_called_with(['push', '/local/Documents', '/sdcard/Documents'])
@patch('src.core.file_transfer.ADBCommandRunner')
@patch('os.path.exists')
def test_push_folder_not_exists(self, mock_exists, mock_command_runner):
"""Test folder push when local folder doesn't exist."""
mock_exists.return_value = False
mock_runner = MagicMock()
mock_command_runner.return_value = mock_runner
transfer = ADBFileTransfer()
result = transfer.push_folder("/local/Documents", "/sdcard/Documents")
assert result is False
mock_runner.run_adb_command.assert_not_called()
def test_cancel_transfer_no_process(self):
"""Test canceling transfer with no current process."""
transfer = ADBFileTransfer()
result = transfer.cancel_transfer()
assert result is False
def test_cancel_transfer_with_process(self):
"""Test canceling transfer with active process."""
transfer = ADBFileTransfer()
mock_process = MagicMock()
mock_process.poll.return_value = None # Process is running
transfer.current_process = mock_process
result = transfer.cancel_transfer()
assert result is True
mock_process.terminate.assert_called_once()
def test_cancel_transfer_finished_process(self):
"""Test canceling transfer with finished process."""
transfer = ADBFileTransfer()
mock_process = MagicMock()
mock_process.poll.return_value = 0 # Process finished
transfer.current_process = mock_process
result = transfer.cancel_transfer()
assert result is False
mock_process.terminate.assert_not_called()
+114
View File
@@ -0,0 +1,114 @@
"""
Tests for platform_tools module.
"""
import os
import unittest
from unittest.mock import Mock, patch, mock_open, MagicMock
import pytest
from src.core.platform_tools import (
is_adb_available,
get_adb_binary_path,
ensure_platform_tools_in_user_dir,
download_and_extract_adb
)
class TestPlatformTools(unittest.TestCase):
"""Test cases for platform tools management."""
def test_is_adb_available_true(self):
"""Test adb availability when binary exists."""
with patch('src.core.platform_tools.get_adb_binary_path', return_value='/path/to/adb'):
with patch('os.path.isfile', return_value=True):
result = is_adb_available()
assert result is True
def test_is_adb_available_false(self):
"""Test adb availability when binary doesn't exist."""
with patch('src.core.platform_tools.get_adb_binary_path', return_value=None):
result = is_adb_available()
assert result is False
def test_get_adb_binary_path_user_dir_success(self):
"""Test getting ADB path from user directory."""
with patch('src.core.platform_tools.ensure_platform_tools_in_user_dir', return_value='/user/adb'):
with patch('os.path.isfile', return_value=True):
result = get_adb_binary_path()
assert result == '/user/adb'
def test_get_adb_binary_path_local_fallback(self):
"""Test fallback to local platform-tools."""
with patch('src.core.platform_tools.ensure_platform_tools_in_user_dir', side_effect=Exception()):
with patch('src.core.platform_utils.get_platform_tools_directory', return_value='/local/platform-tools'):
with patch('src.core.platform_utils.get_adb_binary_name', return_value='adb'):
with patch('os.path.isfile', return_value=True):
with patch('os.path.join', return_value='/local/platform-tools/adb'):
result = get_adb_binary_path()
assert result == '/local/platform-tools/adb'
def test_get_adb_binary_path_windows(self):
"""Test getting ADB path on Windows."""
with patch('src.core.platform_tools.ensure_platform_tools_in_user_dir', return_value='/user/adb.exe'):
with patch('os.path.isfile', return_value=True):
result = get_adb_binary_path()
assert result == '/user/adb.exe'
def test_ensure_platform_tools_simple(self):
"""Test basic platform tools installation."""
# Just test the function doesn't crash with basic mocking
with patch('os.makedirs'):
with patch('os.path.islink', return_value=False):
with patch('os.path.isdir', return_value=False):
with patch('tempfile.mkdtemp', return_value='/tmp/test'):
with patch('requests.get') as mock_get:
with patch('builtins.open', mock_open()):
with patch('zipfile.ZipFile') as mock_zip:
with patch('os.path.isdir', side_effect=lambda p: p == '/tmp/test/platform-tools'):
with patch('shutil.move'):
with patch('os.chmod'):
with patch('os.symlink'):
with patch('shutil.rmtree'):
mock_response = Mock()
mock_response.iter_content.return_value = [b'content']
mock_response.raise_for_status.return_value = None
mock_get.return_value = mock_response
result = ensure_platform_tools_in_user_dir()
assert result is not None
def test_download_and_extract_adb_linux(self):
"""Test ADB download and extraction on Linux."""
with patch('src.core.platform_tools.ensure_platform_tools_in_user_dir', return_value='/test/adb'):
with patch('os.path.isfile', return_value=True):
with patch('os.chmod') as mock_chmod:
with patch('os.name', 'posix'):
result = download_and_extract_adb()
assert result is True
mock_chmod.assert_called_once_with('/test/adb', 0o755)
def test_download_and_extract_adb_windows(self):
"""Test ADB download and extraction on Windows."""
with patch('src.core.platform_tools.ensure_platform_tools_in_user_dir', return_value='/test/adb.exe'):
with patch('os.path.isfile', return_value=True):
with patch('os.name', 'nt'):
result = download_and_extract_adb()
assert result is True
def test_download_and_extract_adb_failure(self):
"""Test ADB download failure when file doesn't exist."""
with patch('src.core.platform_tools.ensure_platform_tools_in_user_dir', return_value='/test/adb'):
with patch('os.path.isfile', return_value=False):
result = download_and_extract_adb()
assert result is False
def test_download_and_extract_adb_exception(self):
"""Test ADB download failure with exception."""
with patch('src.core.platform_tools.ensure_platform_tools_in_user_dir', side_effect=Exception()):
result = download_and_extract_adb()
assert result is False
if __name__ == '__main__':
unittest.main()
+91
View File
@@ -0,0 +1,91 @@
"""Tests for platform utilities module."""
import pytest
import unittest.mock as mock
import sys
import os
from unittest.mock import patch
from src.core.platform_utils import (
get_executable_directory,
get_platform_tools_directory,
get_platform_type
)
class TestPlatformUtils:
"""Test platform utility functions."""
def test_get_executable_directory_frozen(self):
"""Test get_executable_directory when running as executable."""
with patch.object(sys, 'frozen', True, create=True):
with patch.object(sys, 'executable', '/path/to/app'):
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'
# The function calls abspath on the actual platform_utils.py file
mock_abspath.assert_called_once()
def test_get_platform_type_linux(self):
"""Test platform type detection for Linux."""
with patch('sys.platform', 'linux'):
result = get_platform_type()
assert result == 'linux'
def test_get_platform_type_windows(self):
"""Test platform type detection for Windows."""
with patch('sys.platform', 'win32'):
result = get_platform_type()
assert result == 'win32'
def test_get_platform_type_darwin(self):
"""Test platform type detection for macOS."""
with patch('sys.platform', 'darwin'):
result = get_platform_type()
assert result == 'darwin'
def test_get_platform_tools_directory_frozen(self):
"""Test platform-tools directory when running as executable."""
with patch.object(sys, 'frozen', True, create=True):
with patch('src.core.platform_utils.get_executable_directory', return_value='/app/dir'):
result = get_platform_tools_directory()
assert result == '/app/dir/platform-tools'
def test_get_platform_tools_directory_src(self):
"""Test platform-tools directory when running from src."""
with patch.object(sys, 'frozen', False, create=True):
with patch('src.core.platform_utils.get_executable_directory', return_value='/project/src'):
result = get_platform_tools_directory()
assert result == '/project/src/platform-tools'
def test_get_platform_tools_directory_gui_subdirectory(self):
"""Test platform-tools directory when running from src/gui."""
with patch.object(sys, 'frozen', False, create=True):
with patch('src.core.platform_utils.get_executable_directory', return_value='/project/src/gui'):
result = get_platform_tools_directory()
assert result == '/project/src/platform-tools'
def test_get_platform_tools_directory_project_root(self):
"""Test platform-tools directory when running from project root."""
with patch.object(sys, 'frozen', False, create=True):
with patch('src.core.platform_utils.get_executable_directory', return_value='/project'):
with patch('os.path.exists', return_value=True):
result = get_platform_tools_directory()
assert result == '/project/src/platform-tools'
def test_get_platform_tools_directory_fallback(self):
"""Test platform-tools directory fallback behavior."""
with patch.object(sys, 'frozen', False, create=True):
with patch('src.core.platform_utils.get_executable_directory', return_value='/somewhere'):
with patch('os.path.exists', return_value=False):
result = get_platform_tools_directory()
assert result == '/somewhere/src/platform-tools'
+183
View File
@@ -0,0 +1,183 @@
"""Tests for progress tracking module."""
import pytest
import unittest.mock as mock
import time
from unittest.mock import patch, MagicMock
from src.core.progress_tracker import ProgressTracker
class TestProgressTracker:
"""Test progress tracking functionality."""
def test_init(self):
"""Test ProgressTracker initialization."""
tracker = ProgressTracker()
assert tracker.start_time is None
assert tracker.total_bytes == 0
assert tracker.transferred_bytes == 0
assert tracker.current_speed == 0.0
assert tracker.estimated_time_remaining == 0
def test_start_tracking(self):
"""Test starting progress tracking."""
tracker = ProgressTracker()
with patch('time.time', return_value=1000.0):
tracker.start_tracking(1024)
assert tracker.start_time == 1000.0
assert tracker.total_bytes == 1024
assert tracker.transferred_bytes == 0
def test_update_progress_first_update(self):
"""Test first progress update."""
tracker = ProgressTracker()
with patch('time.time', return_value=1000.0):
tracker.start_tracking(1024)
with patch('time.time', return_value=1001.0):
tracker.update_progress(512)
assert tracker.transferred_bytes == 512
assert tracker.current_speed == 512.0 # 512 bytes in 1 second
def test_update_progress_multiple_updates(self):
"""Test multiple progress updates."""
tracker = ProgressTracker()
with patch('time.time', return_value=1000.0):
tracker.start_tracking(1024)
with patch('time.time', return_value=1001.0):
tracker.update_progress(256)
with patch('time.time', return_value=1002.0):
tracker.update_progress(512)
assert tracker.transferred_bytes == 512
assert tracker.current_speed == 256.0 # Average speed
def test_update_progress_zero_time_elapsed(self):
"""Test progress update with zero time elapsed."""
tracker = ProgressTracker()
with patch('time.time', return_value=1000.0):
tracker.start_tracking(1024)
tracker.update_progress(512)
assert tracker.transferred_bytes == 512
assert tracker.current_speed == 0.0 # No time elapsed
def test_get_progress_percentage_no_total(self):
"""Test getting progress percentage with no total bytes."""
tracker = ProgressTracker()
assert tracker.get_progress_percentage() == 0.0
def test_get_progress_percentage_with_progress(self):
"""Test getting progress percentage with progress."""
tracker = ProgressTracker()
tracker.total_bytes = 1000
tracker.transferred_bytes = 250
assert tracker.get_progress_percentage() == 25.0
def test_get_progress_percentage_complete(self):
"""Test getting progress percentage when complete."""
tracker = ProgressTracker()
tracker.total_bytes = 1000
tracker.transferred_bytes = 1000
assert tracker.get_progress_percentage() == 100.0
def test_get_progress_percentage_over_100(self):
"""Test getting progress percentage over 100%."""
tracker = ProgressTracker()
tracker.total_bytes = 1000
tracker.transferred_bytes = 1200
assert tracker.get_progress_percentage() == 100.0 # Capped at 100%
def test_estimate_time_remaining_no_speed(self):
"""Test time estimation with no speed."""
tracker = ProgressTracker()
tracker.total_bytes = 1000
tracker.transferred_bytes = 250
tracker.current_speed = 0.0
assert tracker.estimate_time_remaining() == 0
def test_estimate_time_remaining_with_speed(self):
"""Test time estimation with speed."""
tracker = ProgressTracker()
tracker.total_bytes = 1000
tracker.transferred_bytes = 250
tracker.current_speed = 125.0 # 125 bytes/second
remaining_time = tracker.estimate_time_remaining()
assert remaining_time == 6 # (1000-250)/125 = 6 seconds
def test_estimate_time_remaining_complete(self):
"""Test time estimation when transfer is complete."""
tracker = ProgressTracker()
tracker.total_bytes = 1000
tracker.transferred_bytes = 1000
tracker.current_speed = 100.0
assert tracker.estimate_time_remaining() == 0
def test_reset_tracking(self):
"""Test resetting progress tracking."""
tracker = ProgressTracker()
tracker.start_time = 1000.0
tracker.total_bytes = 1000
tracker.transferred_bytes = 500
tracker.current_speed = 100.0
tracker.reset()
assert tracker.start_time is None
assert tracker.total_bytes == 0
assert tracker.transferred_bytes == 0
assert tracker.current_speed == 0.0
assert tracker.estimated_time_remaining == 0
def test_format_speed_bytes(self):
"""Test formatting speed in bytes per second."""
tracker = ProgressTracker()
tracker.current_speed = 512.0
formatted = tracker.format_speed()
assert formatted == "512.0 B/s"
def test_format_speed_kilobytes(self):
"""Test formatting speed in kilobytes per second."""
tracker = ProgressTracker()
tracker.current_speed = 1536.0 # 1.5 KB/s
formatted = tracker.format_speed()
assert formatted == "1.5 KB/s"
def test_format_speed_megabytes(self):
"""Test formatting speed in megabytes per second."""
tracker = ProgressTracker()
tracker.current_speed = 2097152.0 # 2 MB/s
formatted = tracker.format_speed()
assert formatted == "2.0 MB/s"
def test_format_time_seconds(self):
"""Test formatting time in seconds."""
tracker = ProgressTracker()
formatted = tracker.format_time(30)
assert formatted == "00:30"
def test_format_time_minutes(self):
"""Test formatting time in minutes and seconds."""
tracker = ProgressTracker()
formatted = tracker.format_time(150) # 2:30
assert formatted == "02:30"
def test_format_time_hours(self):
"""Test formatting time in hours, minutes and seconds."""
tracker = ProgressTracker()
formatted = tracker.format_time(3661) # 1:01:01
assert formatted == "01:01:01"