292 lines
13 KiB
Python
292 lines
13 KiB
Python
"""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__]) |