E2e workflow (#20)

Reviewed-on: #20
Co-authored-by: Mathias Scheider <mathias2784@web.de>

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

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"})

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

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()