"""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__])