""" La herramienta de IA: sus direcciones existen y llevan adonde dicen. El generador se prueba por su contrato con la pantalla: qué props viajan, adónde vuelve el envío y qué pasa con lo que no corresponde. Lo que hace el proveedor vive en `tests/unit/test_ai_services.py`. """ import json import pytest from django.core.files.uploadedfile import SimpleUploadedFile from apps.ai.models import ( GenerationKind, GenerationOrigin, GenerationState, ImageGeneration, ) from tests.factories import create_account from tests.images import png_bytes pytestmark = pytest.mark.django_db POOL_ANSWERS = { 'pool_shape': 'kidney', 'want_spa': 'no', 'pool_finish': 'tile', 'pool_water_color': 'dark_blue', 'pool_deck_material': 'pavers', 'pool_deck_color_family': 'grey_charcoal', } PNG = png_bytes(64, 48) @pytest.fixture def account(client): account = create_account() client.force_login(account) return account @pytest.fixture def ready(settings, tmp_path, monkeypatch): """Clave presente, disco temporal y cola sustituida.""" settings.OPENAI_API_KEY = 'sk-prueba' settings.AI_MEDIA_ROOT = tmp_path enqueued = [] monkeypatch.setattr('apps.ai.tasks.generate_image.delay', enqueued.append) return enqueued def generation(**extra) -> ImageGeneration: defaults = { 'kind': GenerationKind.POOL, 'origin': GenerationOrigin.TOOLS_APP, 'answers': POOL_ANSWERS, 'prompt': 'Generate a kidney-shaped pool', } return ImageGeneration.objects.create(**{**defaults, **extra}) def props(client, url): return client.get(url, HTTP_X_INERTIA='true').json()['props'] class TestTheToolHome: def test_it_sends_to_the_pool_generator(self, client, account): response = client.get('/ai') assert response.status_code == 302 assert response.headers['Location'] == '/ai/image-generator/pool' class TestTheGeneratorScreen: @pytest.mark.parametrize( ('url', 'component', 'kind'), [ ('/ai/image-generator/pool', 'Ai/ImageGenerator/Pool', 'POOL'), ('/ai/image-generator/paver', 'Ai/ImageGenerator/Paver', 'PAVER'), ], ) def test_each_one_draws_its_own_screen_with_its_questions( self, client, account, url, component, kind ): response = client.get(url, HTTP_X_INERTIA='true') assert response.status_code == 200 page = response.json() assert page['component'] == component assert page['props']['kind'] == kind assert page['props']['questions'][0]['key'] in {'pool_shape', 'paver_material'} assert page['props']['prompt']['opening'] == 'Generate ' assert page['props']['generation'] is None def test_the_named_generation_travels_with_its_state(self, client, account): pending = generation() page = props(client, f'/ai/image-generator/pool?generation={pending.id}') assert page['generation']['id'] == str(pending.id) assert page['generation']['state'] == 'QUEUED' assert page['generation']['result_url'] is None def test_a_generation_of_another_kind_is_ignored(self, client, account): """La pantalla de piletas no puede explicar un paver.""" paver = generation(kind=GenerationKind.PAVER) page = props(client, f'/ai/image-generator/pool?generation={paver.id}') assert page['generation'] is None def test_a_malformed_identifier_is_ignored(self, client, account): page = props(client, '/ai/image-generator/pool?generation=no-es-un-uuid') assert page['generation'] is None def test_the_named_generation_carries_its_answers(self, client, account): """Son lo que la pantalla muestra mientras corre y lo que precarga al volver a pedir.""" pending = generation() page = props(client, f'/ai/image-generator/pool?generation={pending.id}') assert page['generation']['answers'] == POOL_ANSWERS assert page['generation']['prompt'] == 'Generate a kidney-shaped pool' def test_recent_ones_include_every_state_and_only_this_kind(self, client, account): """Un registro que esconde lo que salió mal hace creer que todo salió bien.""" generation(state=GenerationState.COMPLETED, result_path='results/x.jpg') generation(state=GenerationState.FAILED) generation(state=GenerationState.RUNNING) generation(kind=GenerationKind.PAVER, state=GenerationState.RUNNING) page = props(client, '/ai/image-generator/pool') assert {row['state'] for row in page['recent']} == {'COMPLETED', 'FAILED', 'RUNNING'} assert page['recent_total'] == 3 assert page['active_states'] == ['RUNNING'] def test_recent_ones_come_trimmed_with_their_total(self, client, account, monkeypatch): monkeypatch.setattr('apps.ai.services.RECENT_LIMIT', 2) for _ in range(3): generation(state=GenerationState.COMPLETED, result_path='results/x.jpg') page = props(client, '/ai/image-generator/pool') assert len(page['recent']) == 2 assert page['recent_total'] == 3 assert page['recent_limit'] == 2 class TestSubmitting: def test_it_enqueues_and_comes_back_to_watch_it( self, client, account, ready, django_capture_on_commit_callbacks ): with django_capture_on_commit_callbacks(execute=True): response = client.post( '/ai/image-generator/pool', {'answers': json.dumps(POOL_ANSWERS)} ) created = ImageGeneration.objects.get() assert response.status_code == 302 assert response.headers['Location'] == f'/ai/image-generator/pool?generation={created.id}' assert created.origin == GenerationOrigin.TOOLS_APP assert created.requested_by == account assert created.state == GenerationState.QUEUED assert ready == [str(created.id)] def test_the_photo_is_kept_for_the_task_and_never_served( self, client, account, ready, settings ): photo = SimpleUploadedFile('patio.jpg', PNG, content_type='image/jpeg') client.post( '/ai/image-generator/paver', { 'answers': json.dumps( { 'paver_material': 'clay_brick', 'border_accent': 'no', 'grass_joint': 'no', 'driveway_style': 'classic', } ), 'image': photo, }, ) created = ImageGeneration.objects.get() assert created.source_path == f'sources/{created.id}.jpg' assert (settings.AI_MEDIA_ROOT / created.source_path).is_file() assert 'Do not modify the base image' in created.prompt # Está en `sources/`, no en `results/`: la dirección pública no la sirve. assert client.get(f'/media/ai/{created.id}.jpg').status_code == 404 # La sirve la dirección con sesión, y el pedido la nombra. page = props(client, f'/ai/image-generator/paver?generation={created.id}') assert page['generation']['source_url'] == f'/media/ai/sources/{created.id}.jpg' response = client.get(page['generation']['source_url']) assert response.status_code == 200 assert response['Content-Type'] == 'image/jpeg' assert 'private' in response['Cache-Control'] def test_a_missing_answer_comes_back_pinned_to_its_question(self, client, account, ready): response = client.post( '/ai/image-generator/pool', {'answers': json.dumps({**POOL_ANSWERS, 'pool_finish': ''})}, ) assert response.status_code == 302 assert response.headers['Location'] == '/ai/image-generator/pool' assert client.session['_validation_errors'] == {'pool_finish': {'code': 'ANSWER_MISSING'}} assert not ImageGeneration.objects.exists() def test_a_photo_that_is_not_an_image_is_refused(self, client, account, ready): response = client.post( '/ai/image-generator/pool', { 'answers': json.dumps(POOL_ANSWERS), 'image': SimpleUploadedFile('clave.json', b'{}', content_type='application/json'), }, ) assert response.status_code == 302 assert client.session['_validation_errors'] == {'image': {'code': 'IMAGE_INVALID'}} assert not ImageGeneration.objects.exists() def test_without_a_key_the_request_fails_and_the_screen_says_so( self, client, account, ready, settings ): settings.OPENAI_API_KEY = '' response = client.post('/ai/image-generator/pool', {'answers': json.dumps(POOL_ANSWERS)}) created = ImageGeneration.objects.get() assert response.headers['Location'].endswith(f'?generation={created.id}') page = props(client, response.headers['Location']) assert page['generation']['state'] == 'FAILED' assert page['generation']['error_code'] == 'PROVIDER_NOT_CONFIGURED' assert ready == [] class TestTheQueue: def test_it_lists_every_request_with_its_origin_and_state(self, client, account): generation(origin=GenerationOrigin.WEBSITE, state=GenerationState.COMPLETED) generation(requested_by=account, state=GenerationState.RUNNING) page = props(client, '/ai/image-generator/queue') assert page['total'] == 2 assert page['limit'] == 200 origins = {row['origin'] for row in page['generations']} assert origins == {'WEBSITE', 'TOOLS_APP'} who = {row['requested_by'] for row in page['generations']} assert who == {None, account.email} assert page['active_states'] == ['RUNNING'] assert page['options']['origin'] == ['TOOLS_APP', 'WEBSITE'] class TestPublicMedia: def test_a_result_opens_without_a_session(self, client, settings, tmp_path): settings.AI_MEDIA_ROOT = tmp_path results = tmp_path / 'results' results.mkdir() (results / '0195c0de-0000-7000-8000-000000000001.jpg').write_bytes(PNG) response = client.get('/media/ai/0195c0de-0000-7000-8000-000000000001.jpg') assert response.status_code == 200 assert response['Content-Type'] == 'image/jpeg' assert 'immutable' in response['Cache-Control'] assert b''.join(response.streaming_content) == PNG def test_anything_that_is_not_ours_is_not_found(self, client, settings, tmp_path): settings.AI_MEDIA_ROOT = tmp_path assert client.get('/media/ai/0195c0de-0000-7000-8000-000000000009.jpg').status_code == 404 assert client.get('/media/ai/..%2Fsources%2Fpatio.jpg').status_code == 404 def test_a_source_photo_needs_a_session(self, client, account, settings, tmp_path): """Es el patio real de un cliente: sin sesión se va al ingreso, no a la foto.""" settings.AI_MEDIA_ROOT = tmp_path sources = tmp_path / 'sources' sources.mkdir() (sources / '0195c0de-0000-7000-8000-000000000002.jpg').write_bytes(PNG) url = '/media/ai/sources/0195c0de-0000-7000-8000-000000000002.jpg' assert client.get(url).status_code == 200 assert client.get('/media/ai/sources/patio.jpg').status_code == 404 client.logout() response = client.get(url) assert response.status_code == 302 assert response.headers['Location'].startswith('/login')