wehub-resource-sync
cddb07a176
docs / deploy (push) Has been cancelled
docs / changes (push) Has been cancelled
docs / check-and-build (push) Has been cancelled
build container image / cpu (push) Has been cancelled
build container image / cuda (push) Has been cancelled
build container image / rocm (push) Has been cancelled
frontend checks / frontend-checks (push) Has been cancelled
frontend tests / frontend-tests (push) Has been cancelled
lfs checks / lfs-check (push) Has been cancelled
python checks / python-checks (push) Has been cancelled
python tests / py3.12: macos-default (push) Has been cancelled
python tests / py3.11: windows-cpu (push) Has been cancelled
python tests / py3.12: windows-cpu (push) Has been cancelled
python tests / py3.11: linux-cpu (push) Has been cancelled
typegen checks / typegen-checks (push) Has been cancelled
uv lock checks / uv-lock-checks (push) Has been cancelled
openapi checks / openapi-checks (push) Has been cancelled
python tests / py3.11: macos-default (push) Has been cancelled
python tests / py3.12: linux-cpu (push) Has been cancelled
265 lines
11 KiB
Python
265 lines
11 KiB
Python
"""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"])
|