"""Tests for Compression Module. Tests cover gzip_file, gzip_files_in_directory, ungzip_file, CompressionResult, and error handling. """ import gzip import pytest from pathlib import Path from src.utils.compression import ( gzip_file, gzip_files_in_directory, ungzip_file, CompressionResult, CompressionError, ) # ==================== CompressionResult Tests ==================== def test_compression_result_ratio(): """Test compression ratio calculation.""" result = CompressionResult( source_path=Path('/tmp/test.txt'), output_path=Path('/tmp/test.txt.gz'), original_size=1000, compressed_size=200, ) assert result.compression_ratio == 0.2 assert result.savings_percent == 80.0 def test_compression_result_ratio_zero_original(): """Test compression ratio with zero original size (avoid div by zero).""" result = CompressionResult( source_path=Path('/tmp/empty.txt'), output_path=Path('/tmp/empty.txt.gz'), original_size=0, compressed_size=20, ) assert result.compression_ratio == 1.0 def test_compression_result_repr(): """Test CompressionResult string representation.""" result = CompressionResult( source_path=Path('/tmp/data.TXT'), output_path=Path('/tmp/data.TXT.gz'), original_size=1000, compressed_size=500, ) repr_str = repr(result) assert 'data.TXT' in repr_str assert 'data.TXT.gz' in repr_str assert '50.00%' in repr_str # ==================== gzip_file Tests ==================== def test_gzip_file_success(tmp_path): """Test successful gzip compression.""" # Create test file source = tmp_path / 'test.TXT' source.write_text('Hello, World! ' * 100) result = gzip_file(source) assert result.output_path.exists() assert result.output_path.name == 'test.TXT.gz' assert result.original_size > 0 assert result.compressed_size > 0 assert result.compressed_size < result.original_size assert source.exists() # Original should still exist def test_gzip_file_custom_output_path(tmp_path): """Test gzip with custom output path.""" source = tmp_path / 'data.txt' source.write_text('Test data') output = tmp_path / 'custom' / 'output.gz' result = gzip_file(source, output_path=output) assert result.output_path == output assert output.exists() def test_gzip_file_delete_original(tmp_path): """Test gzip with delete_original=True.""" source = tmp_path / 'to_delete.txt' source.write_text('Delete me after compression') result = gzip_file(source, delete_original=True) assert result.output_path.exists() assert not source.exists() def test_gzip_file_source_not_found(tmp_path): """Test FileNotFoundError for nonexistent source.""" nonexistent = tmp_path / 'does_not_exist.txt' with pytest.raises(FileNotFoundError, match="not found"): gzip_file(nonexistent) def test_gzip_file_source_is_directory(tmp_path): """Test CompressionError when source is a directory.""" directory = tmp_path / 'my_dir' directory.mkdir() with pytest.raises(CompressionError, match="not a file"): gzip_file(directory) def test_gzip_file_content_preserved(tmp_path): """Test that gzip preserves file content correctly.""" original_content = 'This is the original content!\n' * 50 source = tmp_path / 'content.txt' source.write_text(original_content) result = gzip_file(source) # Decompress and verify content with gzip.open(result.output_path, 'rt') as f: decompressed = f.read() assert decompressed == original_content def test_gzip_file_compression_level(tmp_path): """Test that compression level affects output size.""" # Create file with compressible content source = tmp_path / 'compressible.txt' source.write_text('AAAA' * 10000) # Compress with low level output_low = tmp_path / 'low.gz' result_low = gzip_file(source, output_path=output_low, compression_level=1) # Compress with high level output_high = tmp_path / 'high.gz' result_high = gzip_file( source, output_path=output_high, compression_level=9 ) # Higher compression should result in smaller (or equal) file assert result_high.compressed_size <= result_low.compressed_size # ==================== gzip_files_in_directory Tests ==================== def test_gzip_files_in_directory_success(tmp_path): """Test batch compression of files in directory.""" # Create test files for i in range(3): f = tmp_path / f'file{i}.TXT' f.write_text(f'Content {i} ' * 100) successful, failed = gzip_files_in_directory(tmp_path, pattern='*.TXT') assert len(successful) == 3 assert len(failed) == 0 for result in successful: assert result.output_path.exists() assert result.output_path.suffix == '.gz' def test_gzip_files_in_directory_pattern_filtering(tmp_path): """Test that pattern filtering works.""" # Create mixed files (tmp_path / 'data1.TXT').write_text('txt file 1') (tmp_path / 'data2.TXT').write_text('txt file 2') (tmp_path / 'other.csv').write_text('csv file') (tmp_path / 'readme.md').write_text('markdown') successful, failed = gzip_files_in_directory(tmp_path, pattern='*.TXT') assert len(successful) == 2 assert all('TXT' in str(r.source_path) for r in successful) def test_gzip_files_in_directory_delete_originals(tmp_path): """Test batch compression with delete_originals=True.""" files = [] for i in range(2): f = tmp_path / f'delete_me{i}.TXT' f.write_text(f'Content {i}') files.append(f) successful, _ = gzip_files_in_directory( tmp_path, pattern='*.TXT', delete_originals=True ) assert len(successful) == 2 for f in files: assert not f.exists() # Originals deleted def test_gzip_files_in_directory_nonexistent(tmp_path): """Test FileNotFoundError for nonexistent directory.""" nonexistent = tmp_path / 'does_not_exist' with pytest.raises(FileNotFoundError, match="not found"): gzip_files_in_directory(nonexistent) def test_gzip_files_in_directory_not_a_dir(tmp_path): """Test CompressionError when path is not a directory.""" file_path = tmp_path / 'file.txt' file_path.write_text('I am a file') with pytest.raises(CompressionError, match="not a directory"): gzip_files_in_directory(file_path) def test_gzip_files_in_directory_no_matches(tmp_path): """Test empty results when no files match pattern.""" (tmp_path / 'file.csv').write_text('csv') successful, failed = gzip_files_in_directory(tmp_path, pattern='*.TXT') assert len(successful) == 0 assert len(failed) == 0 # ==================== ungzip_file Tests ==================== def test_ungzip_file_success(tmp_path): """Test successful gzip decompression.""" # Create and compress a file source = tmp_path / 'original.txt' original_content = 'Original content for decompression test' source.write_text(original_content) gzip_result = gzip_file(source) # Decompress decompressed = ungzip_file(gzip_result.output_path) assert decompressed.exists() assert decompressed.name == 'original.txt' assert decompressed.read_text() == original_content def test_ungzip_file_custom_output(tmp_path): """Test ungzip with custom output path.""" source = tmp_path / 'data.txt' source.write_text('Data') gzip_result = gzip_file(source) output = tmp_path / 'custom' / 'decompressed.txt' result = ungzip_file(gzip_result.output_path, output_path=output) assert result == output assert output.exists() def test_ungzip_file_delete_original(tmp_path): """Test ungzip with delete_original=True.""" source = tmp_path / 'file.txt' source.write_text('Content') gzip_result = gzip_file(source) ungzip_file(gzip_result.output_path, delete_original=True) assert not gzip_result.output_path.exists() def test_ungzip_file_not_found(tmp_path): """Test FileNotFoundError for nonexistent gzip file.""" nonexistent = tmp_path / 'not_here.gz' with pytest.raises(FileNotFoundError, match="not found"): ungzip_file(nonexistent) def test_ungzip_file_invalid_gzip(tmp_path): """Test CompressionError for invalid gzip file.""" invalid = tmp_path / 'invalid.gz' invalid.write_text('This is not a valid gzip file') with pytest.raises(CompressionError): ungzip_file(invalid) def test_ungzip_file_non_gz_extension(tmp_path): """Test ungzip handles files without .gz extension.""" source = tmp_path / 'data.txt' source.write_text('Data') # Compress to non-.gz name gzip_output = tmp_path / 'compressed.bin' gzip_file(source, output_path=gzip_output) # Decompress - should create .decompressed file result = ungzip_file(gzip_output) assert result.suffix == '.decompressed' # ==================== Round-trip Tests ==================== def test_gzip_ungzip_roundtrip(tmp_path): """Test complete compression/decompression round-trip.""" original = tmp_path / 'roundtrip.TXT' original_content = 'Line 1\nLine 2\nLine 3\n' * 100 original.write_text(original_content) # Compress gzip_result = gzip_file(original, delete_original=True) assert not original.exists() # Decompress restored = ungzip_file(gzip_result.output_path, delete_original=True) assert not gzip_result.output_path.exists() # Verify content assert restored.read_text() == original_content def test_gzip_binary_content(tmp_path): """Test gzip with binary content.""" source = tmp_path / 'binary.dat' binary_content = bytes(range(256)) * 10 source.write_bytes(binary_content) result = gzip_file(source) # Verify via decompression with gzip.open(result.output_path, 'rb') as f: decompressed = f.read() assert decompressed == binary_content