""" El generador de imágenes por la API. Dos generaciones de contrato conviven: - **El de la landing** (`catalog`, `analyses`, `generations`): lo que se defiende es la forma de cada respuesta, que ninguna deja ver un prompt ni la foto del cliente, que sólo se ve lo que entró por la landing, los códigos de cada rechazo y el tope por visitante con su `Retry-After`. - **La v1, deprecada** (`image-generations`, decisión 2 de `ai-image-generator-v2`): las dos direcciones siguen existiendo, la clave sigue siendo obligatoria, y con clave contestan `410 Gone` con un mensaje que apunta a las nuevas. """ import json from io import BytesIO import pytest from django.core.files.uploadedfile import SimpleUploadedFile from PIL import Image from rest_framework.test import APIClient from apps.accounts.models import ApiKey from apps.ai import website from apps.ai.models import ( AiSettings, AnalysisError, AnalysisState, GenerationOrigin, ImageGeneration, PhotoAnalysis, ) from tests.factories import create_account, create_photo_analysis from tests.images import jpeg_bytes pytestmark = pytest.mark.django_db ENDPOINT = '/api/v1/ai/image-generations' SOME_ID = '0195c0de-0000-7000-8000-000000000001' V1_FORM = {'kind': 'POOL', 'answers': json.dumps({'pool_shape': 'kidney'})} CATALOG = '/api/v1/ai/catalog' ANALYSES = '/api/v1/ai/analyses' GENERATIONS = '/api/v1/ai/generations' DRIVEWAY_POUR = {'area': 'driveway', 'service': 'concrete_pour', 'answers': {'turf_strips': False}} VISITOR = {'HTTP_X_VISITOR_IP': '203.0.113.7', 'HTTP_X_VISITOR_USER_AGENT': 'Safari'} @pytest.fixture def api(): account = create_account() _, plain_text = ApiKey.issue(account=account, name='landing') client = APIClient() client.credentials(HTTP_AUTHORIZATION=f'Api-Key {plain_text}') return client @pytest.fixture def media_root(settings, tmp_path, monkeypatch): """Disco aparte, clave de mentira y una cola que no sale a ningún lado.""" settings.AI_MEDIA_ROOT = tmp_path settings.OPENAI_API_KEY = 'sk-prueba' monkeypatch.setattr('apps.ai.tasks.analyze_photo.delay', lambda *args: None) monkeypatch.setattr('apps.ai.tasks.generate_image.delay', lambda *args: None) return tmp_path def website_analysis(media_root, **extra) -> PhotoAnalysis: analysis = create_photo_analysis(origin=GenerationOrigin.WEBSITE, **extra) photo = media_root / analysis.source_path photo.parent.mkdir(parents=True, exist_ok=True) photo.write_bytes(jpeg_bytes(64, 48)) return analysis def marks(size=(64, 48)) -> SimpleUploadedFile: layer = Image.new('RGBA', size, (0, 0, 0, 0)) 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 generate(api, analysis, areas=(DRIVEWAY_POUR,), **fields): data = {'analysis_id': str(analysis.id), 'areas': json.dumps(list(areas)), **fields} return api.post(GENERATIONS, data, format='multipart', **VISITOR) @pytest.mark.parametrize( ('method', 'path'), [ ('get', CATALOG), ('post', ANALYSES), ('get', f'{ANALYSES}/{SOME_ID}'), ('post', GENERATIONS), ('get', f'{GENERATIONS}/{SOME_ID}'), ], ) def test_every_landing_endpoint_needs_the_key(method, path): response = getattr(APIClient(), method)(path) assert response.status_code == 401 assert response.json()['error']['code'] == 'UNAUTHENTICATED' class TestCatalog: def test_it_comes_with_its_version_texts_and_absolute_photos(self, api): response = api.get(CATALOG) assert response.status_code == 200 body = response.json() assert body['version'] == website.catalog_version() assert response['ETag'] == f'"{body["version"]}"' assert body['texts']['ai.services.paver'] == 'Paver' paver = body['catalog']['services'][0] assert paver['image'].startswith('http://testserver/static/ai/paver/service.jpg?v=') def test_not_a_single_prompt_travels(self, api): text = api.get(CATALOG).content.decode() assert '"prompt' not in text assert 'guards' not in text @pytest.mark.parametrize('header', ['"{v}"', 'W/"{v}"', '"otra", "{v}"', '*']) def test_the_stored_version_gets_a_304_without_body(self, api, header): version = website.catalog_version() response = api.get(CATALOG, HTTP_IF_NONE_MATCH=header.format(v=version)) assert response.status_code == 304 assert response.content == b'' assert response['ETag'] == f'"{version}"' def test_an_old_version_gets_the_new_catalog(self, api): response = api.get(CATALOG, HTTP_IF_NONE_MATCH='"vieja"') assert response.status_code == 200 assert 'catalog' in response.json() class TestAnalyses: def test_it_queues_the_analysis_of_a_visitor(self, api, media_root): photo = SimpleUploadedFile('patio.jpg', jpeg_bytes(800, 600), content_type='image/jpeg') response = api.post(ANALYSES, {'image': photo}, format='multipart') assert response.status_code == 202 body = response.json() analysis = PhotoAnalysis.objects.get(id=body['id']) assert analysis.origin == GenerationOrigin.WEBSITE assert analysis.requested_by is None assert body['state'] == 'QUEUED' assert set(body) == { 'id', 'state', 'error_code', 'can_generate', 'areas', 'pool_present', 'confidence', 'points', 'photo_size', 'created_at', 'finished_at', } def test_the_photo_size_is_the_one_after_normalizing(self, api, media_root): """Se reescala a 1536 de lado mayor y no se agranda: es la medida de la capa de marcas.""" large = SimpleUploadedFile('casa.jpg', jpeg_bytes(3000, 2000), content_type='image/jpeg') small = SimpleUploadedFile('patio.jpg', jpeg_bytes(800, 600), content_type='image/jpeg') assert api.post(ANALYSES, {'image': large}, format='multipart').json()['photo_size'] == [ 1536, 1024, ] assert api.post(ANALYSES, {'image': small}, format='multipart').json()['photo_size'] == [ 800, 600, ] def test_a_photo_that_already_fits_keeps_its_own_size(self, api, media_root): """Es lo que sube la landing, ya comprimida a `catalog.photo.max_side`.""" photo = SimpleUploadedFile('casa.jpg', jpeg_bytes(1536, 864), content_type='image/jpeg') response = api.post(ANALYSES, {'image': photo}, format='multipart') assert response.json()['photo_size'] == [1536, 864] def test_without_its_photo_the_size_is_null(self, api, media_root): analysis = create_photo_analysis(origin=GenerationOrigin.WEBSITE) assert api.get(f'{ANALYSES}/{analysis.id}').json()['photo_size'] is None def test_without_a_photo_it_is_a_422_on_the_image(self, api, media_root): response = api.post(ANALYSES, {}, format='multipart') assert response.status_code == 422 error = response.json()['error'] assert error['code'] == 'VALIDATION_FAILED' assert error['details'] == {'field': 'image', 'reason': 'IMAGE_MISSING'} def test_a_file_that_is_not_an_image_is_a_422_with_its_reason(self, api, media_root): text = SimpleUploadedFile('patio.jpg', b'no soy una foto', content_type='image/jpeg') response = api.post(ANALYSES, {'image': text}, format='multipart') assert response.status_code == 422 assert response.json()['error']['details']['reason'] == 'IMAGE_INVALID' assert not PhotoAnalysis.objects.exists() def test_the_detail_answers_with_the_detected_areas(self, api, media_root): analysis = website_analysis(media_root, areas_detected=('driveway', 'pool')) response = api.get(f'{ANALYSES}/{analysis.id}') assert response.status_code == 200 body = response.json() assert body['state'] == 'COMPLETED' assert body['areas'] == ['driveway', 'pool'] assert body['pool_present'] is True assert 'source_url' not in body def test_an_analysis_of_the_tools_screen_is_not_found(self, api, media_root): analysis = create_photo_analysis() assert api.get(f'{ANALYSES}/{analysis.id}').status_code == 404 class TestGenerations: def test_it_queues_a_visitor_request_with_the_landing_prompt(self, api, media_root): analysis = website_analysis(media_root) response = generate(api, analysis, note='Add a fire pit.') assert response.status_code == 202 body = response.json() generation = ImageGeneration.objects.get(id=body['id']) assert generation.origin == GenerationOrigin.WEBSITE assert generation.visitor_fingerprint assert '203.0.113.7' not in generation.visitor_fingerprint assert body['note'] == 'Add a fire pit.' assert body['has_markup'] is False assert body['result_url'] is None assert 'prompt' not in body def test_the_marks_travel_with_their_note(self, api, media_root): analysis = website_analysis(media_root) response = generate(api, analysis, note='Red: a fire pit.', markup=marks()) assert response.status_code == 202 assert response.json()['has_markup'] is True assert ImageGeneration.objects.get().markup_note == 'Red: a fire pit.' def test_marks_drawn_at_the_photo_size_of_the_analysis_are_the_ones_accepted( self, api, media_root ): """La medida que da el análisis es la que exige la capa: no hay otra que calcular.""" analysis = website_analysis(media_root) width, height = api.get(f'{ANALYSES}/{analysis.id}').json()['photo_size'] accepted = generate(api, analysis, note='Red: a fire pit.', markup=marks((width, height))) refused = generate( api, analysis, note='Red: a fire pit.', markup=marks((width + 1, height)) ) assert accepted.status_code == 202 assert refused.status_code == 422 assert refused.json()['error']['details'] == {'field': 'markup', 'reason': 'MARKUP_INVALID'} def test_a_json_body_works_too(self, api, media_root): """Sin marcas no hace falta multipart: las áreas pueden llegar como lista.""" analysis = website_analysis(media_root) data = {'analysis_id': str(analysis.id), 'areas': [DRIVEWAY_POUR]} response = api.post(GENERATIONS, data, format='json', **VISITOR) assert response.status_code == 202 def test_without_the_visitor_it_is_a_422(self, api, media_root): analysis = website_analysis(media_root) data = {'analysis_id': str(analysis.id), 'areas': json.dumps([DRIVEWAY_POUR])} response = api.post(GENERATIONS, data, format='multipart') assert response.status_code == 422 assert response.json()['error']['details']['reason'] == 'VISITOR_MISSING' assert not ImageGeneration.objects.exists() @pytest.mark.parametrize( ('areas', 'field', 'reason'), [ ([], 'areas', 'AREAS_EMPTY'), ([{'area': 'roof', 'service': 'paver'}], 'areas[0].area', 'AREA_INVALID'), ( [{'area': 'driveway', 'service': 'pool_renovation'}], 'areas[0].service', 'SERVICE_NOT_ALLOWED', ), ], ) def test_what_the_catalog_rejects_is_a_422_on_its_field( self, api, media_root, areas, field, reason ): response = generate(api, website_analysis(media_root), areas) assert response.status_code == 422 assert response.json()['error']['details'] == {'field': field, 'reason': reason} def test_areas_as_broken_text_are_a_422(self, api, media_root): analysis = website_analysis(media_root) data = {'analysis_id': str(analysis.id), 'areas': '[{'} response = api.post(GENERATIONS, data, format='multipart', **VISITOR) assert response.status_code == 422 assert response.json()['error']['details'] == { 'field': 'areas', 'reason': 'ANSWER_INVALID', } def test_an_analysis_that_did_not_finish_is_a_409(self, api, media_root): analysis = website_analysis(media_root, state=AnalysisState.RUNNING) response = generate(api, analysis) assert response.status_code == 409 assert response.json()['error']['details']['reason'] == 'ANALYSIS_NOT_READY' def test_when_the_areas_were_not_recognized_the_visitor_chooses_them(self, api, media_root): """`MALFORMED_RESPONSE`: la API lo dice con `can_generate` y deja generar.""" analysis = website_analysis( media_root, state=AnalysisState.FAILED, error_code=AnalysisError.MALFORMED_RESPONSE, areas_detected=(), ) assert api.get(f'{ANALYSES}/{analysis.id}').json()['can_generate'] is True assert generate(api, analysis).status_code == 202 def test_when_the_analysis_request_failed_it_cannot_go_on(self, api, media_root): analysis = website_analysis( media_root, state=AnalysisState.FAILED, error_code=AnalysisError.PROVIDER_UNAVAILABLE ) assert api.get(f'{ANALYSES}/{analysis.id}').json()['can_generate'] is False assert generate(api, analysis).status_code == 409 @pytest.mark.parametrize('identifier', ['', 'no-es-un-uuid', SOME_ID]) def test_an_unknown_analysis_is_a_404(self, api, media_root, identifier): data = {'analysis_id': identifier, 'areas': json.dumps([DRIVEWAY_POUR])} response = api.post(GENERATIONS, data, format='multipart', **VISITOR) assert response.status_code == 404 def test_an_analysis_of_the_tools_screen_is_a_404(self, api, media_root): assert generate(api, create_photo_analysis()).status_code == 404 def test_over_the_limit_it_is_a_429_with_when_to_retry(self, api, media_root): current = AiSettings.current() current.website_hourly_limit = 1 current.save() analysis = website_analysis(media_root) assert generate(api, analysis).status_code == 202 response = generate(api, analysis) assert response.status_code == 429 error = response.json()['error'] assert error['code'] == 'RATE_LIMITED' assert error['details']['reason'] == 'VISITOR_LIMIT_REACHED' assert error['details']['window'] == 'hour' assert error['details']['limit'] == 1 assert int(response['Retry-After']) == error['details']['retry_after'] > 0 assert ImageGeneration.objects.count() == 1 def test_another_visitor_is_not_affected(self, api, media_root): current = AiSettings.current() current.website_hourly_limit = 1 current.save() analysis = website_analysis(media_root) generate(api, analysis) data = {'analysis_id': str(analysis.id), 'areas': json.dumps([DRIVEWAY_POUR])} response = api.post( GENERATIONS, data, format='multipart', HTTP_X_VISITOR_IP='198.51.100.4', HTTP_X_VISITOR_USER_AGENT='Safari', ) assert response.status_code == 202 def test_an_idempotency_key_replays_without_spending_the_quota(self, api, media_root): analysis = website_analysis(media_root) headers = {**VISITOR, 'HTTP_IDEMPOTENCY_KEY': 'visitor-42-try-1'} data = {'analysis_id': str(analysis.id), 'areas': json.dumps([DRIVEWAY_POUR])} first = api.post(GENERATIONS, data, format='multipart', **headers) second = api.post(GENERATIONS, data, format='multipart', **headers) assert first.status_code == second.status_code == 202 assert second['Idempotent-Replay'] == 'true' assert second.json()['id'] == first.json()['id'] assert ImageGeneration.objects.count() == 1 def test_the_detail_has_absolute_urls_to_see_and_to_download(self, api, media_root): analysis = website_analysis(media_root) generation_id = generate(api, analysis).json()['id'] ImageGeneration.objects.filter(id=generation_id).update( state='COMPLETED', result_path='results/0195c0de-0000-7000-8000-000000000009.jpg' ) body = api.get(f'{GENERATIONS}/{generation_id}').json() assert body['state'] == 'COMPLETED' assert body['result_url'] == ( 'http://testserver/media/ai/0195c0de-0000-7000-8000-000000000009.jpg' ) assert body['download_url'] == f'{body["result_url"]}?download=1' def test_a_generation_of_the_tools_screen_is_not_found(self, api, media_root): generation = ImageGeneration.objects.create(origin=GenerationOrigin.TOOLS_APP, prompt='p') assert api.get(f'{GENERATIONS}/{generation.id}').status_code == 404 def test_without_a_key_it_still_responds_401(): """La autenticación va antes que la deprecación: una clave inválida no se entera de nada.""" response = APIClient().post(ENDPOINT, V1_FORM, format='multipart') assert response.status_code == 401 assert response.json()['error']['code'] == 'UNAUTHENTICATED' assert APIClient().get(f'{ENDPOINT}/{SOME_ID}').status_code == 401 class TestCreating: def test_it_is_gone_with_a_code_and_a_message(self, api): response = api.post(ENDPOINT, V1_FORM, format='multipart') assert response.status_code == 410 error = response.json()['error'] assert error['code'] == 'DEPRECATED' assert 'v2' in error['message'] assert 'details' not in error assert not ImageGeneration.objects.exists() def test_the_format_of_the_request_no_longer_matters(self, api): """Antes un JSON a secas era 415; ya no hay contrato de envío que defender.""" response = api.post(ENDPOINT, {'kind': 'POOL', 'answers': {}}, format='json') assert response.status_code == 410 assert response.json()['error']['code'] == 'DEPRECATED' def test_an_idempotency_key_does_not_replay_anything(self, api): headers = {'HTTP_IDEMPOTENCY_KEY': 'landing-visitor-42'} first = api.post(ENDPOINT, V1_FORM, format='multipart', **headers) second = api.post(ENDPOINT, V1_FORM, format='multipart', **headers) assert first.status_code == 410 assert second.status_code == 410 assert 'Idempotent-Replay' not in second class TestAsking: def test_any_identifier_is_gone(self, api): response = api.get(f'{ENDPOINT}/{SOME_ID}') assert response.status_code == 410 assert response.json()['error']['code'] == 'DEPRECATED' def test_a_malformed_identifier_is_still_not_found(self, api): """La ruta exige un UUID; lo que no lo es no llega a la vista.""" assert api.get(f'{ENDPOINT}/no-es-un-uuid').status_code == 404