From ad7dbfdd11ee491deeebaa549a69ce1c3160c787 Mon Sep 17 00:00:00 2001 From: Mathias Scheider Date: Sun, 18 Oct 2026 07:24:49 +0000 Subject: [PATCH] E2e workflow (#20) Reviewed-on: https://gitea.schneiderserver.site/mathias2784/photools3/pulls/20 Co-authored-by: Mathias Scheider --- README.md | 86 ++++++++++++ src/photools3/cli.py | 135 ++++++++++++++---- src/photools3/llm_session.py | 43 ++++++ src/photools3/metadata_writer.py | 33 ++++- src/photools3/models.py | 36 +++++ src/photools3/photo.py | 14 +- src/photools3/processing.py | 128 +++++++++++++++++ test/test_cli.py | 233 ++++++++++++++++++++++--------- test/test_metadata_writer.py | 30 ++++ test/test_photo.py | 9 ++ test/test_processing.py | 129 +++++++++++++++++ 11 files changed, 775 insertions(+), 101 deletions(-) create mode 100644 src/photools3/processing.py create mode 100644 test/test_processing.py diff --git a/README.md b/README.md index e69de29..971625d 100644 --- a/README.md +++ b/README.md @@ -0,0 +1,86 @@ +# PhotoTools3 + +PhotoTools3 uses an OpenAI-compatible vision API to generate photo captions and +keywords, then writes the results to photo metadata. It can also assess photo +quality. + +## Setup + +Install the Python dependencies with `uv` and make sure ExifTool is installed and +available on `PATH` (it is a system dependency, not a Python package): + +```bash +uv sync +``` + +Create a `.env` file in the project root with the LLM connection details and the +prefix used to identify metadata written by PhotoTools3: + +```dotenv +llm_api_base_url=https://your-openai-compatible-api/v1 +llm_api_key=your-api-key +llm_default_temperature=0.2 +metadata_prefix=photools3: +``` + +Keyword suggestions are restricted to the vocabulary in `config/keywords.txt`. + +## Generate metadata + +The `metadata-magic` command accepts one or more image paths and/or directories. +The default mode is `both`, which generates and writes a caption and keywords: + +```bash +uv run photools3 metadata-magic photo.dng +``` + +Choose a mode with `--mode`: + +```bash +# Generate only keywords +uv run photools3 metadata-magic photo.dng --mode keywords + +# Generate only a caption +uv run photools3 metadata-magic photo.dng --mode captions + +# Explicitly request both (the default) +uv run photools3 metadata-magic photo.dng --mode both +``` + +Pass multiple files to process them sequentially as a batch: + +```bash +uv run photools3 metadata-magic photo1.dng photo2.jpg --mode both +``` + +Directories are scanned at the top level by default. Add `--recursive` to +include supported images in subdirectories. You can combine directories and +individual file paths: + +```bash +uv run photools3 metadata-magic ./photos --recursive --mode keywords +uv run photools3 metadata-magic ./photos photo-to-include.dng --mode captions +``` + +The command reports each processed file and exits with a non-zero status if any +file fails. In a batch, it continues processing the remaining files after an +individual failure. Generated metadata is written using ExifTool and XMP +sidecars. + +## Assess photo quality + +The `quality` command reports an assessment without changing photo metadata: + +```bash +uv run photools3 quality photo.dng +``` + +## Help + +Show the available commands and options with: + +```bash +uv run photools3 --help +uv run photools3 metadata-magic --help +uv run photools3 quality --help +``` diff --git a/src/photools3/cli.py b/src/photools3/cli.py index 11a7590..971b7c0 100644 --- a/src/photools3/cli.py +++ b/src/photools3/cli.py @@ -3,13 +3,16 @@ from pathlib import Path from typing import Annotated import typer +from PIL import Image from utils.logging_config import LoggingSettings -from photools3.llm_session import ( - LlmSessionCaptionsAndKeywords, - LlmSessionQuality, -) +from photools3.llm_session import LlmSessionQuality from photools3.photo import Photo +from photools3.processing import ( + ProcessingMode, + process_batch, + process_photo, +) app = typer.Typer(help="PhotoTools3: A CLI for image processing.") LoggingSettings() @@ -21,9 +24,7 @@ def main( ctx: typer.Context, version: bool | None = typer.Option(None, "--version", is_eager=True), ): - """ - Main entry point for PhotoTools3 CLI. - """ + """Main entry point for PhotoTools3 CLI.""" if version: typer.echo("photools3 version 0.1.0") raise typer.Exit() @@ -33,36 +34,118 @@ def main( raise typer.Exit() -@app.command("keywords") -def keywords( - file_path: Annotated[Path, typer.Argument(..., help="Path to the image file")], +@app.command("metadata-magic") +def metadata_magic( + paths: Annotated[ + list[Path], typer.Argument(..., help="Photo file(s) and/or directory paths") + ], + mode: Annotated[ + ProcessingMode, + typer.Option( + "--mode", + case_sensitive=False, + help="Metadata to generate and write", + ), + ] = ProcessingMode.BOTH, + recursive: Annotated[ + bool, + typer.Option( + "--recursive/--no-recursive", + help="Recursively discover photos inside input directories", + ), + ] = False, ): - session = LlmSessionCaptionsAndKeywords() - _run_llm_task(file_path, session) + """Generate and write keywords, captions, or both for photos.""" + _run_processing(paths, mode, recursive) @app.command("quality") def quality( file_path: Annotated[Path, typer.Argument(..., help="Path to the image file")], ): - session = LlmSessionQuality() - _run_llm_task(file_path, session) - - -def _run_llm_task( - file_path: Path, session: LlmSessionCaptionsAndKeywords | LlmSessionQuality -): + """Assess a photo's quality without changing its metadata.""" try: - logger.info(f"Processing: {file_path}") - photo = Photo(file_path) - result = session.call([photo]) - logger.info(result) + with LlmSessionQuality() as session, Photo(file_path) as photo: + result = session.call([photo]) + logger.info("Quality assessment for %s: %s", file_path, result) except FileNotFoundError: - logger.error(f"Error: File '{file_path}' does not exist.") + logger.error("Photo not found: %s", file_path) + raise typer.Exit(code=1) from None + except Exception as exc: + logger.error("Error assessing photo %s: %s", file_path, exc) + raise typer.Exit(code=1) from exc + + +def _run_processing(inputs: list[Path], mode: ProcessingMode, recursive: bool) -> None: + photo_paths = _expand_photo_paths(inputs, recursive) + if not photo_paths: + logger.error("No supported photo files found in the given input paths.") raise typer.Exit(code=1) - except Exception as e: # noqa: BLE001 + + logger.info("Processing %d photo(s) in %s mode", len(photo_paths), mode.value) + if len(photo_paths) == 1: + path = photo_paths[0] + try: + process_photo(path, mode) + except FileNotFoundError: + logger.error("Photo not found: %s", path) + raise typer.Exit(code=1) from None + except Exception as exc: + logger.error("Failed to process %s: %s", path, exc) + raise typer.Exit(code=1) from exc + logger.info("Processed: %s", path) + return + + try: + outcomes = process_batch(photo_paths, mode) + except Exception as exc: + logger.error("Batch processing failed: %s", exc) + raise typer.Exit(code=1) from exc + + failures = 0 + for outcome in outcomes: + if outcome.succeeded: + logger.info("Processed: %s", outcome.path) + else: + failures += 1 + logger.error("Failed to process %s: %s", outcome.path, outcome.error) + + logger.info( + "Batch complete: %d succeeded, %d failed", len(outcomes) - failures, failures + ) + if failures: raise typer.Exit(code=1) - logger.error(f"Error processing image: {e}") + + +def _expand_photo_paths(inputs: list[Path], recursive: bool) -> list[Path]: + """Expand directories to supported image files, preserving input order.""" + supported_extensions = { + ".dng", + *(extension.lower() for extension in Image.registered_extensions()), + } + discovered: list[Path] = [] + + for path in inputs: + if not path.is_dir(): + discovered.append(path) + continue + + try: + candidates = path.rglob("*") if recursive else path.iterdir() + discovered.extend( + sorted( + candidate + for candidate in candidates + if candidate.is_file() + and candidate.suffix.lower() in supported_extensions + ) + ) + except OSError as exc: + logger.error("Could not read directory %s: %s", path, exc) + raise typer.Exit(code=1) from exc + + # Avoid processing the same exact path more than once while retaining order. + return list(dict.fromkeys(discovered)) if __name__ == "__main__": diff --git a/src/photools3/llm_session.py b/src/photools3/llm_session.py index c79a19f..14922af 100644 --- a/src/photools3/llm_session.py +++ b/src/photools3/llm_session.py @@ -12,8 +12,10 @@ from photools3.models import ( LlmMessageType, LlmRequest, LlmRequestMessage, + LlmResponseCaption, LlmResponseCaptionsAndKeywords, LlmResponseGeneral, + LlmResponseKeywords, LlmResponseQuality, ) from photools3.photo import Photo @@ -63,6 +65,15 @@ class LlmSession: api_key=self.api_key, ) + def close(self): + self.client.close() + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, traceback): + self.close() + def client_chat_completion_create(self, request: LlmRequest, choice_no: int = 0): response = self.client.chat.completions.create(**request.model_dump()) return response.choices[choice_no].message.content @@ -171,6 +182,38 @@ class LlmSessionCaptionsAndKeywords(LlmSession): ) +class LlmSessionCaption(LlmSession): + response_model = LlmResponseCaption + prompt = "Write a caption for the following photo" + + def __init__(self, temperature: float = settings.llm_default_temperature): + super().__init__( + system_prompt=( + "You are a helpful photographer's assistant that writes captions " + "for photos. Answer in German.\n" + "Output must be a JSON of the following schema:\n" + f"{self.response_model.model_json_schema()}" + ), + temperature=temperature, + ) + + +class LlmSessionKeywords(LlmSession): + response_model = LlmResponseKeywords + prompt = "Choose appropriate keywords for the following photo" + + def __init__(self, temperature: float = settings.llm_default_temperature): + super().__init__( + system_prompt=( + "You are a helpful photographer's assistant that chooses keywords " + "for photos. Answer in German.\n" + "Output must be a JSON of the following schema:\n" + f"{self.response_model.model_json_schema()}" + ), + temperature=temperature, + ) + + class LlmSessionQuality(LlmSession): response_model = LlmResponseQuality prompt = "Assess the quality of the following photos to help me decide which ones to sort out" diff --git a/src/photools3/metadata_writer.py b/src/photools3/metadata_writer.py index 7370838..4df2a87 100644 --- a/src/photools3/metadata_writer.py +++ b/src/photools3/metadata_writer.py @@ -87,12 +87,12 @@ class MetadataWriter: return metadata def get_caption(self) -> str: - result = self.metadata.get(XMP_KEYS_CAPTION[0], "") + result = (self.metadata or {}).get(XMP_KEYS_CAPTION[0], "") assert isinstance(result, str) return result def get_keywords(self) -> list[str]: - result = self.metadata.get(XMP_KEYS_KEYWORDS[0], []) or [] + result = (self.metadata or {}).get(XMP_KEYS_KEYWORDS[0], []) or [] if isinstance(result, str): result = [result] assert isinstance(result, list) @@ -108,8 +108,11 @@ class MetadataWriter: tags=metadata_tags, params=["-overwrite_original"], ) - except Exception as exc: # noqa: BLE001 + except Exception as exc: log.error(f"Could not write metadata to file {self.sidecar_path}: {exc}") + raise RuntimeError( + f"Could not write metadata to file {self.sidecar_path}" + ) from exc def write_caption( self, caption: Caption, include_prefix: bool = True, override: bool = True @@ -151,9 +154,27 @@ class MetadataWriter: keywords: Keywords | None = None, color: Color | None = None, ): + metadata_tags: XmpMetadata = {} if caption is not None: - self.write_caption(caption) + existing_no_prefix, _ = _separate_prefixed_caption(self.get_caption()) + if existing_no_prefix: + existing_no_prefix = ( + f"{existing_no_prefix}." + if not existing_no_prefix.endswith(".") + else existing_no_prefix + ) + existing_no_prefix = f"{existing_no_prefix} " + caption = f"{existing_no_prefix}{_prefixed(caption)}" + metadata_tags.update({key: caption for key in XMP_KEYS_CAPTION}) + if keywords is not None: - self.write_keywords(keywords) + keywords = list(map(_prefixed, keywords)) + existing = self.get_keywords() + keywords = list(set(existing + keywords)) + metadata_tags.update({tag: keywords for tag in XMP_KEYS_KEYWORDS}) + if color is not None: - self.write_color(color) + metadata_tags["XMP-darktable:Colorlabels"] = color + + if metadata_tags: + self.set_metadata(metadata_tags) diff --git a/src/photools3/models.py b/src/photools3/models.py index f3b27da..d545c00 100644 --- a/src/photools3/models.py +++ b/src/photools3/models.py @@ -86,6 +86,42 @@ class LlmResponseGeneral(BaseModel): text: str +class LlmResponseCaption(BaseModel): + caption: Annotated[ + str, + Field( + description=( + "A brief but precise description of what the picture shows. " + "No interpretation; just what is visible" + ) + ), + ] + + +class LlmResponseKeywords(BaseModel): + tags: list[str] = Field( + description=( + "A list of keywords that properly tag the photo. " + "Allowed keywords: " + + str(load_valid_keywords()) + .replace("'", '"') + .replace("{", "[") + .replace("}", "]") + ) + ) + + @field_validator("tags") + @classmethod + def validate_tags(cls, v: list[str]) -> list[str]: + valid_keywords = load_valid_keywords() + for tag in v: + if tag not in valid_keywords: + raise ValueError( + f"'{tag}' is not a valid keyword. Must be one from the configuration." + ) + return v + + class LlmResponseCaptionsAndKeywords(BaseModel): caption: Annotated[ str, diff --git a/src/photools3/photo.py b/src/photools3/photo.py index 738254f..f57b0ce 100644 --- a/src/photools3/photo.py +++ b/src/photools3/photo.py @@ -1,6 +1,7 @@ import base64 from io import BytesIO from pathlib import Path +from typing import Self from PIL import Image @@ -17,6 +18,15 @@ class Photo: self.path = path self.image = Image.open(path) + def close(self): + self.image.close() + + def __enter__(self) -> Self: + return self + + def __exit__(self, exc_type, exc_value, traceback): + self.close() + def resize(self, image_max_size=1024) -> Photo: """Limiting max dimensions while maintaining aspect ratio""" self.image.thumbnail((image_max_size, image_max_size), Image.Resampling.LANCZOS) @@ -24,7 +34,9 @@ class Photo: def to_rgb(self): if self.image.mode != "RGB": - self.image = self.image.convert("RGB") + original = self.image + self.image = original.convert("RGB") + original.close() return self def get_encoded(self): diff --git a/src/photools3/processing.py b/src/photools3/processing.py new file mode 100644 index 0000000..5e71ee1 --- /dev/null +++ b/src/photools3/processing.py @@ -0,0 +1,128 @@ +import logging +from collections.abc import Iterable +from dataclasses import dataclass +from enum import StrEnum +from pathlib import Path + +from photools3.llm_session import ( + LlmSession, + LlmSessionCaption, + LlmSessionCaptionsAndKeywords, + LlmSessionKeywords, +) +from photools3.metadata_writer import MetadataWriter +from photools3.models import ( + LlmResponseCaption, + LlmResponseCaptionsAndKeywords, + LlmResponseKeywords, +) +from photools3.photo import Photo + +logger = logging.getLogger(__name__) + +type ProcessingResponse = ( + LlmResponseCaption | LlmResponseKeywords | LlmResponseCaptionsAndKeywords +) + + +class ProcessingMode(StrEnum): + KEYWORDS = "keywords" + CAPTIONS = "captions" + BOTH = "both" + + +@dataclass(frozen=True) +class ProcessingOutcome: + path: Path + result: ProcessingResponse | None = None + error: str | None = None + + @property + def succeeded(self) -> bool: + return self.error is None and self.result is not None + + +def _create_session(mode: ProcessingMode) -> LlmSession: + if mode == ProcessingMode.KEYWORDS: + return LlmSessionKeywords() + if mode == ProcessingMode.CAPTIONS: + return LlmSessionCaption() + return LlmSessionCaptionsAndKeywords() + + +def process_photo( + path: Path | str, + mode: ProcessingMode | str = ProcessingMode.BOTH, + *, + session: LlmSession | None = None, +) -> ProcessingResponse: + """Process one photo and write the requested LLM result to its metadata. + + If a session is supplied, its lifecycle remains the caller's responsibility. + """ + path = Path(path) + mode = ProcessingMode(mode) + owns_session = session is None + active_session = session if session is not None else _create_session(mode) + + try: + with Photo(path) as photo: + result = active_session.call([photo]) + + caption: str | None = None + keywords: list[str] | None = None + if mode == ProcessingMode.CAPTIONS: + if not isinstance(result, LlmResponseCaption): + raise TypeError("Caption mode returned an unexpected response type") + caption = result.caption + elif mode == ProcessingMode.KEYWORDS: + if not isinstance(result, LlmResponseKeywords): + raise TypeError("Keywords mode returned an unexpected response type") + keywords = result.tags + else: + if not isinstance(result, LlmResponseCaptionsAndKeywords): + raise TypeError("Both mode returned an unexpected response type") + caption = result.caption + keywords = result.tags + + MetadataWriter(path).write(caption=caption, keywords=keywords) + return result + finally: + if owns_session: + active_session.close() + + +def process_batch( + paths: Iterable[Path | str], + mode: ProcessingMode | str = ProcessingMode.BOTH, +) -> list[ProcessingOutcome]: + """Process photos sequentially, reusing one LLM client for the whole batch. + + A failed photo is recorded in its outcome and does not stop later photos. + Only one decoded image is held at a time. + """ + mode = ProcessingMode(mode) + session = _create_session(mode) + outcomes: list[ProcessingOutcome] = [] + + try: + for item in paths: + path = Path(item) + try: + result = process_photo(path, mode, session=session) + except Exception as exc: + logger.exception("Failed to process %s", path) + # Keep only a string, not the exception/traceback, which can + # retain references to the just-processed image and stack. + outcomes.append( + ProcessingOutcome( + path=path, + error=f"{type(exc).__name__}: {exc}", + ) + ) + else: + outcomes.append(ProcessingOutcome(path=path, result=result)) + finally: + session.close() + + return outcomes diff --git a/test/test_cli.py b/test/test_cli.py index 643a1e7..8bab510 100644 --- a/test/test_cli.py +++ b/test/test_cli.py @@ -1,100 +1,197 @@ -import logging from pathlib import Path +from types import SimpleNamespace from unittest.mock import patch -import pytest from typer.testing import CliRunner from photools3.cli import app +from photools3.processing import ProcessingMode runner = CliRunner() -CLI_LOGGER = "photools3.cli" + +def _success_outcome(path: Path): + return SimpleNamespace(path=path, succeeded=True, error=None) -@pytest.fixture -def mock_photo_logic(tmp_path): - """Fixture to mock the core logic of keywords command.""" +def test_metadata_magic_command_defaults_to_both_for_single_photo(tmp_path): + image_path = tmp_path / "photo.jpg" + image_path.write_bytes(b"test image") + with ( - patch("photools3.cli.LlmSessionCaptionsAndKeywords") as mocked_session, - patch("photools3.cli.Photo") as mocked_photo, + patch("photools3.cli.process_photo") as process_photo, + patch("photools3.cli.process_batch") as process_batch, ): - - # Configure session mock - session_instance = mocked_session.return_value - session_instance.call.return_value = { - "caption": "A beautiful test image", - "args": ["test", "mock", "image"], # Wait, the code uses 'tags' - } - # Let me re-check the code in cli.py - # result.get("tags", []) - session_instance.call.return_value = { - "caption": "A beautiful test image", - "tags": ["test", "mock", "image"], - } - - # Configure photo mock - photo_instance = mocked_photo.return_value - photo_instance.path = tmp_path / "test_image.jpg" - - yield { - "session": session_instance, - "photo_class": mocked_photo, - "photo_instance": photo_instance, - } - - -@pytest.fixture -def dummy_file(tmp_path): - """Fixture to create a dummy file for CLI testing.""" - dummy_path = tmp_path / "test_image.jpg" - dummy_path.write_text("dummy content") - return dummy_path - - -def test_keywords_command_success(mock_photo_logic, dummy_file, caplog): - """Test the 'keywords' command with a valid file path.""" - # Ensure the mock photo instance points to our dummy file - mock_photo_logic["photo_class"].return_value.path = dummy_file - - with caplog.at_level(logging.INFO, logger=CLI_LOGGER): - result = runner.invoke(app, ["keywords", str(dummy_file)]) + result = runner.invoke(app, ["metadata-magic", str(image_path)]) assert result.exit_code == 0 - assert "Processing: " in caplog.text - # The result is logged as a raw dictionary string - assert ( - "{'caption': 'A beautiful test image', 'tags': ['test', 'mock', 'image']}" - in caplog.text + process_photo.assert_called_once_with(image_path, ProcessingMode.BOTH) + process_batch.assert_not_called() + + +def test_metadata_magic_command_selects_caption_mode(tmp_path): + image_path = tmp_path / "photo.jpg" + image_path.write_bytes(b"test image") + + with patch("photools3.cli.process_photo") as process_photo: + result = runner.invoke( + app, ["metadata-magic", str(image_path), "--mode", "captions"] + ) + + assert result.exit_code == 0 + process_photo.assert_called_once_with(image_path, ProcessingMode.CAPTIONS) + + +def test_metadata_magic_command_batches_multiple_files(tmp_path): + paths = [tmp_path / "one.jpg", tmp_path / "two.jpg"] + for path in paths: + path.write_bytes(b"test image") + + with ( + patch("photools3.cli.process_photo") as process_photo, + patch("photools3.cli.process_batch") as process_batch, + ): + process_batch.return_value = [_success_outcome(path) for path in paths] + result = runner.invoke( + app, + ["metadata-magic", *(str(path) for path in paths), "--mode", "keywords"], + ) + + assert result.exit_code == 0 + process_photo.assert_not_called() + process_batch.assert_called_once_with(paths, ProcessingMode.KEYWORDS) + + +def test_metadata_magic_command_expands_directories_recursively(tmp_path): + root = tmp_path / "photos" + nested = root / "nested" + nested.mkdir(parents=True) + top_level_photo = root / "one.jpg" + nested_photo = nested / "two.png" + sidecar = root / "one.jpg.xmp" + for path in (top_level_photo, nested_photo, sidecar): + path.write_bytes(b"placeholder") + + with ( + patch("photools3.cli.process_photo") as process_photo, + patch("photools3.cli.process_batch") as process_batch, + ): + process_batch.return_value = [ + _success_outcome(nested_photo), + _success_outcome(top_level_photo), + ] + result = runner.invoke( + app, ["metadata-magic", str(root), "--recursive", "--mode", "both"] + ) + + assert result.exit_code == 0 + process_photo.assert_not_called() + process_batch.assert_called_once_with( + [nested_photo, top_level_photo], ProcessingMode.BOTH ) -def test_keywords_command_file_not_found(caplog): - """Test the 'keywords' command with a non-existent file path.""" - non_existent_file = Path("non_existent_file.jpg") +def test_metadata_magic_command_does_not_recurse_by_default(tmp_path): + root = tmp_path / "photos" + nested = root / "nested" + nested.mkdir(parents=True) + top_level_photo = root / "one.jpg" + nested_photo = nested / "two.jpg" + top_level_photo.write_bytes(b"placeholder") + nested_photo.write_bytes(b"placeholder") - with caplog.at_level(logging.ERROR, logger=CLI_LOGGER): - result = runner.invoke(app, ["keywords", str(non_existent_file)]) + with ( + patch("photools3.cli.process_photo") as process_photo, + patch("photools3.cli.process_batch") as process_batch, + ): + result = runner.invoke(app, ["metadata-magic", str(root)]) + + assert result.exit_code == 0 + process_photo.assert_called_once_with(top_level_photo, ProcessingMode.BOTH) + process_batch.assert_not_called() + + +def test_metadata_magic_command_returns_failure_if_any_batch_item_fails(tmp_path): + paths = [tmp_path / "one.jpg", tmp_path / "two.jpg"] + for path in paths: + path.write_bytes(b"test image") + + with patch("photools3.cli.process_batch") as process_batch: + process_batch.return_value = [ + _success_outcome(paths[0]), + SimpleNamespace( + path=paths[1], succeeded=False, error="RuntimeError: write failed" + ), + ] + result = runner.invoke(app, ["metadata-magic", *(str(path) for path in paths)]) assert result.exit_code == 1 - assert f"Error: File '{non_existent_file}' does not exist." in caplog.text -def test_keywords_command_error_during_processing(mock_photo_logic, dummy_file): - """Test the 'keywords' command when an exception occurs during processing.""" - # Configure the session mock to raise an exception - mock_photo_logic["session"].call.side_effect = Exception("LLM error") +def test_metadata_magic_command_returns_failure_for_single_photo_error(tmp_path): + image_path = tmp_path / "photo.jpg" + image_path.write_bytes(b"test image") - result = runner.invoke(app, ["keywords", str(dummy_file)]) + with patch("photools3.cli.process_photo", side_effect=RuntimeError("LLM error")): + result = runner.invoke(app, ["metadata-magic", str(image_path)]) + + assert result.exit_code == 1 + + +def test_keywords_alias_is_not_registered(): + result = runner.invoke(app, ["keywords", "photo.jpg"]) + + assert result.exit_code == 2 + assert "No such command 'keywords'" in result.output + + +def test_old_process_command_name_is_not_registered(): + result = runner.invoke(app, ["process", "photo.jpg"]) + + assert result.exit_code == 2 + assert "No such command 'process'" in result.output + + +def test_metadata_magic_command_rejects_empty_directory(tmp_path): + with ( + patch("photools3.cli.process_photo") as process_photo, + patch("photools3.cli.process_batch") as process_batch, + ): + result = runner.invoke(app, ["metadata-magic", str(tmp_path)]) + + assert result.exit_code == 1 + process_photo.assert_not_called() + process_batch.assert_not_called() + + +def test_quality_command_closes_photo_and_session(tmp_path): + image_path = tmp_path / "photo.jpg" + image_path.write_bytes(b"test image") + + with ( + patch("photools3.cli.LlmSessionQuality") as session_class, + patch("photools3.cli.Photo") as photo_class, + ): + result = runner.invoke(app, ["quality", str(image_path)]) + + assert result.exit_code == 0 + session_class.return_value.__enter__.return_value.call.assert_called_once_with( + [photo_class.return_value.__enter__.return_value] + ) + session_class.return_value.__exit__.assert_called_once() + photo_class.return_value.__exit__.assert_called_once() + + +def test_quality_command_reports_error_and_exits(tmp_path): + image_path = tmp_path / "photo.jpg" + + with patch("photools3.cli.Photo", side_effect=FileNotFoundError(image_path)): + result = runner.invoke(app, ["quality", str(image_path)]) - # Note: cli.py raises typer.Exit(code=1) before logging the - # "Error processing image: ..." message (the logger.error call is dead - # code after the raise), so only the exit code can be asserted here. assert result.exit_code == 1 def test_version_command(): - """Test the --version option.""" result = runner.invoke(app, ["--version"]) assert result.exit_code == 0 diff --git a/test/test_metadata_writer.py b/test/test_metadata_writer.py index b2b96a6..0c8aed1 100644 --- a/test/test_metadata_writer.py +++ b/test/test_metadata_writer.py @@ -1,5 +1,6 @@ import shutil from pathlib import Path +from unittest.mock import patch import pytest @@ -211,3 +212,32 @@ def test_set_keywords(make_dng_file, pic, expected_keywords): def test_set_keywords_on_existing(): # TODO ... + + +def test_write_caption_and_keywords_in_one_metadata_update(tmp_path): + writer = MetadataWriter(tmp_path / "photo.jpg") + with ( + patch.object(writer, "get_caption", return_value="Original caption"), + patch.object(writer, "get_keywords", return_value=["existing"]), + patch.object(writer, "set_metadata") as set_metadata, + ): + writer.write(caption="Generated caption", keywords=["architektur"]) + + set_metadata.assert_called_once() + metadata_tags = set_metadata.call_args.args[0] + assert metadata_tags["XMP:Description"] == ( + f"Original caption. {PREFIX}Generated caption" + ) + assert set(metadata_tags["XMP:Subject"]) == { + "existing", + f"{PREFIX}architektur", + } + + +def test_set_metadata_surfaces_write_failures(tmp_path): + writer = MetadataWriter(tmp_path / "photo.jpg") + with patch("photools3.metadata_writer.exiftool.ExifToolHelper") as exif: + exif_instance = exif.return_value.__enter__.return_value + exif_instance.execute.side_effect = OSError("ExifTool failed") + with pytest.raises(RuntimeError, match="Could not write metadata"): + writer.set_metadata({"XMP:Description": "caption"}) diff --git a/test/test_photo.py b/test/test_photo.py index 0c924fe..7d8a61f 100644 --- a/test/test_photo.py +++ b/test/test_photo.py @@ -48,6 +48,15 @@ def test_load_photo(): assert isinstance(photo.image, Image.Image) +def test_photo_context_manager_closes_image(tmp_path): + path = _make_image_file(tmp_path, "RGB") + with Photo(path) as photo: + image = photo.image + assert image.fp is not None + + assert image.fp is None + + def test_resize_method_exists(): """Test that resize method exists and returns a Photo instance""" photo = Photo(PATHS_DNG_EXAMPLES[0]) diff --git a/test/test_processing.py b/test/test_processing.py new file mode 100644 index 0000000..c1a27d3 --- /dev/null +++ b/test/test_processing.py @@ -0,0 +1,129 @@ +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +from PIL import Image + +from photools3.models import ( + LlmResponseCaption, + LlmResponseCaptionsAndKeywords, + LlmResponseKeywords, +) +from photools3.processing import ( + ProcessingMode, + _create_session, + process_batch, + process_photo, +) + + +@pytest.fixture +def image_path(tmp_path): + path = tmp_path / "photo.png" + Image.new("RGB", (20, 20)).save(path) + return path + + +@pytest.mark.parametrize( + ("mode", "response", "expected_metadata"), + [ + ( + ProcessingMode.KEYWORDS, + LlmResponseKeywords(tags=["architektur"]), + {"caption": None, "keywords": ["architektur"]}, + ), + ( + ProcessingMode.CAPTIONS, + LlmResponseCaption(caption="Ein Haus"), + {"caption": "Ein Haus", "keywords": None}, + ), + ( + ProcessingMode.BOTH, + LlmResponseCaptionsAndKeywords(caption="Ein Haus", tags=["architektur"]), + {"caption": "Ein Haus", "keywords": ["architektur"]}, + ), + ], +) +def test_process_photo_writes_only_requested_metadata( + image_path, mode, response, expected_metadata +): + session = MagicMock() + session.call.return_value = response + + with ( + patch("photools3.processing._create_session", return_value=session), + patch("photools3.processing.MetadataWriter") as writer_class, + ): + result = process_photo(image_path, mode) + + assert result == response + session.call.assert_called_once() + session.close.assert_called_once() + writer_class.assert_called_once_with(image_path) + writer_class.return_value.write.assert_called_once_with(**expected_metadata) + + +def test_process_photo_does_not_close_a_caller_owned_session(image_path): + session = MagicMock() + session.call.return_value = LlmResponseCaption(caption="Ein Haus") + + with patch("photools3.processing.MetadataWriter"): + process_photo(image_path, ProcessingMode.CAPTIONS, session=session) + + session.close.assert_not_called() + + +@pytest.mark.parametrize( + ("mode", "constructor"), + [ + (ProcessingMode.KEYWORDS, "LlmSessionKeywords"), + (ProcessingMode.CAPTIONS, "LlmSessionCaption"), + (ProcessingMode.BOTH, "LlmSessionCaptionsAndKeywords"), + ], +) +def test_create_session_selects_mode_specific_response_model(mode, constructor): + with patch(f"photools3.processing.{constructor}") as session_class: + assert _create_session(mode) is session_class.return_value + + session_class.assert_called_once_with() + + +def test_process_batch_is_sequential_reuses_session_and_continues_after_error(): + active_photos = 0 + max_active_photos = 0 + + class TrackingPhoto: + def __init__(self, path): + self.path = path + + def __enter__(self): + nonlocal active_photos, max_active_photos + active_photos += 1 + max_active_photos = max(max_active_photos, active_photos) + return self + + def __exit__(self, exc_type, exc_value, traceback): + nonlocal active_photos + active_photos -= 1 + + session = MagicMock() + response = LlmResponseCaptionsAndKeywords(caption="Ein Haus", tags=["architektur"]) + session.call.side_effect = [response, RuntimeError("temporary error"), response] + paths = [Path("one.jpg"), Path("two.jpg"), Path("three.jpg")] + + with ( + patch("photools3.processing._create_session", return_value=session), + patch("photools3.processing.Photo", TrackingPhoto), + patch("photools3.processing.MetadataWriter") as writer_class, + ): + outcomes = process_batch(paths, ProcessingMode.BOTH) + + assert [outcome.succeeded for outcome in outcomes] == [True, False, True] + assert [outcome.path for outcome in outcomes] == paths + assert outcomes[1].error == "RuntimeError: temporary error" + assert max_active_photos == 1 + assert active_photos == 0 + assert session.call.call_count == 3 + assert all(len(call.args[0]) == 1 for call in session.call.call_args_list) + assert writer_class.return_value.write.call_count == 2 + session.close.assert_called_once()