diff --git a/tests/unit/test_convenience_functions.py b/tests/unit/test_convenience_functions.py index 85735f4..6f8856c 100644 --- a/tests/unit/test_convenience_functions.py +++ b/tests/unit/test_convenience_functions.py @@ -178,3 +178,80 @@ class TestCompressionLevels: assert "compression level" in str(exc_info.value).lower() assert "1" in str(exc_info.value) and "22" in str(exc_info.value) + + +class TestEdgeCaseCoverage: + """Test edge cases to improve coverage.""" + + def test_empty_files_list(self, temp_dir): + """Test create_archive with empty files list.""" + archive_path = temp_dir / "empty.tzst" + + # Should create an empty archive + create_archive(archive_path, []) + + assert archive_path.exists() + contents = list_archive(archive_path) + assert len(contents) == 0 + + def test_create_archive_with_use_temp_file_false(self, temp_dir): + """Test creating archive with use_temp_file=False.""" + test_file = temp_dir / "test.txt" + test_file.write_text("test content") + archive_path = temp_dir / "test.tzst" + + # Test non-atomic mode + create_archive(archive_path, [str(test_file)], use_temp_file=False) + + assert archive_path.exists() + contents = list_archive(archive_path) + assert len(contents) == 1 + + def test_extract_archive_to_specific_path(self, temp_dir): + """Test extracting archive to specific path.""" + # Create test archive + test_file = temp_dir / "test.txt" + test_file.write_text("test content") + archive_path = temp_dir / "test.tzst" + create_archive(archive_path, [str(test_file)]) + + # Extract to specific directory + extract_dir = temp_dir / "extracted" + extract_archive(archive_path, extract_dir) + + assert extract_dir.exists() + assert (extract_dir / "test.txt").exists() + assert (extract_dir / "test.txt").read_text() == "test content" + + def test_list_archive_with_streaming(self, temp_dir): + """Test listing archive with streaming mode.""" + test_file = temp_dir / "test.txt" + test_file.write_text("test content") + archive_path = temp_dir / "test.tzst" + create_archive(archive_path, [str(test_file)]) + + # Test with streaming=True + contents = list_archive(archive_path, streaming=True) + assert len(contents) == 1 + assert contents[0]["name"] == "test.txt" + + def test_test_archive_success(self, temp_dir): + """Test testing a valid archive.""" + test_file = temp_dir / "test.txt" + test_file.write_text("test content") + archive_path = temp_dir / "test.tzst" + create_archive(archive_path, [str(test_file)]) + + # Test archive - should return True for valid archive + result = tzst_test_archive(archive_path) + assert result is True + + def test_test_archive_failure(self, temp_dir): + """Test testing an invalid archive.""" + # Create a file that's not a valid archive + invalid_archive = temp_dir / "invalid.tzst" + invalid_archive.write_text("This is not a valid archive") + + # Test archive - should return False for invalid archive + result = tzst_test_archive(invalid_archive) + assert result is False diff --git a/tests/unit/test_security_and_errors.py b/tests/unit/test_security_and_errors.py index 6ea782c..76b46c6 100644 --- a/tests/unit/test_security_and_errors.py +++ b/tests/unit/test_security_and_errors.py @@ -119,6 +119,132 @@ class TestCompressionValidation: archive_path, file_paths, compression_level=invalid_level ) + def test_specific_compression_level_validation(self, temp_dir): + """Test specific compression level validation to cover missing lines.""" + import tempfile + from pathlib import Path + + with tempfile.NamedTemporaryFile(suffix=".tzst", delete=False) as f: + archive_path = Path(f.name) + + try: + # Test compression level too low (line 60) + with pytest.raises(ValueError, match="Invalid compression level '0'"): + TzstArchive(archive_path, mode="w", compression_level=0) + + # Test compression level too high (line 66) + with pytest.raises(ValueError, match="Invalid compression level '23'"): + TzstArchive(archive_path, mode="w", compression_level=23) + + finally: + archive_path.unlink(missing_ok=True) + + def test_invalid_mode_validation(self, temp_dir): + """Test invalid mode validation to cover missing lines.""" + import tempfile + from pathlib import Path + + with tempfile.NamedTemporaryFile(suffix=".tzst", delete=False) as f: + archive_path = Path(f.name) + + try: + # Test invalid mode (line 54) + with pytest.raises(ValueError, match="Invalid mode 'x'"): + TzstArchive(archive_path, mode="x") + + # Test invalid mode with additional characters + with pytest.raises(ValueError, match="Invalid mode 'rb'"): + TzstArchive(archive_path, mode="rb") + + finally: + archive_path.unlink(missing_ok=True) + + def test_runtime_errors_for_wrong_mode_operations(self, temp_dir): + """Test RuntimeError for operations on wrong mode archives.""" + import tempfile + from pathlib import Path + + with tempfile.NamedTemporaryFile(suffix=".tzst", delete=False) as f: + archive_path = Path(f.name) + + try: + # Create empty archive first + with TzstArchive(archive_path, mode="w") as archive: + pass + + # Test read operations on write mode + with TzstArchive(archive_path, mode="w") as archive: + with pytest.raises(RuntimeError, match="Archive not open for reading"): + archive.getmembers() + + with pytest.raises(RuntimeError, match="Archive not open for reading"): + archive.getnames() + + with pytest.raises(RuntimeError, match="Archive not open for reading"): + archive.extractfile("test") + + finally: + archive_path.unlink(missing_ok=True) + + def test_operations_on_closed_archive(self, temp_dir): + """Test operations on closed archive to cover missing lines.""" + import tempfile + from pathlib import Path + + with tempfile.NamedTemporaryFile(suffix=".tzst", delete=False) as f: + archive_path = Path(f.name) + + try: + # Create and close archive + archive = TzstArchive(archive_path, mode="w") + archive.close() + + # Test operations on closed archive + with pytest.raises(RuntimeError, match="Archive not open"): + archive.getmembers() + + with pytest.raises(RuntimeError, match="Archive not open"): + archive.getnames() + + with pytest.raises(RuntimeError, match="Archive not open"): + archive.extractfile("test") + + finally: + archive_path.unlink(missing_ok=True) + + def test_close_error_handling(self, temp_dir): + """Test error handling in close method.""" + from pathlib import Path + from unittest.mock import MagicMock + + # Create a mock archive that will raise exceptions during close + archive = TzstArchive.__new__(TzstArchive) + archive.path = Path("test.tzst") + archive.mode = "w" + archive.compression_level = 3 + + # Create mock objects that raise exceptions when closed + mock_tarfile = MagicMock() + mock_tarfile.close.side_effect = Exception("Mock tarfile close error") + + mock_stream = MagicMock() + mock_stream.close.side_effect = Exception("Mock stream close error") + + mock_fileobj = MagicMock() + mock_fileobj.close.side_effect = Exception("Mock fileobj close error") + + archive._tarfile = mock_tarfile + archive._compressed_stream = mock_stream + archive._fileobj = mock_fileobj + + # This should not raise an exception despite the mock exceptions + archive.close() + + # Verify all close methods were called + mock_tarfile.close.assert_called_once() + mock_stream.close.assert_called_once() + mock_fileobj.close.assert_called_once() + class TestSpecialFileTypes: """Test handling of special file types and edge cases.""" @@ -207,3 +333,100 @@ class TestAppendModeDocumentation: assert ( "multiple archives" in error_msg.lower() or "recreate" in error_msg.lower() ) + + +class TestSpecificMissingLineCoverage: + """Test specific missing lines from coverage report.""" + + def test_invalid_mode_validation_specific(self, temp_dir): + """Test specific invalid mode validation to cover missing lines.""" + import tempfile + from pathlib import Path + + with tempfile.NamedTemporaryFile(suffix=".tzst", delete=False) as f: + archive_path = Path(f.name) + + try: + # Test invalid mode (line 54) + with pytest.raises(ValueError, match="Invalid mode 'x'"): + TzstArchive(archive_path, mode="x") + + # Test invalid mode with additional characters + with pytest.raises(ValueError, match="Invalid mode 'rb'"): + TzstArchive(archive_path, mode="rb") + + finally: + archive_path.unlink(missing_ok=True) + + def test_compression_level_validation_specific(self, temp_dir): + """Test specific compression level validation to cover missing lines.""" + import tempfile + from pathlib import Path + + with tempfile.NamedTemporaryFile(suffix=".tzst", delete=False) as f: + archive_path = Path(f.name) + + try: + # Test compression level too low (line 60) + with pytest.raises(ValueError, match="Invalid compression level '0'"): + TzstArchive(archive_path, mode="w", compression_level=0) + + # Test compression level too high (line 66) + with pytest.raises(ValueError, match="Invalid compression level '23'"): + TzstArchive(archive_path, mode="w", compression_level=23) + + finally: + archive_path.unlink(missing_ok=True) + + def test_runtime_errors_for_wrong_mode_operations(self, temp_dir): + """Test RuntimeError for operations on wrong mode archives.""" + import tempfile + from pathlib import Path + + with tempfile.NamedTemporaryFile(suffix=".tzst", delete=False) as f: + archive_path = Path(f.name) + + try: + # Create empty archive first + with TzstArchive(archive_path, mode="w") as archive: + pass + + # Test read operations on write mode + with TzstArchive(archive_path, mode="w") as archive: + with pytest.raises(RuntimeError, match="Archive not open for reading"): + archive.getmembers() + + with pytest.raises(RuntimeError, match="Archive not open for reading"): + archive.getnames() + + with pytest.raises(RuntimeError, match="Archive not open for reading"): + archive.extractfile("test") + + finally: + archive_path.unlink(missing_ok=True) + + def test_operations_on_closed_archive_specific(self, temp_dir): + """Test operations on closed archive to cover missing lines.""" + import tempfile + from pathlib import Path + + with tempfile.NamedTemporaryFile(suffix=".tzst", delete=False) as f: + archive_path = Path(f.name) + + try: + # Create and close archive + archive = TzstArchive(archive_path, mode="w") + archive.close() + + # Test operations on closed archive + with pytest.raises(RuntimeError, match="Archive not open"): + archive.getmembers() + + with pytest.raises(RuntimeError, match="Archive not open"): + archive.getnames() + + with pytest.raises(RuntimeError, match="Archive not open"): + archive.extractfile("test") + + finally: + archive_path.unlink(missing_ok=True)