""" La pantalla de sitemaps de un dominio (V8). Lo que se prueba acá no es que la tabla dibuje filas, sino que la pantalla no pueda afirmar lo que no sabe. Tres invariantes concentran casi todos los tests: - Un sitemap que no se pudo **leer** no cambia el resultado de su último **envío**. Son dos hechos distintos y confundirlos pone «Falló» al lado de la fecha de un envío que sí salió bien, que es justo la pregunta que trae al usuario a esta pantalla. - `SKIPPED_UNCHANGED` no es un error: no reenviar lo que no cambió ahorra cupo del dominio, y leerlo como falla empuja a forzar envíos inútiles (FR-026). - La jerarquía índice → hijos viaja como dato —quién declaró a quién y a qué profundidad—, no como sangría. La sangría no la anuncia un lector de pantalla ni sobrevive a ordenar por otra columna. """ from pathlib import Path import httpx import pytest from django.conf import settings from django.utils import timezone from apps.domains.models import AccessState from apps.jobs.budget import reserve from apps.jobs.models import Batch, BatchKind, BatchOrigin, BatchState from apps.sitemaps.models import Sitemap, SitemapKind, SitemapSource, SubmitResult from apps.sitemaps.services import register_sitemap, sync_domain from tests.doubles import FakeClient from tests.factories import create_account, create_credential, create_domain pytestmark = pytest.mark.django_db FIXTURES = Path(settings.BASE_DIR) / 'tests' / 'fixtures' / 'sitemaps' INDEX = 'https://ejemplo.test/sitemap.xml' CHILD_ONE = 'https://ejemplo.test/sitemap-1.xml' CHILD_TWO = 'https://ejemplo.test/sitemap-2.xml' def fake_site(**routes) -> httpx.Client: """ Simula el sitio del usuario, no Google. Va por separado del doble de Search Console porque son dos interlocutores distintos: leer un sitemap es tráfico hacia el sitio del usuario y no gasta cuota de Google. Una ruta que no está devuelve 404, que es el caso que la pantalla tiene que saber contar. """ def respond(request: httpx.Request) -> httpx.Response: body = routes.get(str(request.url)) if body is None: return httpx.Response(404, text='no encontrado') return httpx.Response(200, content=body) return httpx.Client(transport=httpx.MockTransport(respond)) def read_file(name: str) -> bytes: return (FIXTURES / name).read_bytes() @pytest.fixture def domain(client): account = create_account() create_credential(account) domain = create_domain(account) client.force_login(account) return domain def screen(client, domain, **params) -> dict: """Las props que recibe la vista, pedidas como las pide el navegador.""" response = client.get( f'/search-console/domains/{domain.id}/sitemaps', params, HTTP_X_INERTIA='true' ) assert response.status_code == 200 return response.json()['props'] def locations(props: dict) -> list[str]: return [row['location'] for row in props['sitemaps']] def row(props: dict, location: str) -> dict: return next(r for r in props['sitemaps'] if r['location'] == location) def synchronize(domain, **routes) -> Batch: return sync_domain( domain, origin=BatchOrigin.MANUAL, client=FakeClient(submit_sitemap={}), http=fake_site(**routes), ) @pytest.fixture def publish(monkeypatch): """ Publica archivos en el sitio simulado, para el alta. El alta comprueba la dirección antes de guardarla (T094), y quien la dispara —una petición del navegador— no tiene por dónde pasar un cliente HTTP. Así que el sitio se inyecta donde el servicio lee. """ from apps.sitemaps.reader import read as real_read def publisher(**routes): site = fake_site(**routes) monkeypatch.setattr( 'apps.sitemaps.services.read', lambda location, *, client=None, timeout=None: real_read(location, client=site), ) return publisher # --- Jerarquía -------------------------------------------------------------- class TestTheHierarchy: def test_the_children_come_out_below_the_index_that_declared_them(self, client, domain): """ El orden por omisión es el del árbol, no el alfabético. Alfabéticamente `sitemap-1.xml` cae antes que `sitemap.xml`, así que un orden por ubicación pondría a los hijos arriba de su índice y la lista dejaría de leerse como lo que es. """ register_sitemap(domain, INDEX) synchronize( domain, **{ INDEX: read_file('indice.xml'), CHILD_ONE: read_file('urlset.xml'), CHILD_TWO: read_file('urlset_sin_espacio_de_nombres.xml'), }, ) assert locations(screen(client, domain)) == [INDEX, CHILD_ONE, CHILD_TWO] def test_each_child_says_in_words_who_declared_it(self, client, domain): """ La sangría sola no comunica la pertenencia. No la anuncia un lector de pantalla y desaparece en cuanto se ordena por otra columna, así que el nombre del índice viaja en la fila. """ register_sitemap(domain, INDEX) synchronize( domain, **{ INDEX: read_file('indice.xml'), CHILD_ONE: read_file('urlset.xml'), CHILD_TWO: read_file('urlset_sin_espacio_de_nombres.xml'), }, ) props = screen(client, domain) assert row(props, INDEX)['parent_location'] is None assert row(props, INDEX)['depth'] == 0 assert row(props, CHILD_ONE)['parent_location'] == INDEX assert row(props, CHILD_ONE)['depth'] == 1 assert row(props, CHILD_ONE)['source'] == SitemapSource.DISCOVERED def test_the_belonging_survives_sorting_by_another_column(self, client, domain): """ Con otro orden el índice puede quedar treinta renglones más arriba. La contigüidad se pierde —es una propiedad del orden—, pero de quién depende cada archivo es una propiedad de la fila y tiene que seguir ahí. """ register_sitemap(domain, INDEX) synchronize( domain, **{ INDEX: read_file('indice.xml'), CHILD_ONE: read_file('urlset.xml'), CHILD_TWO: read_file('urlset_sin_espacio_de_nombres.xml'), }, ) props = screen(client, domain) assert row(props, CHILD_ONE)['parent_location'] == INDEX def test_an_index_inside_another_arrives_with_its_depth(self, client, domain): root = Sitemap.objects.create(domain=domain, location=INDEX, kind=SitemapKind.INDEX) middle = Sitemap.objects.create( domain=domain, location=CHILD_ONE, parent=root, kind=SitemapKind.INDEX, source=SitemapSource.DISCOVERED, ) Sitemap.objects.create( domain=domain, location='https://ejemplo.test/hoja.xml', parent=middle, source=SitemapSource.DISCOVERED, ) props = screen(client, domain) assert locations(props) == [INDEX, CHILD_ONE, 'https://ejemplo.test/hoja.xml'] assert row(props, 'https://ejemplo.test/hoja.xml')['depth'] == 2 # --- Contrato de la tabla local --------------------------------------------- class TestTheLongTable: def test_all_rows_arrive_for_client_side_pagination(self, client, domain): for number in range(103): Sitemap.objects.create( domain=domain, location=f'https://ejemplo.test/s-{number:03d}.xml' ) assert len(screen(client, domain)['sitemaps']) == 103 def test_rows_carry_the_raw_values_used_by_client_filters(self, client, domain): Sitemap.objects.create( domain=domain, location=INDEX, kind=SitemapKind.INDEX, source=SitemapSource.DECLARED, last_submit_result=SubmitResult.OK, ) sitemap = row(screen(client, domain), INDEX) assert sitemap['kind'] == SitemapKind.INDEX assert sitemap['source'] == SitemapSource.DECLARED assert sitemap['last_submit_result'] == SubmitResult.OK def test_filter_options_use_the_same_raw_values_as_rows(self, client, domain): props = screen(client, domain) assert {option['value'] for option in props['options']['kind']} == set( SitemapKind.values ) assert {option['value'] for option in props['options']['source']} == set( SitemapSource.values ) assert {option['value'] for option in props['options']['result']} == set( SubmitResult.values ) # --- Lo que dice cada fila -------------------------------------------------- class TestWhatEachRowSays: def test_a_failed_read_does_not_dirty_the_result_of_the_last_submit(self, client, domain): """ El invariante central de esta pantalla. Que hoy no se pueda descargar el archivo no cambia lo que Google recibió el martes. Marcar el envío como fallido pondría «Falló» al lado de la fecha de un envío que sí salió bien, y la única pregunta que esta vista responde quedaría contestada al revés. """ register_sitemap(domain, CHILD_ONE) synchronize(domain, **{CHILD_ONE: read_file('urlset.xml')}) synchronize(domain) # el sitio dejó de responder current = row(screen(client, domain), CHILD_ONE) assert current['last_submit_result'] == SubmitResult.OK assert current['last_submitted_at'] is not None assert current['last_error_code'] == 'UNREACHABLE' assert '404' in current['last_error'] def test_a_file_that_is_not_a_sitemap_is_distinguished_from_one_that_does_not_download( self, client, domain ): """ Son dos problemas y se arreglan en dos lugares distintos. «No se pudo descargar» manda a revisar la dirección; «se descargó pero no es un sitemap» manda a revisar el archivo. Un único «falló» obligaría a leer la oración entera para saber a cuál de los dos ir. """ register_sitemap(domain, CHILD_ONE) register_sitemap(domain, CHILD_TWO) synchronize(domain, **{CHILD_ONE: read_file('pagina_de_error.html')}) props = screen(client, domain) assert row(props, CHILD_ONE)['last_error_code'] == 'MALFORMED' assert row(props, CHILD_TWO)['last_error_code'] == 'UNREACHABLE' def test_the_foreign_urls_stay_in_the_row_and_not_only_in_the_batch(self, client, domain): """ El lote lo reemplaza el de la próxima corrida. Si el dato viviera únicamente ahí, mañana nadie podría responder cuál de los sitemaps era el que declaraba direcciones de otro sitio. """ register_sitemap(domain, CHILD_ONE) with_foreign = read_file('urlset.xml').replace( b'https://ejemplo.test/nota/dos', b'https://otro-sitio.test/nota/dos' ) synchronize(domain, **{CHILD_ONE: with_foreign}) current = row(screen(client, domain), CHILD_ONE) assert current['foreign_url_count'] == 1 # No es una falla del sitemap: se leyó bien y se envió bien. assert current['last_error'] is None assert current['last_submit_result'] == SubmitResult.OK def test_the_foreign_urls_notice_disappears_when_the_site_fixes_it(self, client, domain): register_sitemap(domain, CHILD_ONE) with_foreign = read_file('urlset.xml').replace( b'https://ejemplo.test/nota/dos', b'https://otro-sitio.test/nota/dos' ) synchronize(domain, **{CHILD_ONE: with_foreign}) synchronize(domain, **{CHILD_ONE: read_file('urlset.xml')}) assert row(screen(client, domain), CHILD_ONE)['foreign_url_count'] == 0 def test_unchanged_arrives_as_its_own_result_and_not_as_an_error(self, client, domain): """ No reenviar lo que no cambió es el comportamiento correcto (FR-026). Llega como su propio resultado y sin error asociado, para que la pantalla no pueda pintarlo como una falla y empujar a forzar envíos inútiles que gastan cupo. """ register_sitemap(domain, CHILD_ONE) synchronize(domain, **{CHILD_ONE: read_file('urlset.xml')}) synchronize(domain, **{CHILD_ONE: read_file('urlset.xml')}) current = row(screen(client, domain), CHILD_ONE) assert current['last_submit_result'] == SubmitResult.SKIPPED_UNCHANGED assert current['last_error'] is None assert current['last_error_code'] is None def test_a_change_that_could_not_be_submitted_is_not_read_as_unchanged(self, client, domain): """ Es el único caso en que el resultado guardado alcanza para mentir. Cuando el archivo cambió y el envío no llegó a salir —falló, o no quedaba cupo—, `last_submit_result` sigue siendo el de la corrida anterior. Sin el dato de que hay cambios sin enviar, la fila diría que Google tiene lo último justo cuando no lo tiene. """ register_sitemap(domain, CHILD_ONE) synchronize(domain, **{CHILD_ONE: read_file('urlset.xml')}) reserve(domain, domain.automatic_budget) modified = read_file('urlset.xml').replace(b'/nota/dos', b'/nota/tres') synchronize(domain, **{CHILD_ONE: modified}) current = row(screen(client, domain), CHILD_ONE) assert current['last_submit_result'] == SubmitResult.OK assert current['needs_submit'] is True def test_the_url_count_is_the_one_the_file_declares(self, client, domain): """Son URLs declaradas, no indexadas: describen el archivo, no la cobertura.""" register_sitemap(domain, CHILD_ONE) synchronize(domain, **{CHILD_ONE: read_file('urlset.xml')}) assert row(screen(client, domain), CHILD_ONE)['url_count'] == 3 # --- Trabajo en curso ------------------------------------------------------- class TestWhatIsBeingRead: def test_with_an_open_batch_what_is_left_to_read_says_so(self, client, domain): """ «Sin leer» y «leyendo» se ven igual sin la fecha de la corrida abierta. Un sitemap recién registrado parecería roto justo mientras el trabajo está ocurriendo. """ Batch.objects.create( domain=domain, kind=BatchKind.SITEMAP_SYNC, origin=BatchOrigin.MANUAL, state=BatchState.RUNNING, started_at=timezone.now(), ) Sitemap.objects.create(domain=domain, location=INDEX) Sitemap.objects.create(domain=domain, location=CHILD_ONE, last_read_at=timezone.now()) props = screen(client, domain) assert row(props, INDEX)['reading'] is True assert row(props, CHILD_ONE)['reading'] is False assert props['last_sync']['is_terminal'] is False def test_with_the_batch_finished_no_row_says_it_is_being_read(self, client, domain): Batch.objects.create( domain=domain, kind=BatchKind.SITEMAP_SYNC, origin=BatchOrigin.MANUAL, state=BatchState.PARTIAL, started_at=timezone.now(), finished_at=timezone.now(), ) Sitemap.objects.create(domain=domain, location=INDEX) props = screen(client, domain) assert row(props, INDEX)['reading'] is False assert props['last_sync']['state'] == BatchState.PARTIAL def test_the_partial_batch_arrives_with_its_figures_and_its_reasons(self, client, domain): """ `PARTIAL` tiene que verse parcial sin leer las cifras (R-C). Van las cifras y no un texto ya resuelto porque el avance se dibuja desde ellas: una barra llena diría «se procesaron todos» justo en el caso en que eso es falso. """ register_sitemap(domain, CHILD_ONE) register_sitemap(domain, CHILD_TWO) synchronize(domain, **{CHILD_ONE: read_file('urlset.xml')}) batch = screen(client, domain)['last_sync'] assert batch['state'] == BatchState.PARTIAL assert batch['processed_items'] == 1 assert batch['failed_items'] == 1 assert any('404' in reason for reason in batch['summary']['errors']) # --- Notas contra fallas ---------------------------------------------------- class TestTheNotesAreNotFailures: """ La ficha del lote separa los dos bloques con `item`, fila por fila. El servidor parte cada entrada de `summary['errors']` en «dirección: motivo» y deja `item` en nulo en todo lo que no tenga forma de dirección. Ese criterio sólo sirve mientras escribir la dirección adelante y contar un fallido sean la misma decisión: cualquier anotación que empiece con una dirección sin serlo se lee como un intento que falló, y un corte por cupo aparece bajo «Los ítems que fallaron», que es justo lo que R-C prohíbe. """ def detail(self, client, batch) -> dict: response = client.get(f'/search-console/batches/{batch.id}', HTTP_X_INERTIA='true') assert response.status_code == 200 return response.json()['props'] def test_only_the_attempts_that_failed_arrive_with_an_item(self, client, domain): """ El aviso de URLs ajenas no es una falla: el archivo se leyó y se envió. Se escribía empezando por la dirección del sitemap, con lo que llegaba a la pantalla indistinguible de un intento fallido aunque no incrementara `failed_items`. """ register_sitemap(domain, CHILD_ONE) with_foreign = read_file('urlset.xml').replace( b'https://ejemplo.test/nota/dos', b'https://otro-sitio.test/nota/dos' ) batch = synchronize(domain, **{CHILD_ONE: with_foreign}) props = self.detail(client, batch) with_item = [entry for entry in props['failures'] if entry['item'] is not None] assert props['batch']['failed_items'] == 0 assert props['failures'], 'La corrida tenía que dejar su aviso anotado.' assert with_item == [], ( 'Una anotación que no es una falla llegó con forma de intento fallido: ' f'{with_item}. La pantalla la va a listar como un ítem que falló.' ) def test_a_read_that_failed_does_arrive_with_its_item(self, client, domain): """La contracara: lo que sí falló tiene que seguir teniendo su dirección.""" register_sitemap(domain, CHILD_ONE) batch = synchronize(domain) props = self.detail(client, batch) assert props['batch']['failed_items'] == 1 assert [entry['item'] for entry in props['failures']] == [CHILD_ONE] def test_a_run_stopped_by_quota_closes_partial_without_leaving_any_gap(self, client, domain): """ El caso que obliga a que la barra no llegue al borde (R-C). Se leyó todo lo que había para leer y no quedó cupo para enviar. El lote cierra en `PARTIAL` con `processed == total`, `failed == 0` y nada pendiente: las cifras son exactamente las de un lote terminado, así que la incompletitud no puede quedar dicha sólo por el color del badge. """ register_sitemap(domain, CHILD_ONE) synchronize(domain, **{CHILD_ONE: read_file('urlset.xml')}) reserve(domain, domain.automatic_budget) modified = read_file('urlset.xml').replace(b'/nota/dos', b'/nota/tres') batch = synchronize(domain, **{CHILD_ONE: modified}) data = self.detail(client, batch)['batch'] assert data['state'] == BatchState.PARTIAL assert data['processed_items'] == data['total_items'] assert data['failed_items'] == 0 assert data['pending_items'] == 0 # --- Alta ------------------------------------------------------------------- class TestTheRegistration: def test_a_valid_location_ends_up_registered_and_visible(self, client, domain, publish): publish(**{INDEX: read_file('indice.xml')}) response = client.post( f'/search-console/domains/{domain.id}/sitemaps/new', {'location': INDEX} ) assert response.status_code == 302 assert locations(screen(client, domain)) == [INDEX] def test_registering_reads_the_file_but_neither_saves_it_nor_calls_google( self, client, domain, publish ): """ El alta descarga el archivo una vez, y sólo para poder rechazarlo (T094). Lo que **no** hace es guardar nada de esa lectura ni encolar trabajo. Si anotara la cantidad de URLs, la fila diría que descubrió cuatro mientras la cobertura sigue en cero, y esa diferencia no la podría explicar nadie: registrarlas es trabajo de la sincronización. """ publish(**{INDEX: read_file('indice.xml')}) client.post(f'/search-console/domains/{domain.id}/sitemaps/new', {'location': INDEX}) sitemap = Sitemap.objects.get(domain=domain) assert sitemap.last_read_at is None assert sitemap.url_count == 0 assert not Batch.objects.filter(domain=domain).exists() def test_a_location_without_a_scheme_returns_with_the_error_next_to_the_field( self, client, domain ): response = client.post( f'/search-console/domains/{domain.id}/sitemaps/new', {'location': 'ejemplo.com/sitemap.xml'}, ) assert response.headers['Location'] == f'/search-console/domains/{domain.id}/sitemaps?new=' assert not Sitemap.objects.filter(domain=domain).exists() # Se comprueba sobre las props que efectivamente recibe la pantalla: lo # que importa es que el mensaje llegue a donde se dibuja el campo. assert screen(client, domain)['errors']['location']['code'] == 'SITEMAP_LOCATION_RELATIVE' def test_an_empty_location_too(self, client, domain): client.post(f'/search-console/domains/{domain.id}/sitemaps/new', {'location': ' '}) assert not Sitemap.objects.filter(domain=domain).exists() assert screen(client, domain)['errors']['location'] def test_registering_the_same_location_twice_does_not_duplicate_it( self, client, domain, publish ): publish(**{INDEX: read_file('indice.xml')}) client.post(f'/search-console/domains/{domain.id}/sitemaps/new', {'location': INDEX}) client.post(f'/search-console/domains/{domain.id}/sitemaps/new', {'location': INDEX}) assert Sitemap.objects.filter(domain=domain).count() == 1 def test_an_unreadable_address_is_rejected_at_the_form(self, client, domain, publish): """ Se rechaza al pegar la dirección, no al día siguiente (T094). Antes el alta guardaba cualquier cosa y el error aparecía en el resumen del lote de la madrugada, lejos del formulario donde se escribió. """ publish() client.post(f'/search-console/domains/{domain.id}/sitemaps/new', {'location': INDEX}) assert not Sitemap.objects.filter(domain=domain).exists() assert screen(client, domain)['errors']['location']['code'] == 'UNREACHABLE' # --- Sincronización --------------------------------------------------------- class TestTheSynchronization: def test_a_domain_without_confirmed_access_brings_the_code_of_the_error(self, client): """ El código viaja junto al mensaje (RT-08). Sin él, «no hay credencial comprobada» y «este dominio no tiene acceso» llegarían como dos oraciones que la pantalla no puede distinguir, y la acción que ofrece —revisar la conexión, comprobar el acceso— tendría que ser la misma para las dos. """ account = create_account() create_credential(account) without_access = create_domain(account, access_state=AccessState.ACCESS_LOST) client.force_login(account) response = client.post(f'/search-console/domains/{without_access.id}/sync') assert response.status_code == 302 assert response.headers['Location'] == ( f'/search-console/domains/{without_access.id}/sitemaps?tab=sync' ) errors = screen(client, without_access)['errors'] assert errors['sync']['code'] == 'DOMAIN_NOT_OPERATIONAL' # El dominio viaja por su identificador y su estado, no por su nombre # metido en una oración: es lo que deja a la pantalla armar el enlace a # la ficha donde el acceso se resuelve. assert errors['sync']['params']['domain_id'] == str(without_access.id) def test_without_any_run_the_screen_does_not_invent_a_batch(self, client, domain): assert screen(client, domain)['last_sync'] is None # --- Acceso ----------------------------------------------------------------- class TestTheAccess: def test_sitemaps_created_by_another_profile_are_shared(self, client, domain): other_account = create_account(email='ajena@ejemplo.test') other_domain = create_domain(other_account, hostname='ajeno.test') assert client.get(f'/search-console/domains/{other_domain.id}/sitemaps').status_code == 200 def test_without_a_session_it_sends_to_the_login(self, client): account = create_account(email='otra@ejemplo.test') their_domain = create_domain(account, hostname='suyo.test') response = client.get(f'/search-console/domains/{their_domain.id}/sitemaps') assert response.headers['Location'].startswith('/login')