""" Contrato de cobertura: listado, resumen y exportación. Lo que se verifica en la exportación no es que el archivo exista, sino que **cada fila lleve su fecha de obtención y la zona en que está expresada**. El archivo se abre en otro lado, meses después: un estado sin esos dos datos deja de significar algo, y en una planilla nadie tiene a mano el contexto que la pantalla sí daba. El juego de columnas es el mismo que muestra la tabla de V7, y por eso se fija acá entero. Una columna que se ve en pantalla y no está en el archivo convierte al CSV en el resumen de otra cosa; y el archivo es lo que sobrevive a la pantalla, así que la diferencia se descubre cuando ya no hay a quién preguntarle. """ import csv import io from datetime import datetime from zoneinfo import ZoneInfo import pytest from django.conf import settings as django_settings from rest_framework.test import APIClient from apps.accounts.models import ApiKey from apps.coverage.services import inspect_domain from apps.jobs.models import BatchOrigin from apps.sitemaps.models import CoverageState, Url from tests.doubles import FakeClient, response from tests.factories import create_account, create_credential, create_domain pytestmark = pytest.mark.django_db @pytest.fixture def account(): account = create_account() create_credential(account) return account @pytest.fixture def domain(account): dom = create_domain(account) Url.objects.create(domain=dom, loc='https://ejemplo.test/uno', in_sitemap=True) Url.objects.create(domain=dom, loc='https://ejemplo.test/dos', in_sitemap=True) return dom @pytest.fixture def client(account): api = APIClient() _, plain_text = ApiKey.issue(account=account, name='pipeline') api.credentials(HTTP_AUTHORIZATION=f'Api-Key {plain_text}') return api def test_the_listing_brings_the_state_next_to_its_date(client, domain): data = client.get(f'/api/v1/domains/{domain.id}/coverage').json() assert data['count'] == 2 for item in data['items']: assert 'coverage_state' in item assert 'last_checked_at' in item def test_a_url_that_was_not_checked_travels_as_unknown_and_without_a_date(client, domain): item = client.get(f'/api/v1/domains/{domain.id}/coverage').json()['items'][0] assert item['coverage_state'] == CoverageState.UNKNOWN # Nulo, no una fecha inventada ni la de creación de la fila. assert item['last_checked_at'] is None def test_the_summary_leaves_the_unchecked_ones_out_of_the_percentage(client, domain): inspect_domain( domain, origin=BatchOrigin.MANUAL, urls=[Url.objects.filter(domain=domain).first()], client=FakeClient(inspect_url=response('url_inspection_indexada')), ) data = client.get(f'/api/v1/domains/{domain.id}/coverage/summary').json() assert data['with_data'] == 1 assert data['without_data'] == 1 assert data['indexed_percentage'] == 100.0 def test_the_filter_by_state_works(client, domain): inspect_domain( domain, origin=BatchOrigin.MANUAL, urls=[Url.objects.filter(domain=domain).first()], client=FakeClient(inspect_url=response('url_inspection_indexada')), ) # El filtro viaja en su forma pública; la respuesta sigue trayendo el código. indexed = client.get(f'/api/v1/domains/{domain.id}/coverage?state=indexed').json() without_data = client.get(f'/api/v1/domains/{domain.id}/coverage?state=unknown').json() assert indexed['count'] == 1 assert without_data['count'] == 1 def test_the_export_carries_the_date_and_the_time_zone_in_each_row(client, domain): inspect_domain( domain, origin=BatchOrigin.MANUAL, urls=[Url.objects.filter(domain=domain).first()], client=FakeClient(inspect_url=response('url_inspection_indexada')), ) http_response = client.get(f'/api/v1/domains/{domain.id}/coverage/export') content = b''.join(http_response.streaming_content).decode('utf-8') assert http_response.status_code == 200 assert 'attachment' in http_response['Content-Disposition'] assert domain.hostname in http_response['Content-Disposition'] rows = list(csv.reader(io.StringIO(content))) header, *body = [row for row in rows if row] # La cabecera va en inglés y no se traduce: el archivo lo lee una planilla o # un script, y un nombre de columna que cambia con el idioma de quien exportó # rompe cualquier fórmula armada contra la exportación anterior. assert header == [ 'url', 'state', # La etiqueta legible del código de al lado. Es la única columna que # acompaña al idioma de la cuenta. 'state_label', 'fetched_at', # El texto original de Google. Es lo que permite entender un «sin # indexar, otro motivo» sin volver a gastar cuota preguntándole. 'google_reason', 'google_canonical', # Cuándo Google visitó la página, que no es cuándo se lo preguntamos. 'last_crawl', 'in_sitemap', 'display_timezone', ] assert len(body) == 2 indexed = next(row for row in body if row[1] == CoverageState.INDEXED) assert indexed[3], 'La fila con dato tiene que traer su fecha de obtención' assert indexed[4] == 'Submitted and indexed' assert indexed[8] == django_settings.DISPLAY_TIMEZONE def test_the_dates_in_the_file_are_in_the_time_zone_the_file_declares(client, domain): """ La última columna nombra una zona; las fechas tienen que estar en ésa. Declarar una zona al lado de un instante escrito en UTC son dos afirmaciones sobre la misma fecha que no coinciden, y quien abre la planilla meses después no tiene cómo saber cuál manda. La zona se lee de la configuración y no se escribe acá: es un valor de la instalación, y un test que lo fija en duro convierte cambiarlo en una falla que no dice qué se rompió. """ inspect_domain( domain, origin=BatchOrigin.MANUAL, urls=[Url.objects.filter(domain=domain).first()], client=FakeClient(inspect_url=response('url_inspection_indexada')), ) http_response = client.get(f'/api/v1/domains/{domain.id}/coverage/export') content = b''.join(http_response.streaming_content).decode('utf-8') body = [row for row in csv.reader(io.StringIO(content)) if row][1:] indexed = next(row for row in body if row[1] == CoverageState.INDEXED) obtained = datetime.fromisoformat(indexed[3]) assert obtained.utcoffset() == datetime.now(ZoneInfo(indexed[8])).utcoffset() def test_the_export_does_not_make_up_a_date_for_what_was_not_checked(client, domain): http_response = client.get(f'/api/v1/domains/{domain.id}/coverage/export') content = b''.join(http_response.streaming_content).decode('utf-8') rows = [row for row in csv.reader(io.StringIO(content)) if row][1:] for row in rows: assert row[1] == CoverageState.UNKNOWN assert row[3] == '' def test_enqueuing_inspections_returns_202_with_the_batch_still_queued(client, domain, monkeypatch): """ 202 y el lote en cola: el trabajo se encoló, no ocurrió. Antes la petición recorría las URLs una por una —hasta dos mil consultas a Google— y contestaba `202` igual, o sea que la respuesta afirmaba haber encolado algo que en realidad se estaba ejecutando ahí mismo. Lo que se fija acá es que la petición vuelva sin haber consultado nada, y que el estado que devuelve sea el que corresponde: todavía no empezó. """ enqueued = [] monkeypatch.setattr( 'apps.coverage.tasks.inspect_one_domain.delay', lambda *args, **kwargs: enqueued.append((args, kwargs)), ) http_response = client.post(f'/api/v1/domains/{domain.id}/inspections') body = http_response.json() assert http_response.status_code == 202 assert body['kind'] == 'URL_INSPECTION' assert body['state'] == 'QUEUED' assert body['total_items'] == 0 assert len(enqueued) == 1 def test_a_domain_created_by_another_profile_is_in_the_company_scope(client): foreign = create_account(email='ajena@ejemplo.test') other = create_domain(foreign, hostname='ajeno.test') assert client.get(f'/api/v1/domains/{other.id}/coverage').status_code == 200