"""Tests for ImageService (images_default.py). Covers subfolder forwarding for all strategies and the delete_images_on_board silent-failure contract (Points 2 & 3 from PR review). """ from unittest.mock import MagicMock import pytest from PIL import Image from invokeai.app.services.image_records.image_records_common import ( ImageCategory, ImageRecord, ResourceOrigin, ) from invokeai.app.services.images.images_default import ImageService from invokeai.app.util.misc import get_iso_timestamp @pytest.fixture def image_service() -> ImageService: svc = ImageService() invoker = MagicMock() # Wire up service references invoker.services.names.create_image_name.return_value = "abc12345-test.png" invoker.services.image_records.get.return_value = _make_record(image_subfolder="some/sub") invoker.services.board_image_records.get_board_for_image.return_value = None invoker.services.urls.get_image_url.return_value = "http://localhost/img.png" invoker.services.configuration.image_subfolder_strategy = "flat" svc.start(invoker) return svc def _make_record( image_name: str = "abc12345-test.png", image_subfolder: str = "", is_intermediate: bool = False, ) -> ImageRecord: now = get_iso_timestamp() return ImageRecord( image_name=image_name, image_origin=ResourceOrigin.INTERNAL, image_category=ImageCategory.GENERAL, width=64, height=64, created_at=now, updated_at=now, is_intermediate=is_intermediate, starred=False, has_workflow=False, image_subfolder=image_subfolder, ) # ── Point 2: subfolder forwarding tests ── class TestCreateSubfolderForwarding: """Verify that create() computes and forwards the correct subfolder for each strategy.""" @pytest.mark.parametrize( "strategy_name,expected_subfolder", [ ("flat", ""), ("type", "general"), ("hash", "ab"), # first 2 chars of "abc12345-test.png" ], ids=["flat", "type", "hash"], ) def test_create_forwards_subfolder(self, image_service: ImageService, strategy_name: str, expected_subfolder: str): invoker = image_service._ImageService__invoker # type: ignore invoker.services.configuration.image_subfolder_strategy = strategy_name # Make get_dto work by returning a record with the expected subfolder invoker.services.image_records.get.return_value = _make_record(image_subfolder=expected_subfolder) image = Image.new("RGB", (64, 64)) image_service.create( image=image, image_origin=ResourceOrigin.INTERNAL, image_category=ImageCategory.GENERAL, ) # Assert image_records.save was called with the right subfolder save_call = invoker.services.image_records.save save_call.assert_called_once() assert save_call.call_args.kwargs["image_subfolder"] == expected_subfolder # Assert image_files.save was called with the same subfolder file_save = invoker.services.image_files.save file_save.assert_called_once() assert file_save.call_args.kwargs["image_subfolder"] == expected_subfolder def test_create_date_strategy_produces_date_subfolder(self, image_service: ImageService): invoker = image_service._ImageService__invoker # type: ignore invoker.services.configuration.image_subfolder_strategy = "date" invoker.services.image_records.get.return_value = _make_record(image_subfolder="2026/04/05") image = Image.new("RGB", (64, 64)) image_service.create( image=image, image_origin=ResourceOrigin.INTERNAL, image_category=ImageCategory.GENERAL, ) subfolder = invoker.services.image_records.save.call_args.kwargs["image_subfolder"] # Date strategy should produce YYYY/MM/DD format parts = subfolder.split("/") assert len(parts) == 3 assert all(p.isdigit() for p in parts) def test_create_type_strategy_intermediate(self, image_service: ImageService): invoker = image_service._ImageService__invoker # type: ignore invoker.services.configuration.image_subfolder_strategy = "type" invoker.services.image_records.get.return_value = _make_record(image_subfolder="intermediate") image = Image.new("RGB", (64, 64)) image_service.create( image=image, image_origin=ResourceOrigin.INTERNAL, image_category=ImageCategory.GENERAL, is_intermediate=True, ) subfolder = invoker.services.image_records.save.call_args.kwargs["image_subfolder"] assert subfolder == "intermediate" class TestReadOperationsForwardSubfolder: """Verify that read operations look up the record and forward image_subfolder.""" def test_get_pil_image(self, image_service: ImageService): invoker = image_service._ImageService__invoker # type: ignore invoker.services.image_records.get.return_value = _make_record(image_subfolder="2026/01/01") image_service.get_pil_image("test.png") invoker.services.image_files.get.assert_called_once_with("test.png", image_subfolder="2026/01/01") def test_get_workflow(self, image_service: ImageService): invoker = image_service._ImageService__invoker # type: ignore invoker.services.image_records.get.return_value = _make_record(image_subfolder="general") image_service.get_workflow("test.png") invoker.services.image_files.get_workflow.assert_called_once_with("test.png", image_subfolder="general") def test_get_graph(self, image_service: ImageService): invoker = image_service._ImageService__invoker # type: ignore invoker.services.image_records.get.return_value = _make_record(image_subfolder="general") image_service.get_graph("test.png") invoker.services.image_files.get_graph.assert_called_once_with("test.png", image_subfolder="general") def test_get_path(self, image_service: ImageService): invoker = image_service._ImageService__invoker # type: ignore invoker.services.image_records.get.return_value = _make_record(image_subfolder="ab") image_service.get_path("test.png") invoker.services.image_files.get_path.assert_called_once_with("test.png", False, image_subfolder="ab") def test_get_path_thumbnail(self, image_service: ImageService): invoker = image_service._ImageService__invoker # type: ignore invoker.services.image_records.get.return_value = _make_record(image_subfolder="ab") image_service.get_path("test.png", thumbnail=True) invoker.services.image_files.get_path.assert_called_once_with("test.png", True, image_subfolder="ab") class TestDeleteForwardsSubfolder: """Verify that delete operations forward image_subfolder.""" def test_delete_forwards_subfolder(self, image_service: ImageService): invoker = image_service._ImageService__invoker # type: ignore invoker.services.image_records.get.return_value = _make_record(image_subfolder="2026/04/05") image_service.delete("test.png") invoker.services.image_files.delete.assert_called_once_with("test.png", image_subfolder="2026/04/05") invoker.services.image_records.delete.assert_called_once_with("test.png") def test_delete_intermediates_forwards_subfolder(self, image_service: ImageService): invoker = image_service._ImageService__invoker # type: ignore invoker.services.image_records.delete_intermediates.return_value = [ ("img1.png", "intermediate"), ("img2.png", "intermediate"), ] count = image_service.delete_intermediates() assert count == 2 calls = invoker.services.image_files.delete.call_args_list assert calls[0].args == ("img1.png",) assert calls[0].kwargs == {"image_subfolder": "intermediate"} assert calls[1].args == ("img2.png",) assert calls[1].kwargs == {"image_subfolder": "intermediate"} # ── Point 3: delete_images_on_board silent-failure contract ── class TestDeleteImagesOnBoardContract: """Tests for the silent-failure behavior of delete_images_on_board.""" def test_db_rows_deleted_even_when_file_delete_fails(self, image_service: ImageService): """Current behavior: DB rows are deleted even if file cleanup fails for some images. This test documents the contract so any change is intentional.""" invoker = image_service._ImageService__invoker # type: ignore invoker.services.board_image_records.get_all_board_image_names_for_board.return_value = [ "good.png", "bad.png", ] # First image record lookup succeeds, second fails good_record = _make_record(image_name="good.png", image_subfolder="general") bad_record = _make_record(image_name="bad.png", image_subfolder="bad/path") invoker.services.image_records.get.side_effect = [good_record, bad_record] # File delete succeeds for first, fails for second invoker.services.image_files.delete.side_effect = [None, Exception("disk error")] image_service.delete_images_on_board("board-1") # DB rows are still deleted for all images invoker.services.image_records.delete_many.assert_called_once_with(["good.png", "bad.png"]) def test_file_cleanup_failure_does_not_raise(self, image_service: ImageService): """File cleanup errors are swallowed, not propagated.""" invoker = image_service._ImageService__invoker # type: ignore invoker.services.board_image_records.get_all_board_image_names_for_board.return_value = ["img.png"] record = _make_record(image_name="img.png", image_subfolder="sub") invoker.services.image_records.get.return_value = record invoker.services.image_files.delete.side_effect = Exception("permission denied") # Should not raise image_service.delete_images_on_board("board-1") # DB delete still happens invoker.services.image_records.delete_many.assert_called_once() def test_record_lookup_failure_does_not_block_others(self, image_service: ImageService): """If getting the record for one image fails, other images are still processed.""" invoker = image_service._ImageService__invoker # type: ignore invoker.services.board_image_records.get_all_board_image_names_for_board.return_value = [ "missing.png", "ok.png", ] ok_record = _make_record(image_name="ok.png", image_subfolder="") invoker.services.image_records.get.side_effect = [Exception("not found"), ok_record] image_service.delete_images_on_board("board-1") # File delete was attempted for the second image only invoker.services.image_files.delete.assert_called_once_with("ok.png", image_subfolder="") # DB rows are still deleted for all invoker.services.image_records.delete_many.assert_called_once_with(["missing.png", "ok.png"])