added initial test suite and fixed pytest env problems
This commit is contained in:
Jason Ross
@@ -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__])
|
||||
Reference in New Issue
Block a user