""" Dobles de Google y de OpenAI, alimentados con respuestas grabadas. El corte está en el cliente HTTP y no en nuestro código: se simula lo que el proveedor contesta, no lo que nosotros concluimos. Un doble que devolviera «la credencial es válida» probaría que sabemos escribir un `if`; uno que devuelve el JSON que Google devolvió de verdad prueba que sabemos leerlo. Las respuestas de `tests/fixtures/gsc/` son copias de las reales, con los datos identificatorios cambiados. Su forma es exactamente la que llega. """ import json from pathlib import Path from types import SimpleNamespace import httpx from django.conf import settings from googleapiclient.errors import HttpError from openai import APIConnectionError, APIStatusError, APITimeoutError, RateLimitError FIXTURES = Path(settings.BASE_DIR) / 'tests' / 'fixtures' / 'gsc' def response(name: str) -> dict: return json.loads((FIXTURES / f'{name}.json').read_text(encoding='utf-8')) class _Response: """Imita el objeto de respuesta que espera HttpError.""" def __init__(self, status: int): self.status = status self.reason = '' def http_error(name: str) -> HttpError: """Arma el error tal como lo levanta la biblioteca de Google.""" body = response(name) return HttpError(_Response(body['error']['code']), json.dumps(body).encode()) class FakeClient: """ Reemplaza a SearchConsoleClient sin tocar la red. Registra las reservas con las que se lo llamó: varios tests comprueban no sólo el resultado sino que la llamada se haya hecho con cupo concedido, que es el invariante que el producto no puede romper. """ def __init__(self, **results): self.results = results self.calls: list[tuple[str, object]] = [] def _resolve(self, operation, reservation): self.calls.append((operation, reservation)) if reservation is None or reservation.granted <= 0: raise AssertionError(f'«{operation}» se llamó sin reserva concedida.') result = self.results.get(operation) if isinstance(result, HttpError): # El doble traduce igual que el cliente real. Si levantara el error # crudo de Google, quien llama recibiría algo que en producción # nunca ve, y de paso la clasificación —que es la lógica que hay que # probar— quedaría sin ejercitar. from apps.gsc.client import _from_http_error raise _from_http_error(result) if isinstance(result, Exception): raise result if callable(result): return result() return result if result is not None else {} def list_sites(self, *, reservation): return self._resolve('list_sites', reservation) def get_site(self, *, property_uri, reservation): return self._resolve('get_site', reservation) def list_sitemaps(self, *, property_uri, reservation): return self._resolve('list_sitemaps', reservation) def submit_sitemap(self, *, property_uri, location, reservation): return self._resolve('submit_sitemap', reservation) def delete_sitemap(self, *, property_uri, location, reservation): return self._resolve('delete_sitemap', reservation) def inspect_url(self, *, property_uri, url, reservation): return self._resolve('inspect_url', reservation) def search_analytics( self, *, property_uri, start_date, end_date, dimensions, row_limit=25_000, start_row=0, ): """ La única operación del cliente que **no** pide reserva. Es a propósito y está declarada también en `test_gsc_guard`: el presupuesto de cuota modela el cupo agotable de la inspección de URLs, y Search Analytics tiene un límite de tasa que se renueva solo. Un doble que exigiera reserva acá impondría una regla que la aplicación decidió no tener, y haría fallar al módulo por cumplirla bien. La respuesta se declara por combinación de dimensiones, no una sola para todas: el módulo pregunta dos veces por rango —una con la dimensión de página y otra sin ella— y lo que hay que poder probar es justamente que las dos respuestas son distintas y no se suman. """ key = f'search_analytics:{",".join(dimensions)}' self.calls.append((key, None)) result = self.results.get(key, self.results.get('search_analytics')) if isinstance(result, Exception): raise result if callable(result): return result(start_date, end_date, start_row) return result if result is not None else {} class FakeIndexingClient: """ Reemplaza a `IndexingClient` sin tocar la red. Se le declara la respuesta **por dirección**, y no una sola para todas, porque lo que hay que poder probar es la mezcla: un lote donde la tercera falla y las dos primeras ya salieron es el caso que define el producto. A diferencia de `FakeClient`, este doble no exige reserva de cupo: la indexación no lleva contador propio a propósito, y un doble que lo pidiera estaría imponiendo una regla que la aplicación decidió no tener. """ def __init__(self, answers=None, default=None): # Cada valor es un `IndexingCall` ya armado, o el error clasificado que # hay que devolver como tal. La clave es la dirección. self.answers = answers or {} self.default = default self.published: list[str] = [] def publish(self, url, *, notification_type='URL_UPDATED'): from apps.indexing.client import IndexingCall from apps.indexing.errors import message_for self.published.append(url) answer = self.answers.get(url, self.default) request_body = {'url': url, 'type': notification_type} if answer is None: # Lo que Google contesta cuando acepta: el eco del aviso con su # momento. Se copia la forma real porque es lo que se guarda como # evidencia, y una forma inventada probaría que sabemos guardar un # diccionario, no que guardamos lo que llegó. return IndexingCall( request_body=request_body, response_status=200, response_body={ 'urlNotificationMetadata': { 'url': url, 'latestUpdate': {'url': url, 'type': notification_type}, } }, ) if isinstance(answer, IndexingCall): return answer code, status = answer return IndexingCall( request_body=request_body, response_status=status, response_body={'error': {'code': status, 'status': code}}, error_code=code, error_message=message_for(code), ) # --------------------------------------------------------------------------- # OpenAI # --------------------------------------------------------------------------- class FakeResponses: """ El recurso `responses` del SDK de OpenAI, con las respuestas en orden. Se le dan los resultados **por llamada**, no uno para todas, porque lo que hay que poder probar es la secuencia: el proveedor que cae y después contesta, el JSON malo seguido del bueno. El último se repite si se lo llama más veces de las declaradas. Registra cada llamada con sus argumentos para que un test mire qué se le mandó al modelo. """ def __init__(self, *outcomes): self.outcomes = list(outcomes) self.calls: list[dict] = [] def create(self, **kwargs): self.calls.append(kwargs) outcome = self.outcomes.pop(0) if len(self.outcomes) > 1 else self.outcomes[0] if isinstance(outcome, Exception): raise outcome return outcome def vision_response(text: str, usage: dict | None = None): """ Lo que la Responses API devuelve, en lo que el servicio lee de ella. `output_text` es el texto ya juntado y `usage` sabe volcarse a diccionario, como el objeto real. """ return SimpleNamespace( output_text=text, usage=SimpleNamespace(model_dump=lambda: usage or {'total_tokens': 42}), ) def _openai_request(): return httpx.Request('POST', 'https://api.openai.com/v1/responses') def openai_status_error(status_code: int) -> APIStatusError: """Un error con código HTTP, tal como lo levanta el SDK.""" response = httpx.Response(status_code, request=_openai_request()) return APIStatusError('rechazado', response=response, body=None) def openai_rate_limit_error() -> RateLimitError: response = httpx.Response(429, request=_openai_request()) return RateLimitError('esperá', response=response, body=None) def openai_timeout_error() -> APITimeoutError: return APITimeoutError(_openai_request()) def openai_connection_error() -> APIConnectionError: return APIConnectionError(request=_openai_request())