""" El generador para la landing: su prompt, su catálogo, su pedido y su tope. Lo que se defiende son las decisiones de la owner (2026-10-08): - en la landing la nota del visitante **manda** sobre las secciones, salvo contra el cierre fijo y contra el guard de servicios, que no se apaga; - la landing recibe el catálogo sin un solo prompt, con sus textos en inglés y una versión que cambia cuando cambia cualquier cosa que ve, fotos incluidas; - cada visitante tiene un tope por hora y por día, configurable, que cuentan sólo las generaciones; - el aviso del webhook va firmado y sólo lleva la versión. """ import hashlib import hmac import json from datetime import date, timedelta from io import BytesIO import httpx import pytest from django.core.files.uploadedfile import SimpleUploadedFile from django.utils import timezone from PIL import Image from apps.ai import catalog, images, quota, services, webhook, website from apps.ai.markup import MarkupMode from apps.ai.models import AiSettings, GenerationOrigin, ImageGeneration from apps.ai.prompt import ( AnswerInvalid, Markup, build_prompt, build_website_prompt, clean_request, recipe, website_services_only, ) from tests.factories import create_photo_analysis from tests.images import jpeg_bytes DRIVEWAY_POUR = {'area': 'driveway', 'service': 'concrete_pour', 'answers': {'turf_strips': False}} NOTE = 'Add a fire pit in the middle of the patio.' FINGERPRINT = 'a' * 64 def request_of(*areas, pool_present=False): return clean_request(list(areas), pool_present) class TestWebsitePrompt: def test_the_body_is_the_one_of_tools(self): """Preámbulo y secciones son los mismos: la nota cambia lo que puede, no lo que se pide.""" request = request_of(DRIVEWAY_POUR) tools = build_prompt(request, False).split('\n\n') landing = build_website_prompt(request, False).split('\n\n') assert landing[:2] == tools[:2] def test_without_a_note_there_is_neither_notes_nor_markup(self): text = build_website_prompt(request_of(DRIVEWAY_POUR), False) assert '## Special Notes' not in text assert '## Markup' not in text def test_a_note_without_marks_goes_alone_and_leads(self): text = build_website_prompt(request_of(DRIVEWAY_POUR), False, NOTE) notes = text.split('## Special Notes\n', 1)[1].split('\n\n', 1)[0] assert 'follow the note' in notes assert notes.endswith(NOTE) assert '## Markup' not in text def test_with_marks_the_note_goes_with_them_and_only_there(self): """Una marca sin la nota al lado se ignora: la nota tiene que viajar con las marcas.""" markup_request = Markup(mode=MarkupMode.COPY, note=NOTE) text = build_website_prompt(request_of(DRIVEWAY_POUR), False, '', (), markup_request) block = text.split('## Markup\n', 1)[1].split('\n\n', 1)[0] assert 'follow the note' in block assert block.endswith(NOTE) assert '## Special Notes' not in text assert text.count(NOTE) == 1 def test_the_tools_guards_do_not_apply(self): """El visitante no los ve ni puede bajarlos: ni levantados ni bajados aparecen.""" text = build_website_prompt(request_of(DRIVEWAY_POUR), False, NOTE) for guard in catalog.GUARDS: for line in (guard.raised, guard.lowered, guard.notes_raised, guard.notes_lowered): if line: assert line not in text def test_the_services_guard_is_always_there_with_every_service(self): for note in ('', NOTE): assert website_services_only() in build_website_prompt( request_of(DRIVEWAY_POUR), False, note ) for schema in catalog.SCHEMAS: assert schema.label in website_services_only() def test_the_fixed_closing_is_the_same_as_in_tools(self): pieces = recipe() text = build_website_prompt(request_of(DRIVEWAY_POUR), True, NOTE) closing = text.rsplit('\n\n', 1)[1].split('\n') assert closing[0] == pieces['pool_untouched'] assert closing[1] == pieces['no_people'] assert closing[-1] == pieces['realistic'] class TestPublicCatalog: def test_no_prompt_travels(self): payload = json.dumps(website.catalog_payload('/static/')) for key in ('"prompt"', '"prompt_tokens"', '"prompt_true"', '"prompt_false"', 'guards'): assert key not in payload def test_it_asks_the_same_as_the_screen(self): """Sale de `catalog_props`: los mismos servicios, campos y opciones, en el mismo orden.""" screen = catalog.catalog_props('/static/') public = catalog.public_catalog_props('/static/') assert public['areas'] == screen['areas'] assert public['groupable_areas'] == screen['groupable_areas'] for public_schema, schema in zip(public['services'], screen['services'], strict=True): assert public_schema['service'] == schema['service'] assert [field['id'] for field in public_schema['fields']] == [ field['id'] for field in schema['fields'] ] def test_it_says_to_what_size_the_visitor_photo_is_compressed(self): payload = website.catalog_payload('/static/') assert payload['catalog']['photo'] == {'max_side': images.SOURCE_MAX_SIDE} def test_the_texts_are_the_english_ones_of_the_catalog(self): texts = website.catalog_texts() assert texts['ai.services.paver'] == 'Paver' assert texts['ai.areas.driveway'] == 'Driveway' assert all(key.startswith(website.TEXT_PREFIXES) for key in texts) assert 'ai.config.title' not in texts def test_every_photo_carries_the_hash_of_its_content(self): payload = website.catalog_payload('https://tools.test/static/') option = payload['catalog']['services'][0]['fields'][0]['options'][0] path = 'ai/paver/material/travertine.jpg' assert option['image'].startswith(f'https://tools.test/static/{path}?v=') assert len(option['image'].rsplit('?v=', 1)[1]) == website.DIGEST_LENGTH def test_the_version_does_not_depend_on_the_domain(self): assert website.catalog_version() == website.catalog_version() assert len(website.catalog_version()) == 64 def test_the_version_changes_when_a_photo_changes(self, monkeypatch): """Una foto reemplazada con el mismo nombre tiene que llegar a la landing.""" before = website.catalog_version() monkeypatch.setattr(website, '_static_digest', lambda relative: 'otro') assert website.catalog_version() != before def test_the_version_changes_when_a_text_changes(self, monkeypatch): before = website.catalog_version() monkeypatch.setattr(website, 'catalog_texts', lambda: {'ai.services.paver': 'Pavers'}) assert website.catalog_version() != before @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 enqueued(monkeypatch): calls = [] monkeypatch.setattr('apps.ai.tasks.generate_image.delay', calls.append) return calls 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 analysis_with_photo(media_root): analysis = create_photo_analysis(origin=GenerationOrigin.WEBSITE) photo = media_root / analysis.source_path photo.parent.mkdir(parents=True, exist_ok=True) photo.write_bytes(jpeg_bytes(64, 48)) return analysis @pytest.mark.django_db class TestWebsiteRequest: def test_it_comes_from_a_visitor_with_the_landing_prompt(self, media_root, enqueued): analysis = analysis_with_photo(media_root) generation = services.request_website_generation( analysis, [DRIVEWAY_POUR], NOTE, fingerprint=FINGERPRINT ) assert generation.origin == GenerationOrigin.WEBSITE assert generation.requested_by is None assert generation.visitor_fingerprint == FINGERPRINT assert generation.guards == {} assert generation.notes == NOTE assert website_services_only() in generation.prompt assert ImageGeneration.objects.filter(pk=generation.pk).exists() def test_with_marks_the_note_is_theirs(self, media_root, enqueued): analysis = analysis_with_photo(media_root) generation = services.request_website_generation( analysis, [DRIVEWAY_POUR], NOTE, markup_layer=marks(), fingerprint=FINGERPRINT ) assert generation.markup_note == NOTE assert generation.notes == '' assert generation.markup_path == f'markups/{generation.id}.png' assert '## Markup' in generation.prompt def test_marks_without_a_note_are_refused(self, media_root, enqueued): with pytest.raises(AnswerInvalid) as raised: services.request_website_generation( analysis_with_photo(media_root), [DRIVEWAY_POUR], '', markup_layer=marks(), fingerprint=FINGERPRINT, ) assert raised.value.code == 'MARKUP_NOTE_MISSING' assert not ImageGeneration.objects.exists() def test_a_request_the_catalog_rejects_does_not_spend_the_quota(self, media_root, enqueued): """Se valida antes de contar: un pedido mal armado no consume cupo.""" current = AiSettings.current() current.website_daily_limit = 1 current.save() analysis = analysis_with_photo(media_root) with pytest.raises(AnswerInvalid): services.request_website_generation(analysis, [], fingerprint=FINGERPRINT) generation = services.request_website_generation( analysis, [DRIVEWAY_POUR], fingerprint=FINGERPRINT ) assert generation.pk is not None def test_over_the_limit_nothing_is_registered(self, media_root, enqueued): current = AiSettings.current() current.website_hourly_limit = 1 current.save() analysis = analysis_with_photo(media_root) services.request_website_generation(analysis, [DRIVEWAY_POUR], fingerprint=FINGERPRINT) with pytest.raises(quota.VisitorLimitReached): services.request_website_generation(analysis, [DRIVEWAY_POUR], fingerprint=FINGERPRINT) assert ImageGeneration.objects.count() == 1 def website_generation(fingerprint=FINGERPRINT, minutes_ago=0, origin=GenerationOrigin.WEBSITE): generation = ImageGeneration.objects.create( origin=origin, prompt='p', visitor_fingerprint=fingerprint ) if minutes_ago: moment = timezone.now() - timedelta(minutes=minutes_ago) ImageGeneration.objects.filter(pk=generation.pk).update(created_at=moment) return generation @pytest.mark.django_db class TestQuota: def limits(self, hourly, daily): current = AiSettings.current() current.website_hourly_limit = hourly current.website_daily_limit = daily current.save() def test_under_both_limits_it_lets_the_request_through(self): self.limits(hourly=2, daily=5) website_generation() quota.ensure_visitor_quota(FINGERPRINT) def test_the_hourly_window_says_when_it_frees_up(self): self.limits(hourly=2, daily=5) website_generation(minutes_ago=50) website_generation(minutes_ago=10) with pytest.raises(quota.VisitorLimitReached) as raised: quota.ensure_visitor_quota(FINGERPRINT) assert raised.value.window == quota.WINDOW_HOUR assert raised.value.limit == 2 # La más vieja de la hora cumple 60 minutos dentro de 10. assert 9 * 60 <= raised.value.retry_after <= 10 * 60 def test_what_is_older_than_an_hour_counts_only_for_the_day(self): self.limits(hourly=2, daily=3) website_generation(minutes_ago=90) website_generation(minutes_ago=80) quota.ensure_visitor_quota(FINGERPRINT) website_generation(minutes_ago=5) with pytest.raises(quota.VisitorLimitReached) as raised: quota.ensure_visitor_quota(FINGERPRINT) assert raised.value.window == quota.WINDOW_DAY assert raised.value.limit == 3 def test_other_visitors_and_the_tools_screen_do_not_count(self): self.limits(hourly=1, daily=1) website_generation(fingerprint='b' * 64) website_generation(origin=GenerationOrigin.TOOLS_APP) quota.ensure_visitor_quota(FINGERPRINT) def test_a_change_in_config_applies_to_the_next_request(self): self.limits(hourly=1, daily=5) website_generation() with pytest.raises(quota.VisitorLimitReached): quota.ensure_visitor_quota(FINGERPRINT) self.limits(hourly=2, daily=5) quota.ensure_visitor_quota(FINGERPRINT) class TestFingerprint: def test_it_does_not_keep_the_ip(self): fingerprint = quota.visitor_fingerprint('203.0.113.7', 'Safari') assert '203.0.113.7' not in fingerprint assert len(fingerprint) == 64 def test_it_changes_with_the_ip_the_browser_and_the_day(self, monkeypatch): """El día entra en la huella: la ventana diaria se reinicia sola al cambiar la fecha.""" base = quota.visitor_fingerprint('203.0.113.7', 'Safari') assert quota.visitor_fingerprint('203.0.113.8', 'Safari') != base assert quota.visitor_fingerprint('203.0.113.7', 'Firefox') != base monkeypatch.setattr('apps.core.dates.today', lambda: date(2030, 1, 2)) assert quota.visitor_fingerprint('203.0.113.7', 'Safari') != base class TestWebhook: def configured(self, settings): settings.LANDING_CATALOG_WEBHOOK_URL = 'https://landing.test/webhooks/tools/catalog' settings.LANDING_WEBHOOK_SECRET = 'secreto' def test_it_sends_only_the_version_signed_with_the_timestamp(self, settings): self.configured(settings) received = [] def answer(request): received.append(request) return httpx.Response(204) delivery = webhook.notify_catalog( client=httpx.Client(transport=httpx.MockTransport(answer)) ) request = received[0] assert str(request.url) == settings.LANDING_CATALOG_WEBHOOK_URL assert json.loads(request.content) == {'version': website.catalog_version()} timestamp = request.headers[webhook.TIMESTAMP_HEADER] expected = hmac.new( b'secreto', timestamp.encode() + b'.' + request.content, hashlib.sha256 ).hexdigest() assert request.headers[webhook.SIGNATURE_HEADER] == f'sha256={expected}' assert delivery.status_code == 204 def test_without_url_or_secret_it_does_not_send(self, settings): settings.LANDING_CATALOG_WEBHOOK_URL = '' settings.LANDING_WEBHOOK_SECRET = 'secreto' with pytest.raises(webhook.WebhookNotConfigured): webhook.notify_catalog() def test_a_landing_that_answers_with_an_error_is_a_failure(self, settings): self.configured(settings) transport = httpx.MockTransport(lambda request: httpx.Response(401)) with pytest.raises(httpx.HTTPStatusError): webhook.notify_catalog(client=httpx.Client(transport=transport))