#Una prueba de velocidad no basta
Que cuarenta consultas terminen deprisa no demuestra que el programa use como máximo tres workers. Necesitamos observar los picos, provocar errores de forma controlada y comprobar el estado después de la salida. Esta página contiene las pruebas completas y explica qué concluye cada una.
Las pruebas usan el código de las dos páginas anteriores. Crean otro servidor de laboratorio con puerto 0: el sistema operativo le asigna un puerto libre. Se publica solo en loopback y se cierra al terminar. No es necesario arrancar manualmente laboratorio.py para ejecutar pruebas.py.
#1. Muchas entradas y capacidad finita
El primer caso entrega cuarenta rutas mediante un generador. Es un número mayor que los tres workers y las seis posiciones de cola. Aumenta la tasa únicamente en el laboratorio para que los tres workers puedan alcanzar HTTP a la vez.
Las aserciones exigen cuarenta resultados ok, pico de tres operaciones activas, pico de cola seis y como máximo tres puertos de cliente durante ese primer caso. Eso muestra reutilización de conexiones reales para este servidor HTTP/1.1. El contador de ámbitos HTTP del cliente y la observación TCP del servidor son mediciones distintas que se complementan.
Comparamos además asyncio.all_tasks() antes y después. El conjunto contiene la tarea de prueba que ya existía; al terminar una operación no deben aparecer otras nuevas todavía vivas. Es una comprobación en este proceso controlado, no un mecanismo para cancelar todas las tareas de un servidor compartido.
#2. Fallos individuales y tamaño del cuerpo
Un mismo lote contiene /error, /large y /ok. Deben aparecer http_error, too_large y ok en ese orden de entrada. Si el primer fallo derribase todo el lote, perderíamos el último resultado y la prueba fallaría.
Para /large se aceptan bloques de 1024 bytes hasta detectar 17 408 bytes frente al máximo de 16 384. No prometemos que el sistema operativo no haya recibido más datos en su buffer. Sí comprobamos que la aplicación no acumula los 32 KiB del cuerpo y sale del contexto de respuesta al detectar el exceso.
#3. Lectura lenta frente a duración total
/slow espera 600 ms antes de responder y recibe un timeout de lectura de 50 ms: esperamos ReadTimeout. /trickle entrega un bloque cada 80 ms, por debajo de su espera de lectura de 150 ms, pero recibe un plazo total de 180 ms: esperamos deadline.
Estas dos pruebas evitan confundir inactividad de transporte con duración total. Los márgenes del laboratorio son deliberados; una máquina extremadamente saturada puede retrasar el event loop y requerir ampliar proporcionalmente los tiempos de fixture y prueba. No conviertas una coincidencia de milisegundos en una garantía de producción.
#4. Tiempo de cola y deadline del lote
Con un worker, tasa de cinco inicios por segundo y plazo de entrada de 80 ms, algunas entradas vencerán antes de poder enviar HTTP. El número de peticiones debe ser menor que el de entradas. No enviamos trabajo que ya perdió su utilidad únicamente para completar una estadística.
Otro caso entrega cien rutas lentas, pero da al lote solo 80 ms. La salida debe indicar plazo_lote, conservar cualquier resultado terminado, declarar admitidos sin resultado y cerrar todas sus tareas. El servidor puede seguir unos instantes en la demora que ya empezó: cerrar el cliente no mata los hilos remotos.
#5. Cancelación externa y pool ocupado
La prueba de cancelación espera a que el servidor haya recibido una petición antes de cancelar al dueño. Así no se limita a cancelar una corrutina que todavía no arrancó. Después exige que CancelledError se propague y que no queden tareas nuevas vivas.
La prueba de pool abre un streaming y lo mantiene sin consumir con un cliente de una sola conexión. La segunda petición debe recibir PoolTimeout. Al salir del contexto de streaming, otra consulta vuelve a funcionar. Demuestra por qué cerrar respuestas es parte de la capacidad del sistema, no solo de la limpieza final.
#Código completo de las comprobaciones
assert expresa una propiedad que debe cumplirse; si no se cumple, Python lanza AssertionError y el programa falla. Ejecuta las pruebas sin la opción -O, que elimina estas comprobaciones. En una base de código más grande podrías llevarlas a unittest o pytest; aquí no añadimos otra biblioteca para entender el mecanismo.
El último bloque arranca el servidor en un hilo. try/finally garantiza solicitar su parada y esperar ese hilo aunque una prueba falle. El contexto exterior cierra el socket de escucha. Los hilos de respuesta pertenecen al fixture con demoras acotadas, tal como explicamos en su página.
Código de pruebas.py, listo para ejecutar con los dos archivos anteriores
import asyncio
import threading
import httpx
from comprobador import ejecutar
from laboratorio import Laboratorio
def sin_tareas_nuevas(antes):
assert asyncio.all_tasks() <= antes
def cerrado(informe):
m = informe["metricas"]
assert m["activos"] == m["http_activos"] == 0
assert m["cliente_cerrado"] and m["tareas_terminadas"]
async def pruebas(servidor):
base = f"http://127.0.0.1:{servidor.server_port}"
antes = asyncio.all_tasks()
informe = await ejecutar(
(f"/ok?i={i}" for i in range(40)), base_url=base,
plazo=5, plazo_lote=5, por_segundo=200,
)
assert len(informe["resultados"]) == 40
assert all(r["estado"] == "ok" for r in informe["resultados"])
assert informe["metricas"]["pico_activos"] == 3
assert informe["metricas"]["pico_http"] == 3
assert informe["metricas"]["pico_cola"] == 6
with servidor.lock:
assert len(servidor.puertos_cliente) <= 3
cerrado(informe)
sin_tareas_nuevas(antes)
print("40 entradas: 3 workers, cola 6, hasta 3 conexiones reutilizadas")
informe = await ejecutar(["/error", "/large", "/ok"], base_url=base)
assert [r["estado"] for r in informe["resultados"]] == [
"http_error", "too_large", "ok"
]
assert informe["resultados"][1]["bytes"] == 17_408
cerrado(informe)
print("Estado HTTP, límite de bytes y resultado válido conservados")
informe = await ejecutar(["/slow"], base_url=base, lectura=0.05)
assert informe["resultados"][0]["estado"] == "transport_timeout"
assert informe["resultados"][0]["detalle"] == "ReadTimeout"
cerrado(informe)
informe = await ejecutar(["/trickle"], base_url=base, plazo=0.18, lectura=0.15)
assert informe["resultados"][0]["estado"] == "deadline"
cerrado(informe)
print("ReadTimeout y deadline total comprobados por separado")
informe = await ejecutar(
["/ok"] * 8, base_url=base, workers=1, capacidad=2,
por_segundo=5, plazo=0.08,
)
assert any(r["estado"] == "deadline" for r in informe["resultados"])
assert informe["metricas"]["peticiones"] < 8
cerrado(informe)
print("Las entradas vencidas no generan peticiones innecesarias")
informe = await ejecutar(
["/slow"] * 100, base_url=base,
plazo=2, lectura=2, plazo_lote=0.08,
)
assert informe["motivo"] == "plazo_lote"
assert informe["metricas"]["sin_resultado"] > 0
cerrado(informe)
sin_tareas_nuevas(antes)
print("Deadline del lote: cierre completo y pendientes explícitos")
with servidor.lock:
previas = servidor.peticiones
tarea = asyncio.create_task(ejecutar(
["/slow"] * 20, base_url=base, plazo=2, lectura=2,
))
async with asyncio.timeout(2):
while True:
with servidor.lock:
empezada = servidor.peticiones > previas
if empezada:
break
await asyncio.sleep(0.005)
tarea.cancel()
try:
await tarea
except asyncio.CancelledError:
pass
else:
raise AssertionError("se perdió la cancelación externa")
sin_tareas_nuevas(antes)
print("Cancelación externa propagada, sin tareas hijas vivas")
timeout = httpx.Timeout(1, pool=0.03)
async with httpx.AsyncClient(
base_url=base, trust_env=False, timeout=timeout,
limits=httpx.Limits(max_connections=1),
) as cliente:
async with cliente.stream("GET", "/trickle"):
try:
await cliente.get("/ok")
except httpx.PoolTimeout:
pass
else:
raise AssertionError("la conexión ocupada debía agotar el pool")
assert (await cliente.get("/ok")).status_code == 200
print("PoolTimeout reproducido; la conexión vuelve a estar disponible al cerrar")
rechazado = False
try:
await ejecutar(["http://otro-destino/"], base_url=base)
except* ValueError:
rechazado = True
assert rechazado
print("Destino fuera del laboratorio rechazado")
sin_tareas_nuevas(antes)
if __name__ == "__main__":
with Laboratorio(("127.0.0.1", 0)) as servidor:
hilo = threading.Thread(
target=servidor.serve_forever,
kwargs={"poll_interval": 0.05}, daemon=True,
)
hilo.start()
try:
asyncio.run(pruebas(servidor), debug=True)
finally:
servidor.shutdown()
hilo.join()
print("Comprobaciones completas")
#Ejecutar y leer la evidencia
python -W error pruebas.py
-W error convierte advertencias en errores, lo que ayuda a detectar recursos o corrutinas mal administrados. El modo debug del bucle también está activado en el código. La salida termina con:
40 entradas: 3 workers, cola 6, hasta 3 conexiones reutilizadas
Estado HTTP, límite de bytes y resultado válido conservados
ReadTimeout y deadline total comprobados por separado
Las entradas vencidas no generan peticiones innecesarias
Deadline del lote: cierre completo y pendientes explícitos
Cancelación externa propagada, sin tareas hijas vivas
PoolTimeout reproducido; la conexión vuelve a estar disponible al cerrar
Destino fuera del laboratorio rechazado
Comprobaciones completas
Los tiempos de cada consulta no se fijan en la salida esperada. Las pruebas verifican estados, límites y cierre, que son las garantías de la herramienta. La prueba del pool utiliza su propio cliente porque pretende provocar deliberadamente una configuración que el motor normal evita.
#Ejercicios que obligan a razonar
Cambia capacidad=6 por capacidad=1 en el primer caso y ajusta únicamente la aserción de cola. Los resultados deben seguir siendo cuarenta, pero el productor esperará más. Cambia workers a uno y predice qué picos deben bajar; el límite de conexiones debe cambiar junto al número de workers.
Después elimina temporalmente async with cliente.stream(...) y abre respuestas manualmente sin cerrarlas. La prueba del pool te da un ejemplo del síntoma que producirías. Restaura el contexto; no conviertas ese defecto en una solución aumentando conexiones indefinidamente.
Por último, introduce un NameError en comprobar. El programa debe fallar y limpiar sus tareas, en vez de entregar un informe de falsos errores de red. Esa distinción protege la capacidad de diagnosticar el sistema que construyes.
#Cómo evolucionarlo sin perder el control
| Necesidad nueva | Cambio justificado | Garantía que debes volver a probar |
|---|---|---|
| Consultar tus servicios reales | Catálogo administrado de destinos HTTPS y credenciales fuera del código | Destinos permitidos, TLS, redirecciones y secretos ausentes de logs |
| Hacer reintentos | Presupuesto compartido y política por error | Cada intento respeta tasa y deadline; no multiplica efectos |
| Procesar millones de rutas | Fuente incremental y colector de resultados acotado | Memoria estable y backpressure de extremo a extremo |
| Conservar resultados tras reinicio | Escritura durable y estado de ejecución | Recuperación sin confundir pendiente con confirmado |
| Ejecutar periódicamente | Dueño de calendario y exclusión de lotes si procede | No solapar ejecuciones hasta saturar recursos |
| Varias instancias del servicio | Reparto o coordinación de cuotas y trabajos | La suma de instancias respeta la capacidad externa |
No añadas todas esas piezas de golpe. La herramienta actual tiene un contrato comprobado. Introduce una necesidad, explica qué propiedad cambia y añade una prueba que pueda detectar su incumplimiento.
#Qué se ha verificado y qué sigue siendo un límite
Se ejecutaron los bloques publicados con Python 3.12.3 y HTTPX 0.28.1, usando sockets HTTP reales en loopback. Se comprobaron los resultados, cuerpos, timeouts, límites de actividad/cola, reutilización, cancelación y cierre mostrados. También se ejecutaron los ejemplos de fundamentos, carrera, hilos y procesos de los capítulos previos.
No se ha medido capacidad de producción, HTTPS público, proxies, HTTP/2 ni una caída abrupta durante persistencia: el proyecto no promete esas propiedades. Una cola en memoria y un JSON impreso al final no conservan progreso ante un apagado. Los límites están explícitos para que puedas extender el diseño con una razón, no asumir garantías que todavía no existen.
Al terminar puedes cerrar tu servidor manual con Ctrl+C. El servidor creado por pruebas.py ya se cerró por su propio finally.