"""ImageUploadService 테스트 (W5 F23 Phase 4). image_service.py의 upload_custom_image를 ImageUploadService로 이관. """ from __future__ import annotations from types import SimpleNamespace from unittest.mock import MagicMock import pytest from app.core.errors import AppError from app.services.image_upload_service import ImageUploadService @pytest.fixture def svc(): """ImageUploadService MagicMock 주입 인스턴스.""" instance = ImageUploadService.__new__(ImageUploadService) instance._db = MagicMock() instance._project_id = "p1" instance._actor_id = "u1" instance._logger = MagicMock() return instance def test_upload_entity_not_found_raises(svc): """entity_id 주어졌으나 프로젝트에 없으면 AppError(404).""" svc._db.query.return_value.filter.return_value.first.return_value = None with pytest.raises(AppError) as exc_info: svc.upload_custom_image(b"x", "a.png", entity_id="e1") assert exc_info.value.code == "image.entity_not_found" assert exc_info.value.status_code == 404 def test_upload_still_not_found_raises(svc): """still_id 주어졌으나 프로젝트에 없으면 AppError(404).""" svc._db.query.return_value.filter.return_value.first.return_value = None with pytest.raises(AppError) as exc_info: svc.upload_custom_image(b"x", "a.png", still_id="st1") assert exc_info.value.code == "image.still_not_found" assert exc_info.value.status_code == 404 def test_upload_with_entity_saves_reference(svc, monkeypatch, tmp_path): """entity_id가 있으면 asset_type='reference', episode_id 자동 결정.""" entity = SimpleNamespace(id="e1", project_id="p1") link = SimpleNamespace(episode_id="ep1") query_call = [0] def _query(_model): query_call[0] += 1 q = MagicMock() if query_call[0] == 1: # entity ownership q.filter.return_value.first.return_value = entity elif query_call[0] == 2: # EntityEpisodeLink q.filter.return_value.first.return_value = link return q svc._db.query.side_effect = _query monkeypatch.setattr("app.services.image_upload_service.settings", SimpleNamespace(projects_dir=str(tmp_path))) monkeypatch.setattr("app.services.image_upload_service._new_id", lambda: "img-id") monkeypatch.setattr("app.services.image_upload_service._now", lambda: "2026-01-01T00:00:00Z") monkeypatch.setattr("app.services.image_upload_service.auto_set_primary", MagicMock()) monkeypatch.setattr( "app.services.image_upload_service.image_to_dict", lambda asset: { "id": asset.id, "asset_type": asset.asset_type, "episode_id": asset.episode_id, "file_path": asset.file_path, }, ) result = svc.upload_custom_image(b"bytes", "img.png", entity_id="e1", ip="1.2.3.4") added_asset = svc._db.add.call_args[0][0] assert added_asset.id == "img-id" assert added_asset.asset_type == "reference" assert added_asset.episode_id == "ep1" assert added_asset.entity_id == "e1" assert added_asset.prompt_used == "uploaded" assert added_asset.generation_model == "manual_upload" assert added_asset.status == "generated" assert str(added_asset.file_path).endswith("/reference/img-id.png") svc._db.commit.assert_called_once() log_kwargs = svc._logger.log.call_args.kwargs assert log_kwargs["action"] == "image.upload" assert log_kwargs["ip_address"] == "1.2.3.4" assert result["id"] == "img-id" def test_upload_with_still_saves_scene(svc, monkeypatch, tmp_path): """still_id가 있으면 asset_type='scene', still.episode_id 사용.""" still = SimpleNamespace(id="st1", project_id="p1", episode_id="ep2") query_call = [0] def _query(_model): query_call[0] += 1 q = MagicMock() if query_call[0] == 1: # still ownership q.filter.return_value.first.return_value = still elif query_call[0] == 2: # episode lookup via still (re-query) q.filter.return_value.first.return_value = still return q svc._db.query.side_effect = _query monkeypatch.setattr("app.services.image_upload_service.settings", SimpleNamespace(projects_dir=str(tmp_path))) monkeypatch.setattr("app.services.image_upload_service._new_id", lambda: "img-x") monkeypatch.setattr("app.services.image_upload_service._now", lambda: "2026-01-01T00:00:00Z") monkeypatch.setattr("app.services.image_upload_service.auto_set_primary", MagicMock()) monkeypatch.setattr( "app.services.image_upload_service.image_to_dict", lambda asset: {"id": asset.id, "asset_type": asset.asset_type, "episode_id": asset.episode_id}, ) result = svc.upload_custom_image(b"bytes", "a.jpg", still_id="st1") added_asset = svc._db.add.call_args[0][0] assert added_asset.asset_type == "scene" assert added_asset.episode_id == "ep2" assert added_asset.still_id == "st1" assert str(added_asset.file_path).endswith("/scene/img-x.jpg") assert result["asset_type"] == "scene" def test_upload_fallback_episode_shared(svc, monkeypatch, tmp_path): """entity_id/still_id 둘 다 없으면 episode_id='shared'.""" svc._db.query.return_value.filter.return_value.first.return_value = None monkeypatch.setattr("app.services.image_upload_service.settings", SimpleNamespace(projects_dir=str(tmp_path))) monkeypatch.setattr("app.services.image_upload_service._new_id", lambda: "img-n") monkeypatch.setattr("app.services.image_upload_service._now", lambda: "2026-01-01T00:00:00Z") monkeypatch.setattr("app.services.image_upload_service.auto_set_primary", MagicMock()) monkeypatch.setattr("app.services.image_upload_service.image_to_dict", lambda a: {"id": a.id}) svc.upload_custom_image(b"bytes", "n") added_asset = svc._db.add.call_args[0][0] assert added_asset.episode_id == "shared" assert added_asset.asset_type == "scene" # entity_id 없으므로 scene def test_upload_entity_without_link_defaults_to_shared(svc, monkeypatch, tmp_path): """entity는 있으나 EntityEpisodeLink 없으면 episode_id='shared'.""" entity = SimpleNamespace(id="e1", project_id="p1") query_call = [0] def _query(_model): query_call[0] += 1 q = MagicMock() if query_call[0] == 1: q.filter.return_value.first.return_value = entity else: # EntityEpisodeLink — 없음 q.filter.return_value.first.return_value = None return q svc._db.query.side_effect = _query monkeypatch.setattr("app.services.image_upload_service.settings", SimpleNamespace(projects_dir=str(tmp_path))) monkeypatch.setattr("app.services.image_upload_service._new_id", lambda: "img-z") monkeypatch.setattr("app.services.image_upload_service._now", lambda: "2026-01-01T00:00:00Z") monkeypatch.setattr("app.services.image_upload_service.auto_set_primary", MagicMock()) monkeypatch.setattr("app.services.image_upload_service.image_to_dict", lambda a: {"id": a.id}) svc.upload_custom_image(b"bytes", "x.png", entity_id="e1") added_asset = svc._db.add.call_args[0][0] assert added_asset.episode_id == "shared" assert added_asset.asset_type == "reference" def test_upload_extension_preserves_known_suffix(svc, monkeypatch, tmp_path): """filename 확장자 보존 (.jpg → .jpg).""" svc._db.query.return_value.filter.return_value.first.return_value = None monkeypatch.setattr("app.services.image_upload_service.settings", SimpleNamespace(projects_dir=str(tmp_path))) monkeypatch.setattr("app.services.image_upload_service._new_id", lambda: "id-jpg") monkeypatch.setattr("app.services.image_upload_service._now", lambda: "2026-01-01T00:00:00Z") monkeypatch.setattr("app.services.image_upload_service.auto_set_primary", MagicMock()) monkeypatch.setattr("app.services.image_upload_service.image_to_dict", lambda a: {"id": a.id}) svc.upload_custom_image(b"bytes", "pic.jpg") assert str(svc._db.add.call_args[0][0].file_path).endswith("/id-jpg.jpg") def test_upload_no_extension_defaults_to_png(svc, monkeypatch, tmp_path): """확장자 없으면 .png 기본.""" svc._db.query.return_value.filter.return_value.first.return_value = None monkeypatch.setattr("app.services.image_upload_service.settings", SimpleNamespace(projects_dir=str(tmp_path))) monkeypatch.setattr("app.services.image_upload_service._new_id", lambda: "noext") monkeypatch.setattr("app.services.image_upload_service._now", lambda: "2026-01-01T00:00:00Z") monkeypatch.setattr("app.services.image_upload_service.auto_set_primary", MagicMock()) monkeypatch.setattr("app.services.image_upload_service.image_to_dict", lambda a: {"id": a.id}) svc.upload_custom_image(b"bytes", "noextension") assert str(svc._db.add.call_args[0][0].file_path).endswith("/noext.png") def test_image_service_facade_injects_upload_svc(): """ImageService.__init__가 _upload_svc를 주입하고 delegate가 동일 함수 호출.""" from app.services.image_service import ImageService facade = ImageService.__new__(ImageService) facade._db = MagicMock() facade._project_id = "p1" facade._actor_id = "u1" facade._logger = MagicMock() facade._upload_svc = MagicMock(spec=ImageUploadService) facade.upload_custom_image( b"bytes", "a.png", entity_id="e", still_id="s", episode_id="ep", ip="1.2.3.4", ) facade._upload_svc.upload_custom_image.assert_called_once_with( file_bytes=b"bytes", filename="a.png", entity_id="e", still_id="s", episode_id="ep", ip="1.2.3.4", )