""" El servicio de generación, con el proveedor reemplazado. Nada de acá habla con OpenAI: el cliente se sustituye por uno de mentira que contesta lo que cada caso necesita. Lo que se verifica es lo nuestro: qué se pide según haya foto o no, dónde queda el archivo, qué código cierra cada fallo, que un pedido terminado no se paga dos veces, y que el pedido v2 sale de un análisis terminado con el prompt congelado y las áreas limpias. """ import base64 from io import BytesIO from pathlib import Path from types import SimpleNamespace import httpx import pytest from django.core.files.uploadedfile import SimpleUploadedFile from openai import APIConnectionError, APIStatusError, APITimeoutError, RateLimitError from PIL import Image from apps.ai import prompt, services, watermark from apps.ai.markup import MarkupMode from apps.ai.models import ( AiSettings, AnalysisError, AnalysisState, GenerationError, GenerationKind, GenerationOrigin, GenerationState, ImageGeneration, ) from apps.ai.prompt import AnswerInvalid from apps.ai.watermark import Watermark from tests.factories import create_account, create_photo_analysis from tests.images import jpeg_bytes, png_bytes pytestmark = pytest.mark.django_db PNG = png_bytes(64, 48) class FakeImages: """El recurso `images` del SDK: registra la llamada y contesta lo que se le dijo.""" def __init__(self, outcome): self.outcome = outcome self.calls = [] def _answer(self, method, kwargs): self.calls.append((method, kwargs)) if isinstance(self.outcome, Exception): raise self.outcome return self.outcome def generate(self, **kwargs): return self._answer('generate', kwargs) def edit(self, **kwargs): return self._answer('edit', kwargs) def completed_response(): return SimpleNamespace( data=[SimpleNamespace(b64_json=base64.b64encode(PNG).decode())], usage=SimpleNamespace(model_dump=lambda: {'total_tokens': 42}), ) def status_error(status_code): request = httpx.Request('POST', 'https://api.openai.com/v1/images/generations') response = httpx.Response(status_code, request=request) return APIStatusError('rechazado', response=response, body=None) def rate_limit_error(): request = httpx.Request('POST', 'https://api.openai.com/v1/images/generations') response = httpx.Response(429, request=request) return RateLimitError('esperá', response=response, body=None) @pytest.fixture def media_root(settings, tmp_path): settings.AI_MEDIA_ROOT = tmp_path settings.OPENAI_API_KEY = 'sk-prueba' return tmp_path @pytest.fixture def fake_images(monkeypatch): """Instala un cliente falso y devuelve el recurso para inspeccionarlo.""" def install(outcome): images = FakeImages(outcome) monkeypatch.setattr(services.provider, 'client', lambda: SimpleNamespace(images=images)) return images return install def queued(**extra) -> ImageGeneration: return ImageGeneration.objects.create( kind=GenerationKind.POOL, origin=GenerationOrigin.TOOLS_APP, answers={'pool_shape': 'lap'}, prompt='Generate a lap-shaped pool', **extra, ) class TestExecute: def test_without_a_photo_it_creates_and_stores_the_png(self, media_root, fake_images): images = fake_images(completed_response()) generation = queued() services.execute(generation) generation.refresh_from_db() assert generation.state == GenerationState.COMPLETED assert generation.usage == { 'model': 'gpt-image-2.5-flare', 'quality': 'high', 'total_tokens': 42, } assert generation.started_at is not None and generation.finished_at is not None method, kwargs = images.calls[0] assert method == 'generate' assert kwargs['model'] == 'gpt-image-2.5-flare' assert kwargs['quality'] == 'high' assert kwargs['prompt'] == 'Generate a lap-shaped pool' # Lo que se guarda no es el PNG del proveedor sino su JPEG: es lo que # la landing enlaza y lo que no se borra nunca, y pesa diez veces menos. stored = media_root / generation.result_path assert stored.suffix == '.jpg' assert Image.open(BytesIO(stored.read_bytes())).format == 'JPEG' assert generation.result_url == f'/media/ai/{stored.name}' def test_with_a_photo_it_edits_over_it(self, media_root, fake_images): images = fake_images(completed_response()) source = media_root / 'sources' / 'patio.jpg' source.parent.mkdir(parents=True) source.write_bytes(b'foto') generation = queued(source_path='sources/patio.jpg') services.execute(generation) method, kwargs = images.calls[0] assert method == 'edit' assert 'image' in kwargs # El patio es real: el modelo de edición es el que cuida lo que ya está. # Sin nadie que haya tocado «Config», es el de la v1. assert kwargs['model'] == 'gpt-image-2.5-sunburst' assert kwargs['quality'] == 'high' def test_the_samples_travel_after_the_photo_in_the_order_the_prompt_names( self, media_root, fake_images ): """ El proveedor recibe las imágenes sin rótulo: la posición es lo único que las identifica, y el prompt dice «Image 2» contando la foto como la 1. """ images = fake_images(completed_response()) source = media_root / 'sources' / 'patio.jpg' source.parent.mkdir(parents=True) source.write_bytes(b'foto') generation = queued( source_path='sources/patio.jpg', references=[ 'ai/concrete_overlay/stone_color/irregular/gray_escape.jpg', 'ai/concrete_overlay/stone_color/irregular/shadow_stone.jpg', ], ) services.execute(generation) _method, kwargs = images.calls[0] sent = kwargs['image'] assert [Path(handle.name).name for handle in sent] == [ 'patio.jpg', 'gray_escape.jpg', 'shadow_stone.jpg', ] def test_a_request_without_samples_sends_the_photo_alone(self, media_root, fake_images): """Una lista de un elemento es otra forma de pedido: sin muestras va la foto y nada más.""" images = fake_images(completed_response()) source = media_root / 'sources' / 'patio.jpg' source.parent.mkdir(parents=True) source.write_bytes(b'foto') generation = queued(source_path='sources/patio.jpg') services.execute(generation) _method, kwargs = images.calls[0] assert not isinstance(kwargs['image'], list) def test_a_missing_sample_fails_without_calling_the_provider(self, media_root, fake_images): """ Que falte es un error nuestro, no un caso a contemplar: generar igual daría un color que nadie eligió mientras el prompt jura que la imagen está. Con su propio código, para no mandar a buscar el problema del lado del proveedor. """ images = fake_images(completed_response()) source = media_root / 'sources' / 'patio.jpg' source.parent.mkdir(parents=True) source.write_bytes(b'foto') generation = queued( source_path='sources/patio.jpg', references=['ai/concrete_overlay/stone_color/does_not_exist.jpg'], ) services.execute(generation) generation.refresh_from_db() assert generation.state == GenerationState.FAILED assert generation.error_code == GenerationError.REFERENCE_MISSING assert images.calls == [] def test_a_finished_generation_is_not_paid_twice(self, media_root, fake_images): """La cola reconoce al terminar, así que una tarea puede llegar dos veces.""" images = fake_images(completed_response()) generation = queued(state=GenerationState.COMPLETED) services.execute(generation) assert images.calls == [] def test_an_empty_answer_fails_with_its_own_code(self, media_root, fake_images): fake_images(SimpleNamespace(data=[], usage=None)) generation = queued() services.execute(generation) assert generation.state == GenerationState.FAILED assert generation.error_code == GenerationError.PROVIDER_EMPTY_RESPONSE def test_a_rejection_closes_the_generation(self, media_root, fake_images): fake_images(status_error(400)) generation = queued() services.execute(generation) assert generation.state == GenerationState.FAILED assert generation.error_code == GenerationError.PROVIDER_REJECTED @pytest.mark.parametrize( ('error', 'code'), [ (rate_limit_error(), GenerationError.PROVIDER_RATE_LIMIT), (status_error(503), GenerationError.PROVIDER_UNAVAILABLE), ( APITimeoutError(httpx.Request('POST', 'https://x')), GenerationError.PROVIDER_UNAVAILABLE, ), ( APIConnectionError(request=httpx.Request('POST', 'https://x')), GenerationError.PROVIDER_UNAVAILABLE, ), ], ) def test_a_transient_failure_goes_back_to_the_queue_for_the_task( self, media_root, fake_images, error, code ): fake_images(error) generation = queued() with pytest.raises(services.TransientProviderError) as raised: services.execute(generation) assert raised.value.code == code generation.refresh_from_db() assert generation.state == GenerationState.QUEUED assert generation.error_code == '' def test_anything_else_is_unexpected(self, media_root, fake_images): fake_images(RuntimeError('se rompió')) generation = queued() services.execute(generation) assert generation.error_code == GenerationError.UNEXPECTED class TestWatermark: """El resultado se guarda con el badge que dice la configuración al ejecutar.""" @pytest.fixture def burned_with(self, monkeypatch): """Registra con qué archivo de badge se codificó el resultado.""" seen = [] encode = services.images.encode_result def spy(raw, badge=None): seen.append(badge) return encode(raw, badge) monkeypatch.setattr(services.images, 'encode_result', spy) return seen def test_the_configured_badge_is_burned(self, media_root, fake_images, burned_with): fake_images(completed_response()) settings_row = AiSettings.current() settings_row.watermark = Watermark.CLAY settings_row.save() generation = queued() services.execute(generation) generation.refresh_from_db() assert generation.state == GenerationState.COMPLETED assert [badge.name for badge in burned_with] == ['clay.png'] def test_rotating_picks_the_badge_by_the_request(self, media_root, fake_images, burned_with): fake_images(completed_response()) settings_row = AiSettings.current() settings_row.watermark = Watermark.ROTATE settings_row.save() generation = queued() services.execute(generation) expected = watermark.badge_for(Watermark.ROTATE, generation.id) assert [badge.name for badge in burned_with] == [f'{expected}.png'] def test_a_missing_badge_fails_without_calling_the_provider( self, media_root, fake_images, monkeypatch ): images = fake_images(completed_response()) def missing(badge): raise watermark.BadgeMissing(badge) monkeypatch.setattr(services.watermark, 'badge_path', missing) generation = queued() services.execute(generation) assert images.calls == [] assert generation.state == GenerationState.FAILED assert generation.error_code == GenerationError.UNEXPECTED DRIVEWAY_POUR ={'area': 'driveway', 'service': 'concrete_pour', 'answers': {'turf_strips': False}} POOL_RENOVATION = { 'area': 'pool', 'service': 'pool_renovation', 'answers': { 'plaster_color': 'sky_blue', 'waterline_color': 'blues', 'waterline_size': '6x6', 'add_sun_shelf': 'with_bubblers', }, } @pytest.fixture def enqueued(monkeypatch): calls = [] monkeypatch.setattr('apps.ai.tasks.generate_image.delay', calls.append) return calls class TestRequest: def test_it_freezes_the_prompt_and_enqueues_on_commit( self, media_root, enqueued, django_capture_on_commit_callbacks ): account = create_account() analysis = create_photo_analysis(account, pool_present=True) with django_capture_on_commit_callbacks(execute=True): generation = services.request_generation( account, analysis, [POOL_RENOVATION, DRIVEWAY_POUR] ) assert generation.state == GenerationState.QUEUED assert generation.analysis == analysis assert generation.requested_by == account assert generation.origin == GenerationOrigin.TOOLS_APP cleaned = prompt.clean_request([POOL_RENOVATION, DRIVEWAY_POUR], True) # El acabado de la pileta manda su muestra de agua: el prompt congelado # es el que la nombra. references = prompt.references_of(cleaned) assert generation.prompt == prompt.build_prompt(cleaned, True, references=references) assert generation.references == ['ai/pool_renovation/plaster_color/reference/sky_blue.jpg'] assert enqueued == [str(generation.id)] def test_an_area_left_to_the_notes_is_stored_that_way(self, media_root, enqueued): analysis = create_photo_analysis() generation = services.request_generation( None, analysis, [{**DRIVEWAY_POUR, 'whole_area': False}] ) assert generation.areas == [{**DRIVEWAY_POUR, 'whole_area': False}] assert prompt.recipe()['area_instruction_by_notes'] in generation.prompt def test_the_areas_are_stored_clean_and_in_the_order_of_the_taxonomy( self, media_root, enqueued ): analysis = create_photo_analysis(pool_present=True) generation = services.request_generation(None, analysis, [POOL_RENOVATION, DRIVEWAY_POUR]) # El interruptor del área se guarda siempre, aunque no haya llegado: lo # que se lee del pedido tiene que decir cómo se armó su prompt. assert generation.areas == [ {**DRIVEWAY_POUR, 'whole_area': True}, {**POOL_RENOVATION, 'whole_area': True}, ] assert generation.kind == '' assert generation.answers == {} assert generation.is_v2 is True assert generation.requested_by is None def test_the_notes_are_stored_apart_and_also_go_into_the_prompt(self, media_root, enqueued): """ Se guardan las dos veces a propósito: en el prompt porque es lo que se mandó, y aparte porque es lo único que la persona redactó y hay que poder devolvérselo al formulario para ajustar y volver a pedir. """ analysis = create_photo_analysis() generation = services.request_generation( None, analysis, [DRIVEWAY_POUR], ' Leave the mailbox alone. ' ) assert generation.notes == 'Leave the mailbox alone.' assert 'Leave the mailbox alone.' in generation.prompt def test_without_notes_nothing_extra_reaches_the_prompt(self, media_root, enqueued): analysis = create_photo_analysis() generation = services.request_generation(None, analysis, [DRIVEWAY_POUR]) assert generation.notes == '' assert generation.prompt == prompt.build_prompt( prompt.clean_request([DRIVEWAY_POUR], False), False ) def test_a_lowered_guard_is_stored_with_the_request_and_changes_the_prompt( self, media_root, enqueued ): """ El guard se guarda entero, no sólo lo que se bajó: es lo que explica un resultado que ninguna respuesta pide, y lo que devuelve el formulario como estaba al ajustar. """ analysis = create_photo_analysis() generation = services.request_generation( None, analysis, [DRIVEWAY_POUR], '', {'coverage': False} ) assert generation.guards == {'coverage': False, 'objects': True} assert 'or in the Special Notes' in generation.prompt def test_without_guards_the_prompt_is_the_one_with_every_rule_in_place( self, media_root, enqueued ): analysis = create_photo_analysis() generation = services.request_generation(None, analysis, [DRIVEWAY_POUR]) assert generation.guards == {'coverage': True, 'objects': True} assert generation.prompt == prompt.build_prompt( prompt.clean_request([DRIVEWAY_POUR], False), False ) def test_the_sample_of_the_chosen_color_is_stored_with_the_request(self, media_root, enqueued): """ Se guarda y no se recalcula al ejecutar: el prompt la nombra por número, y afinar el catálogo en el medio dejaría el texto apuntando a otra. """ analysis = create_photo_analysis() overlay = { 'area': 'entryway', 'service': 'concrete_overlay', 'answers': {'grout_color': 'sand', 'stone_color': 'shadow_stone'}, } generation = services.request_generation(None, analysis, [overlay]) assert generation.references == [ 'ai/concrete_overlay/stone_color/irregular/shadow_stone.jpg' ] assert '- Image 2: the stone color and texture for Entryway.' in generation.prompt def test_a_request_without_samples_stores_none_and_says_nothing(self, media_root, enqueued): analysis = create_photo_analysis() generation = services.request_generation(None, analysis, [DRIVEWAY_POUR]) assert generation.references == [] assert '## Design References' not in generation.prompt def test_the_photo_is_the_one_of_the_analysis(self, media_root, enqueued): """No se vuelve a subir ni a comprimir: se copia la ruta y siempre se edita sobre ella.""" analysis = create_photo_analysis() generation = services.request_generation(None, analysis, [DRIVEWAY_POUR]) assert generation.source_path == analysis.source_path == f'sources/{analysis.id}.jpg' assert generation.has_source_image is True def test_without_a_key_it_fails_without_enqueueing(self, media_root, settings, enqueued): settings.OPENAI_API_KEY = '' analysis = create_photo_analysis() generation = services.request_generation(None, analysis, [DRIVEWAY_POUR]) assert generation.state == GenerationState.FAILED assert generation.error_code == GenerationError.PROVIDER_NOT_CONFIGURED assert enqueued == [] @pytest.mark.parametrize( ('state', 'error_code'), [ (AnalysisState.QUEUED, ''), (AnalysisState.RUNNING, ''), (AnalysisState.RETRYING, ''), (AnalysisState.FAILED, AnalysisError.PROVIDER_UNAVAILABLE), (AnalysisState.FAILED, AnalysisError.UNEXPECTED), ], ) def test_an_analysis_that_did_not_finish_cannot_be_generated_on( self, media_root, enqueued, state, error_code ): analysis = create_photo_analysis(state=state, error_code=error_code) with pytest.raises(services.AnalysisNotReady) as raised: services.request_generation(None, analysis, [DRIVEWAY_POUR]) assert raised.value.code == 'ANALYSIS_NOT_READY' assert not ImageGeneration.objects.exists() assert enqueued == [] def test_an_unreadable_answer_still_lets_the_person_choose_the_areas( self, media_root, enqueued ): """ El modelo contestó algo ilegible: la foto sirve, las áreas las elige la persona. Sin detección no hay pileta, así que pedirla se rechaza. """ analysis = create_photo_analysis( state=AnalysisState.FAILED, error_code=AnalysisError.MALFORMED_RESPONSE, areas_detected=(), ) generation = services.request_generation(None, analysis, [DRIVEWAY_POUR]) assert generation.analysis == analysis assert analysis.can_generate assert not analysis.has_pool def test_the_pool_is_allowed_only_when_the_analysis_saw_one(self, media_root, enqueued): """R41 pasa por `has_pool`, que fusiona `pool_present` y `pool` en la lista.""" without_pool = create_photo_analysis(pool_present=False) with pytest.raises(AnswerInvalid) as raised: services.request_generation(None, without_pool, [POOL_RENOVATION]) assert (raised.value.code, raised.value.key) == ('AREA_NOT_ALLOWED', 'areas[0].area') assert not ImageGeneration.objects.exists() with_pool_in_list = create_photo_analysis(areas_detected=('pool',), pool_present=False) generation = services.request_generation(None, with_pool_in_list, [POOL_RENOVATION]) assert generation.areas == [{**POOL_RENOVATION, 'whole_area': True}] def test_what_the_catalog_rejects_leaves_no_trace(self, media_root, enqueued): analysis = create_photo_analysis() with pytest.raises(AnswerInvalid) as raised: services.request_generation(None, analysis, []) assert raised.value.code == 'AREAS_EMPTY' assert not ImageGeneration.objects.exists() assert enqueued == [] class TestResultFile: def test_only_our_own_names_resolve(self, media_root): results = media_root / 'results' results.mkdir() (results / '0195c0de-0000-7000-8000-000000000001.jpg').write_bytes(PNG) assert services.result_file('0195c0de-0000-7000-8000-000000000001.jpg') is not None assert services.result_file('0195c0de-0000-7000-8000-000000000002.jpg') is None assert services.result_file('../sources/patio.jpg') is None assert services.result_file('0195c0de-0000-7000-8000-000000000001.png') is None def test_the_marks_are_png_and_only_our_own_names_resolve(self, media_root): markups = media_root / 'markups' markups.mkdir() (markups / '0195c0de-0000-7000-8000-000000000001.png').write_bytes(PNG) assert services.markup_file('0195c0de-0000-7000-8000-000000000001.png') is not None assert services.markup_file('0195c0de-0000-7000-8000-000000000001.jpg') is None assert services.markup_file('../sources/patio.png') is None #: La foto de las pruebas de marcas, del tamaño que tiene que tener la capa. PHOTO_SIZE = (64, 48) def marks(size=PHOTO_SIZE, painted=True) -> SimpleUploadedFile: """La capa que sube el editor: transparente, con un rectángulo pintado.""" layer = Image.new('RGBA', size, (0, 0, 0, 0)) if painted: layer.paste((239, 68, 68, 255), (10, 10, 30, 20)) buffer = BytesIO() layer.save(buffer, format='PNG') return SimpleUploadedFile('markup.png', buffer.getvalue(), content_type='image/png') def analysis_with_photo(media_root): analysis = create_photo_analysis() photo = media_root / analysis.source_path photo.parent.mkdir(parents=True, exist_ok=True) photo.write_bytes(jpeg_bytes(*PHOTO_SIZE)) return analysis class TestMarkup: NOTE = 'Red: a paver walkway.' def test_the_layer_is_stored_with_its_note_and_the_configured_mode(self, media_root, enqueued): """El modo se lee al pedir y se congela: el prompt que se arma acá depende de él.""" settings_row = AiSettings.current() settings_row.markup_mode = MarkupMode.LAYER settings_row.save() analysis = analysis_with_photo(media_root) generation = services.request_generation( None, analysis, [DRIVEWAY_POUR], markup_layer=marks(), markup_note=self.NOTE, ) assert generation.markup_path == f'markups/{generation.id}.png' assert (media_root / generation.markup_path).is_file() assert generation.markup_note == self.NOTE assert generation.markup_mode == 'layer' assert '## Markup' in generation.prompt assert 'is a transparent layer' in generation.prompt def test_without_marks_nothing_changes(self, media_root, enqueued): generation = services.request_generation( None, analysis_with_photo(media_root), [DRIVEWAY_POUR] ) assert generation.markup_path == '' assert generation.markup_mode == '' assert '## Markup' not in generation.prompt @pytest.mark.parametrize( ('overrides', 'code'), [ ({'markup_note': ''}, 'MARKUP_NOTE_MISSING'), ({'markup_layer': marks(size=(32, 24))}, 'MARKUP_INVALID'), ({'markup_layer': marks(painted=False)}, 'MARKUP_INVALID'), ({'markup_layer': None}, 'MARKUP_INVALID'), ], ) def test_what_does_not_serve_is_refused_and_leaves_no_trace( self, media_root, enqueued, overrides, code ): """Una nota sin capa también se rechaza: ignorarla haría creer que llegó al modelo.""" arguments = {'markup_layer': marks(), 'markup_note': self.NOTE} with pytest.raises(AnswerInvalid) as raised: services.request_generation( None, analysis_with_photo(media_root), [DRIVEWAY_POUR], **{**arguments, **overrides} ) assert (raised.value.code, raised.value.key) == (code, 'markup') assert not ImageGeneration.objects.exists() assert not (media_root / 'markups').exists() assert enqueued == [] def requested(self, media_root, mode): settings_row = AiSettings.current() settings_row.markup_mode = mode settings_row.save() return services.request_generation( None, analysis_with_photo(media_root), [DRIVEWAY_POUR], markup_layer=marks(), markup_note=self.NOTE, ) def test_with_the_copy_it_travels_last_and_there_is_no_mask( self, media_root, enqueued, fake_images ): images_resource = fake_images(completed_response()) generation = self.requested(media_root, MarkupMode.COPY) services.execute(generation) _method, kwargs = images_resource.calls[0] photo, copy = kwargs['image'] assert Path(photo.name).name == Path(generation.source_path).name assert copy[0] == 'markup.png' assert 'mask' not in kwargs def test_with_the_layer_the_stored_drawing_travels_last_and_there_is_no_mask( self, media_root, enqueued, fake_images ): images_resource = fake_images(completed_response()) generation = self.requested(media_root, MarkupMode.LAYER) services.execute(generation) _method, kwargs = images_resource.calls[0] _photo, drawing = kwargs['image'] assert drawing[1] == (media_root / generation.markup_path).read_bytes() assert 'mask' not in kwargs def test_the_mode_frozen_in_the_request_wins_over_the_current_one( self, media_root, enqueued, fake_images ): """Cambiar «Config» con un pedido en cola no puede dejar su prompt apuntando a otra cosa.""" images_resource = fake_images(completed_response()) generation = self.requested(media_root, MarkupMode.COPY) settings_row = AiSettings.current() settings_row.markup_mode = MarkupMode.LAYER settings_row.save() services.execute(generation) _method, kwargs = images_resource.calls[0] _photo, drawing = kwargs['image'] assert drawing[1] != (media_root / generation.markup_path).read_bytes() def test_the_result_is_used_as_it_comes(self, media_root, enqueued, fake_images): """Las marcas son un extra: nada recorta el resultado a la zona marcada (D4).""" fake_images(completed_response()) generation = self.requested(media_root, MarkupMode.LAYER) services.execute(generation) generation.refresh_from_db() stored = Image.open(BytesIO((media_root / generation.result_path).read_bytes())) assert generation.state == GenerationState.COMPLETED assert stored.size == Image.open(BytesIO(PNG)).size