diff --git a/src/services/metadata_factory.py b/src/services/metadata_factory.py index 3d9e11fe..ef5aec68 100644 --- a/src/services/metadata_factory.py +++ b/src/services/metadata_factory.py @@ -9,6 +9,7 @@ extensibility for supporting new file formats (PDF, Office docs, etc.). from pathlib import Path from src.services.image_handler import ImageHandler +from src.services.pdf_handler import PDFHandler from src.utils.exceptions import UnsupportedFormatError @@ -45,10 +46,10 @@ class MetadataFactory: if Path(filepath).is_file(): if ext in [".jpg", ".jpeg", ".png"]: return ImageHandler(filepath) + elif ext == ".pdf": + return PDFHandler(filepath) # TODO: implement other handlers - # elif ext == ".pdf": - # return PDFHandler(filepath) # elif ext == ".xlsx": # return ExcelHandler(filepath) # elif ext == ".pptx": diff --git a/src/services/metadata_handler.py b/src/services/metadata_handler.py index 8d0e94f0..5a189f23 100644 --- a/src/services/metadata_handler.py +++ b/src/services/metadata_handler.py @@ -61,7 +61,7 @@ class MetadataHandler(ABC): pass @abstractmethod - def save(self, output_path: str | None = None) -> None: + def save(self, output_path: str) -> None: """ Save the processed file with cleaned metadata. diff --git a/src/services/pdf_handler.py b/src/services/pdf_handler.py new file mode 100644 index 00000000..57b03b1b --- /dev/null +++ b/src/services/pdf_handler.py @@ -0,0 +1,102 @@ +""" +PDF metadata handler for PDF files. +Does not support encrypted files. + +This module provides the PdfHandler class which implements the MetadataHandler +interface for PDF files. It delegates the actual metadata operations to +format-specific processors (PdfProcessor). +""" + +import shutil +from pathlib import Path + +from pypdf import PdfReader, PdfWriter + +from src.services.metadata_handler import MetadataHandler +from src.utils.exceptions import ( + MetadataNotFoundError, + MetadataReadingError, + UnsupportedFormatError, +) + + +class PDFHandler(MetadataHandler): + """ + PDF metadata handler for PDF files. + """ + + def __init__(self, filepath: str): + """ + Initialize the pdf handler. + + Args: + filepath: Path to the pdf file to process. + """ + super().__init__(filepath) + self.keys_to_delete: list[str] = [] + + def _detect_format(self) -> str: + """ + Detect actual pdf format using, not file extension. + + Returns: + Normalized format string ('pdf'). + + Raises: + UnsupportedFormatError: If format is not supported or undetectable. + """ + normalise = Path(self.filepath).suffix.lower() + if normalise != ".pdf": + raise UnsupportedFormatError(f"Unsupported format: {normalise}") + + return normalise[1:] + + def read(self): + """ + Extract metadata from the file. + + Uses actual format detection to select the appropriate processor. + """ + with PdfReader(Path(self.filepath)) as reader: + if reader.is_encrypted: + raise MetadataReadingError("File is encrypted.") + + if reader.metadata is None: + raise MetadataNotFoundError("No metadata found in the file.") + + for key, value in reader.metadata.items(): + self.metadata[key] = value + self.keys_to_delete.append(key) + return self.metadata + + def wipe(self) -> None: + """ + Remove metadata from PDF file. + """ + with PdfReader(Path(self.filepath)) as reader: + metadata = reader.metadata + if metadata is None: + raise MetadataNotFoundError("No metadata found in the file.") + + for key in list(metadata): + if key in self.keys_to_delete: + del metadata[key] + + self.processed_metadata = metadata + + def save(self, output_path: str | Path | None = None) -> None: + """ + Save the changes to a copy of the original file. + """ + destination_file_path = Path(output_path) if output_path else None + if not destination_file_path: + raise ValueError("output_path is required") + + shutil.copy2(self.filepath, destination_file_path) + with PdfReader(destination_file_path) as reader, PdfWriter() as writer: + # Copy all pages + for page in reader.pages: + writer.add_page(page) + + writer.add_metadata(self.processed_metadata) + writer.write(destination_file_path) diff --git a/src/utils/display.py b/src/utils/display.py index ccf0fc33..270e15e6 100644 --- a/src/utils/display.py +++ b/src/utils/display.py @@ -31,6 +31,11 @@ def print_metadata_table(metadata: dict[str, Any]): # Define the groups using simple lists of keys groups = { + "📄 Document Info": [ + "Author", + "/Author", + "/Creator", + ], "📸 Device Info": ["Make", "Model", "Software", "ExifVersion"], "⚙️ Exposure Settings": [ "ExposureTime", @@ -49,7 +54,14 @@ def print_metadata_table(metadata: dict[str, Any]): "Orientation", "ResolutionUnit", ], - "📅 Dates": ["DateTime", "DateTimeOriginal", "DateTimeDigitized", "OffsetTime"], + "📅 Dates": [ + "DateTime", + "DateTimeOriginal", + "DateTimeDigitized", + "OffsetTime", + "/CreationDate", + "/ModDate", + ], } # Create the main table diff --git a/src/utils/exceptions.py b/src/utils/exceptions.py index 8cf00cd7..9d6484b6 100644 --- a/src/utils/exceptions.py +++ b/src/utils/exceptions.py @@ -28,3 +28,9 @@ class MetadataProcessingError(MetadataException): """Raised when an error occurs during metadata processing.""" pass + + +class MetadataReadingError(MetadataException): + """Raised when an error occurs during metadata reading.""" + + pass diff --git a/src/utils/formatter.py b/src/utils/formatter.py index 997bfa3a..be2544c6 100644 --- a/src/utils/formatter.py +++ b/src/utils/formatter.py @@ -31,6 +31,10 @@ def clean_value(value: Any) -> str: if isinstance(value, tuple) or isinstance(value, list): return "/".join(map(str, value)) + if isinstance(value, str) and value.startswith("D:"): + clean_iso = f"{value[2:6]}-{value[6:8]}-{value[8:10]}T{value[10:12]}:{value[12:14]}:{value[14:16]}" + return clean_iso + # Handle empty values if value == "": return "-" diff --git a/tests/assets/test_pdfs/file-example_PDF_1MB.pdf b/tests/assets/test_pdfs/file-example_PDF_1MB.pdf new file mode 100644 index 00000000..13ce5749 Binary files /dev/null and b/tests/assets/test_pdfs/file-example_PDF_1MB.pdf differ diff --git a/tests/assets/test_pdfs/sample.pdf b/tests/assets/test_pdfs/sample.pdf new file mode 100644 index 00000000..c01805e8 Binary files /dev/null and b/tests/assets/test_pdfs/sample.pdf differ diff --git a/tests/conftest.py b/tests/conftest.py index f2167b54..e000b353 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -12,6 +12,7 @@ import pytest TESTS_DIR = Path(__file__).parent ASSETS_DIR = TESTS_DIR / "assets" TEST_IMAGES_DIR = ASSETS_DIR / "test_images" +TEST_PDFS_DIR = ASSETS_DIR / "test_pdfs" @pytest.fixture @@ -46,3 +47,31 @@ def get_png_test_file() -> str: def get_test_images_dir() -> str: """Get test images directory path as string.""" return str(TEST_IMAGES_DIR) + + +@pytest.fixture +def pdf_test_file() -> Path: + """Return path to a PDF test file with metadata.""" + return TEST_PDFS_DIR / "sample.pdf" + + +@pytest.fixture +def test_pdfs_dir() -> Path: + """Return path to test PDFs directory.""" + return TEST_PDFS_DIR + + +# String versions for parametrize (PDF) +def get_pdf_test_file() -> str: + """Get PDF test file path as string.""" + return str(TEST_PDFS_DIR / "sample.pdf") + + +def get_large_pdf_test_file() -> str: + """Get large PDF test file path as string.""" + return str(TEST_PDFS_DIR / "file-example_PDF_1MB.pdf") + + +def get_test_pdfs_dir() -> str: + """Get test PDFs directory path as string.""" + return str(TEST_PDFS_DIR) diff --git a/tests/e2e/test_read_command.py b/tests/e2e/test_read_command.py index 9972a77c..fc17ad3e 100644 --- a/tests/e2e/test_read_command.py +++ b/tests/e2e/test_read_command.py @@ -6,7 +6,13 @@ from typer.testing import CliRunner from src.main import app # Import path helpers from conftest -from tests.conftest import get_jpg_test_file, get_png_test_file, get_test_images_dir +from tests.conftest import ( + get_jpg_test_file, + get_pdf_test_file, + get_png_test_file, + get_test_images_dir, + get_test_pdfs_dir, +) runner = CliRunner() @@ -14,6 +20,8 @@ runner = CliRunner() JPG_TEST_FILE = get_jpg_test_file() PNG_TEST_FILE = get_png_test_file() TEST_DIR = get_test_images_dir() +PDF_TEST_FILE = get_pdf_test_file() +PDF_DIR = get_test_pdfs_dir() # ============== Success Case Tests ============== @@ -74,3 +82,27 @@ def test_read_command_file_not_found(): assert result.exit_code == 2 assert "Invalid value for 'FILE_PATH'" in result.stderr assert "does not exist" in result.stderr + + +# ============== PDF E2E Read Tests ============== + + +def test_read_command_pdf_single_file_success(): + """ + Test the 'read' command with a single PDF file. + """ + result = runner.invoke(app, ["read", PDF_TEST_FILE]) + + assert result.exit_code == 0, f"Failed with: {result.stdout}" + assert "Reading" in result.stdout + assert Path(PDF_TEST_FILE).name in result.stdout + + +def test_read_command_recursive_pdf_success(): + """ + Test the 'read' command with recursive directory processing for PDF. + """ + result = runner.invoke(app, ["read", PDF_DIR, "-r", "-ext", "pdf"]) + + assert result.exit_code == 0, f"Failed with: {result.stdout}" + assert "Reading" in result.stdout diff --git a/tests/e2e/test_scrub_command.py b/tests/e2e/test_scrub_command.py index bc64e482..5e593cd9 100644 --- a/tests/e2e/test_scrub_command.py +++ b/tests/e2e/test_scrub_command.py @@ -12,7 +12,13 @@ from typer.testing import CliRunner from src.main import app # Import path helpers from conftest -from tests.conftest import get_jpg_test_file, get_png_test_file, get_test_images_dir +from tests.conftest import ( + get_jpg_test_file, + get_pdf_test_file, + get_png_test_file, + get_test_images_dir, + get_test_pdfs_dir, +) runner = CliRunner() @@ -20,6 +26,8 @@ runner = CliRunner() JPG_TEST_FILE = get_jpg_test_file() PNG_TEST_FILE = get_png_test_file() EXAMPLES_DIR = get_test_images_dir() +PDF_TEST_FILE = get_pdf_test_file() +PDF_DIR = get_test_pdfs_dir() @pytest.fixture @@ -127,3 +135,69 @@ def test_scrub_command_requires_recursive_with_ext(): result = runner.invoke(app, ["scrub", JPG_TEST_FILE, "-ext", "jpg"]) assert result.exit_code != 0 + + +# ============== PDF E2E Tests ============== + + +def test_scrub_command_pdf_single_file_success(output_dir): + """ + Test the 'scrub' command with a single PDF file. + """ + result = runner.invoke(app, ["scrub", PDF_TEST_FILE, "--output", str(output_dir)]) + + assert result.exit_code == 0, f"Failed with: {result.stdout}" + # Check output file was created (has processed_ prefix) + output_file = output_dir / f"processed_{Path(PDF_TEST_FILE).name}" + assert output_file.exists() + + +def test_scrub_command_recursive_pdf_success(output_dir): + """ + Test the 'scrub' command with recursive directory processing for PDF. + """ + result = runner.invoke( + app, ["scrub", PDF_DIR, "-r", "-ext", "pdf", "--output", str(output_dir)] + ) + + assert result.exit_code == 0, f"Failed with: {result.stdout}" + # Check at least one output file was created + output_files = list(output_dir.glob("processed_*.pdf")) + assert len(output_files) > 0 + + +def test_scrub_command_pdf_dry_run(output_dir): + """ + Test that --dry-run doesn't create PDF files. + """ + result = runner.invoke( + app, ["scrub", PDF_TEST_FILE, "--output", str(output_dir), "--dry-run"] + ) + + assert result.exit_code == 0, f"Failed with: {result.stdout}" + assert "DRY-RUN" in result.stdout + # No files should be created in dry-run mode + output_file = output_dir / f"processed_{Path(PDF_TEST_FILE).name}" + assert not output_file.exists() + + +def test_scrub_command_pdf_with_workers(output_dir): + """ + Test the --workers option for concurrent PDF processing. + """ + result = runner.invoke( + app, + [ + "scrub", + PDF_DIR, + "-r", + "-ext", + "pdf", + "--output", + str(output_dir), + "--workers", + "2", + ], + ) + + assert result.exit_code == 0, f"Failed with: {result.stdout}" diff --git a/tests/integration/test_metadata_factory.py b/tests/integration/test_metadata_factory.py index 6136bcd2..3f3fbbc3 100644 --- a/tests/integration/test_metadata_factory.py +++ b/tests/integration/test_metadata_factory.py @@ -3,22 +3,26 @@ from pathlib import Path import pytest +from src.core.jpeg_metadata import JpegProcessor +from src.core.png_metadata import PngProcessor from src.services.metadata_factory import MetadataFactory +from src.services.pdf_handler import PDFHandler from src.utils.exceptions import MetadataNotFoundError, UnsupportedFormatError # Import path helpers from conftest -from tests.conftest import get_jpg_test_file, get_png_test_file +from tests.conftest import get_jpg_test_file, get_pdf_test_file, get_png_test_file # Test file paths (cross-platform) JPG_TEST_FILE = get_jpg_test_file() PNG_TEST_FILE = get_png_test_file() +PDF_TEST_FILE = get_pdf_test_file() # ============== Success Case Tests ============== @pytest.mark.parametrize("x", [JPG_TEST_FILE, PNG_TEST_FILE]) -def test_read_image_metadata(x): +def test_read_metadata(x): """ test for reading image metadata """ @@ -30,8 +34,12 @@ def test_read_image_metadata(x): # checks if metadata is not empty assert handler.metadata == metadata - # checks if tags_to_delete or text_keys_to_delete is not empty - assert handler.tags_to_delete is not None or handler.text_keys_to_delete is not None + if isinstance(handler, JpegProcessor | PngProcessor): + # checks if tags_to_delete or text_keys_to_delete is not empty + assert ( + handler.tags_to_delete is not None + or handler.text_keys_to_delete is not None + ) # checks if metadata is a dictionary assert isinstance(metadata, dict) @@ -146,3 +154,92 @@ def test_save_without_output_path_raises_error(): handler.wipe() with pytest.raises(ValueError): handler.save(None) + + +# ============== PDF Integration Tests ============== + + +def test_read_pdf_metadata_via_factory(): + """ + Test reading PDF metadata through MetadataFactory. + """ + assert Path(PDF_TEST_FILE).exists(), f"Test file not found: {PDF_TEST_FILE}" + handler = MetadataFactory.get_handler(PDF_TEST_FILE) + + # Verify correct handler type is returned + assert isinstance(handler, PDFHandler) + + metadata = handler.read() + assert handler.metadata == metadata + assert isinstance(metadata, dict) + assert handler.keys_to_delete is not None + + +def test_wipe_pdf_metadata_via_factory(): + """ + Test wiping PDF metadata through MetadataFactory. + """ + assert Path(PDF_TEST_FILE).exists(), f"Test file not found: {PDF_TEST_FILE}" + handler = MetadataFactory.get_handler(PDF_TEST_FILE) + metadata = handler.read() + handler.wipe() + + assert handler.processed_metadata != metadata + + +def test_save_processed_pdf_metadata_via_factory(): + """ + Test saving processed PDF metadata through MetadataFactory. + """ + output_dir = Path("./tests/assets/output") + output_dir.mkdir(parents=True, exist_ok=True) + + handler = MetadataFactory.get_handler(PDF_TEST_FILE) + metadata = handler.read() + handler.wipe() + + assert handler.processed_metadata != metadata + + output_file = output_dir / Path(PDF_TEST_FILE).name + handler.save(str(output_file)) + + assert output_file.exists() + shutil.rmtree(output_dir) + + +def test_pdf_format_detection_via_factory(): + """ + Test that format detection returns 'pdf' for PDF files. + """ + handler = MetadataFactory.get_handler(PDF_TEST_FILE) + detected = handler._detect_format() + assert detected == "pdf" + + +def test_output_pdf_file_has_less_metadata(): + """ + Test that the output PDF file has metadata stripped. + """ + output_dir = Path("./tests/assets/output") + output_dir.mkdir(parents=True, exist_ok=True) + + # Process original file + handler = MetadataFactory.get_handler(PDF_TEST_FILE) + original_metadata = handler.read() + original_count = len(original_metadata) + handler.wipe() + + # Save processed file + output_file = output_dir / Path(PDF_TEST_FILE).name + handler.save(str(output_file)) + + # Read output file and verify metadata is reduced or gone + try: + output_handler = MetadataFactory.get_handler(str(output_file)) + output_metadata = output_handler.read() + assert len(output_metadata) < original_count + except MetadataNotFoundError: + # If no metadata found, that's expected for fully stripped files + pass + + shutil.rmtree(output_dir) diff --git a/tests/unit/test_pdf_handler.py b/tests/unit/test_pdf_handler.py new file mode 100644 index 00000000..55001c07 --- /dev/null +++ b/tests/unit/test_pdf_handler.py @@ -0,0 +1,223 @@ +""" +Unit tests for PDFHandler. + +Tests the PDFHandler class in isolation, focusing on individual methods +and edge cases including encrypted PDFs, missing metadata, and corrupted files. +""" + +import shutil +from pathlib import Path + +import pytest +from pypdf.errors import PdfReadError + +from src.services.pdf_handler import PDFHandler +from src.utils.exceptions import ( + MetadataNotFoundError, + MetadataReadingError, + UnsupportedFormatError, +) + +# Import path helpers from conftest +from tests.conftest import get_large_pdf_test_file, get_pdf_test_file + +# Test file paths (cross-platform) +PDF_TEST_FILE = get_pdf_test_file() +LARGE_PDF_TEST_FILE = get_large_pdf_test_file() + + +# ============== Success Case Tests ============== + + +@pytest.mark.parametrize("pdf_file", [PDF_TEST_FILE, LARGE_PDF_TEST_FILE]) +def test_read_pdf_metadata(pdf_file): + """ + Test reading metadata from PDF files. + Verifies that read() extracts metadata and populates keys_to_delete. + """ + assert Path(pdf_file).exists(), f"Test file not found: {pdf_file}" + handler = PDFHandler(pdf_file) + metadata = handler.read() + + # Check metadata was extracted + assert handler.metadata == metadata + assert isinstance(metadata, dict) + + # Check keys_to_delete is populated + assert handler.keys_to_delete is not None + assert len(handler.keys_to_delete) > 0 + + +@pytest.mark.parametrize("pdf_file", [PDF_TEST_FILE, LARGE_PDF_TEST_FILE]) +def test_wipe_pdf_metadata(pdf_file): + """ + Test wiping metadata from PDF files. + Verifies that wipe() removes metadata entries. + """ + assert Path(pdf_file).exists(), f"Test file not found: {pdf_file}" + handler = PDFHandler(pdf_file) + metadata = handler.read() + handler.wipe() + + # processed_metadata should differ from original + assert handler.processed_metadata != metadata + + +@pytest.mark.parametrize("pdf_file", [PDF_TEST_FILE, LARGE_PDF_TEST_FILE]) +def test_save_processed_pdf_metadata(pdf_file): + """ + Test saving processed PDF to output path. + """ + output_dir = Path("./tests/assets/output") + output_dir.mkdir(parents=True, exist_ok=True) + + handler = PDFHandler(pdf_file) + handler.read() + handler.wipe() + + output_file = output_dir / Path(pdf_file).name + handler.save(str(output_file)) + + # Verify output file exists + assert output_file.exists() + + # Cleanup + shutil.rmtree(output_dir) + + +def test_format_detection_works(): + """ + Test that _detect_format() correctly identifies PDF files. + """ + handler = PDFHandler(PDF_TEST_FILE) + detected = handler._detect_format() + assert detected == "pdf" + + +@pytest.mark.parametrize("pdf_file", [PDF_TEST_FILE, LARGE_PDF_TEST_FILE]) +def test_output_file_has_less_metadata(pdf_file): + """ + Test that the output file has metadata stripped. + """ + output_dir = Path("./tests/assets/output") + output_dir.mkdir(parents=True, exist_ok=True) + + # Process original file + handler = PDFHandler(pdf_file) + original_metadata = handler.read() + original_count = len(original_metadata) + handler.wipe() + + # Save processed file + output_file = output_dir / Path(pdf_file).name + handler.save(str(output_file)) + + # Read output file and verify metadata is reduced or gone + try: + output_handler = PDFHandler(str(output_file)) + output_metadata = output_handler.read() + assert len(output_metadata) < original_count + except MetadataNotFoundError: + # If no metadata found, that's expected for fully stripped files + pass + + # Cleanup + shutil.rmtree(output_dir) + + +# ============== Error Case Tests ============== + + +def test_unsupported_format_raises_error(tmp_path): + """ + Test that non-PDF files raise UnsupportedFormatError. + """ + # Create a fake text file with .txt extension + fake_file = tmp_path / "test.txt" + fake_file.write_text("not a pdf") + + handler = PDFHandler(str(fake_file)) + with pytest.raises(UnsupportedFormatError): + handler._detect_format() + + +def test_save_without_output_path_raises_error(): + """ + Test that save() raises ValueError when output_path is None. + """ + handler = PDFHandler(PDF_TEST_FILE) + handler.read() + handler.wipe() + with pytest.raises(ValueError): + handler.save(None) + + +# ============== Edge Case Tests ============== + + +def test_encrypted_pdf_raises_error(tmp_path): + """ + Test that encrypted PDFs raise MetadataReadingError. + Note: This test creates a minimal encrypted PDF for testing. + """ + # Create a minimal encrypted PDF structure + encrypted_pdf = tmp_path / "encrypted.pdf" + # Write minimal PDF with encryption marker + # This is a simplified test - in production you'd use a real encrypted PDF + encrypted_pdf.write_bytes( + b"%PDF-1.4\n" + b"1 0 obj\n<>\nendobj\n" + b"2 0 obj\n<>\nendobj\n" + b"3 0 obj\n<>\nendobj\n" + b"xref\n0 4\n" + b"0000000000 65535 f \n" + b"0000000009 00000 n \n" + b"0000000058 00000 n \n" + b"0000000115 00000 n \n" + b"trailer\n<>>>\n" + b"startxref\n191\n%%EOF" + ) + + handler = PDFHandler(str(encrypted_pdf)) + # Encrypted PDF should raise MetadataReadingError or PdfReadError + with pytest.raises((MetadataReadingError, PdfReadError, Exception)): + handler.read() + + +def test_pdf_without_metadata_raises_error(tmp_path): + """ + Test that PDFs with no metadata raise MetadataNotFoundError. + """ + # Create a minimal PDF without metadata + no_metadata_pdf = tmp_path / "no_metadata.pdf" + no_metadata_pdf.write_bytes( + b"%PDF-1.4\n" + b"1 0 obj\n<>\nendobj\n" + b"2 0 obj\n<>\nendobj\n" + b"3 0 obj\n<>\nendobj\n" + b"xref\n0 4\n" + b"0000000000 65535 f \n" + b"0000000009 00000 n \n" + b"0000000058 00000 n \n" + b"0000000115 00000 n \n" + b"trailer\n<>\n" + b"startxref\n191\n%%EOF" + ) + + handler = PDFHandler(str(no_metadata_pdf)) + with pytest.raises((MetadataNotFoundError, Exception)): + handler.read() + + +def test_corrupted_pdf_graceful_error(tmp_path): + """ + Test that corrupted PDFs are handled gracefully. + """ + # Create a corrupted PDF (invalid structure) + corrupted_pdf = tmp_path / "corrupted.pdf" + corrupted_pdf.write_bytes(b"not a valid pdf content at all") + + handler = PDFHandler(str(corrupted_pdf)) + # Should raise PdfReadError or similar from pypdf + with pytest.raises((PdfReadError, Exception)): + handler.read()