#!/usr/bin/env python3 """ stress_test.py — Stress test 200 RPS / 60s contra la API face-match-node. Propósito: Medir la infraestructura Node.js/Fastify/SQLite sin credenciales RENAPER reales. Cada transacción se registra en la tabla load_test_results de la DB SQLite. Al final emite un resumen completo de KPIs. Uso: python3 stress_test.py [--rps 200] [--duration 60] [--url http://localhost:3100] La API fallará con 502 (RENAPER no tiene credenciales) pero eso es esperado. Lo que medimos es latencia, throughput real, errores HTTP y distribución de respuestas. """ import asyncio import aiohttp import sqlite3 import time import json import argparse import sys import os import math import statistics from datetime import datetime from pathlib import Path # ─── Payload de prueba ─────────────────────────────────────────────────────── # Imagen base64 mínima válida (1x1 PNG blanco — pasa validación de base64) FAKE_SELFIE_B64 = ( "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhf" "DwAChwGA60e6kgAAAABJRU5ErkJggg==" ) # ─── Tabla en SQLite para registrar el stress test ─────────────────────────── CREATE_TABLE_SQL = """ CREATE TABLE IF NOT EXISTS load_test_results ( id INTEGER PRIMARY KEY AUTOINCREMENT, test_run_id TEXT NOT NULL, seq INTEGER NOT NULL, sent_at REAL NOT NULL, -- epoch float con microsegundos response_ms INTEGER, -- latencia en ms (NULL si timeout/error red) http_status INTEGER, -- 200/400/401/429/502/503/NULL api_status TEXT, -- campo "Status" de la respuesta JSON api_message TEXT, -- campo "Message" de la respuesta JSON error TEXT, -- excepción de red si hubo created_at TEXT DEFAULT (datetime('now','localtime')) ); CREATE INDEX IF NOT EXISTS idx_ltr_run ON load_test_results(test_run_id); CREATE INDEX IF NOT EXISTS idx_ltr_seq ON load_test_results(seq); """ INSERT_SQL = """ INSERT INTO load_test_results (test_run_id, seq, sent_at, response_ms, http_status, api_status, api_message, error) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """ # ─── Resultado de un request individual ────────────────────────────────────── class ReqResult: __slots__ = ("seq", "sent_at", "response_ms", "http_status", "api_status", "api_message", "error") def __init__(self, seq, sent_at, response_ms=None, http_status=None, api_status=None, api_message=None, error=None): self.seq = seq self.sent_at = sent_at self.response_ms = response_ms self.http_status = http_status self.api_status = api_status self.api_message = api_message self.error = error # ─── Worker async: envía un request y devuelve ReqResult ───────────────────── async def send_request(session: aiohttp.ClientSession, url: str, apikey: str, nationalid: str, gender: str, seq: int) -> ReqResult: sent_at = time.time() t0 = time.perf_counter() payload = { "apikey": apikey, "selfie_image": FAKE_SELFIE_B64, "nationalid": nationalid, "gender": gender, } try: async with session.post(url, data=payload, timeout=aiohttp.ClientTimeout(total=35)) as resp: elapsed_ms = int((time.perf_counter() - t0) * 1000) http_status = resp.status api_status = None api_message = None try: body = await resp.json(content_type=None) api_status = body.get("Status") api_message = body.get("Message") except Exception: pass return ReqResult(seq, sent_at, elapsed_ms, http_status, api_status, api_message) except asyncio.TimeoutError: elapsed_ms = int((time.perf_counter() - t0) * 1000) return ReqResult(seq, sent_at, elapsed_ms, None, None, None, "timeout") except Exception as exc: elapsed_ms = int((time.perf_counter() - t0) * 1000) return ReqResult(seq, sent_at, elapsed_ms, None, None, None, str(exc)[:200]) # ─── Productor de tokens de rate — genera exactamente rps requests/seg ──────── async def rate_limiter(rps: int, duration: float, queue: asyncio.Queue): """ Encola un token cada 1/rps segundos durante `duration` segundos. Usa un bucle con sleep adaptativo para mantener RPS exacto incluso cuando el event loop tiene jitter. """ interval = 1.0 / rps start = time.perf_counter() seq = 0 next_tick = start while True: now = time.perf_counter() elapsed = now - start if elapsed >= duration: break # Esperar hasta el próximo tick wait = next_tick - now if wait > 0: await asyncio.sleep(wait) seq += 1 await queue.put(seq) next_tick += interval # Señal de fin para todos los consumers for _ in range(rps * 2): await queue.put(None) # ─── Consumer: saca tokens del queue y lanza requests ───────────────────────── async def consumer(session, url, apikey, nationalid, gender, queue: asyncio.Queue, results: list, sem: asyncio.Semaphore): while True: seq = await queue.get() if seq is None: queue.task_done() break async with sem: result = await send_request(session, url, apikey, nationalid, gender, seq) results.append(result) queue.task_done() # ─── Persist batch a SQLite (en hilo separado para no bloquear el event loop) ─ def persist_results_sync(db_path: str, run_id: str, results: list[ReqResult]): conn = sqlite3.connect(db_path, timeout=30) conn.executescript(CREATE_TABLE_SQL) conn.commit() rows = [ (run_id, r.seq, r.sent_at, r.response_ms, r.http_status, r.api_status, r.api_message, r.error) for r in results ] conn.executemany(INSERT_SQL, rows) conn.commit() conn.close() # ─── Calculadora de percentiles ──────────────────────────────────────────────── def percentile(data: list, p: float) -> float: if not data: return 0.0 sorted_data = sorted(data) idx = (p / 100) * (len(sorted_data) - 1) lo = int(idx) hi = lo + 1 if hi >= len(sorted_data): return float(sorted_data[-1]) frac = idx - lo return sorted_data[lo] * (1 - frac) + sorted_data[hi] * frac # ─── Generador de resumen / KPIs ─────────────────────────────────────────────── def print_summary(results: list[ReqResult], run_id: str, target_rps: int, duration: float, actual_elapsed: float, db_path: str): total = len(results) if total == 0: print("\n[STRESS TEST] Sin resultados — ¿el servidor está corriendo?") return # Separar por tipo de resultado latencies = [r.response_ms for r in results if r.response_ms is not None] errors_net = [r for r in results if r.error is not None] timeouts = [r for r in results if r.error == "timeout"] by_status: dict[int | None, int] = {} for r in results: key = r.http_status by_status[key] = by_status.get(key, 0) + 1 by_api_status: dict[str | None, int] = {} for r in results: key = r.api_status by_api_status[key] = by_api_status.get(key, 0) + 1 # Throughput real real_rps = total / actual_elapsed if actual_elapsed > 0 else 0 # Latencias avg_ms = statistics.mean(latencies) if latencies else 0 median_ms = statistics.median(latencies) if latencies else 0 p90_ms = percentile(latencies, 90) p95_ms = percentile(latencies, 95) p99_ms = percentile(latencies, 99) max_ms = max(latencies) if latencies else 0 min_ms = min(latencies) if latencies else 0 stdev_ms = statistics.stdev(latencies) if len(latencies) > 1 else 0 # Tasa de éxito (HTTP 200) ok_200 = by_status.get(200, 0) success_rate = (ok_200 / total * 100) if total > 0 else 0 # Error rate error_rate = (len(errors_net) / total * 100) if total > 0 else 0 # Requests que llegaron al servidor (cualquier HTTP status) reached_server = total - len(errors_net) sep = "=" * 65 print(f"\n{sep}") print(f" STRESS TEST — RESUMEN DE KPIs") print(f" Run ID : {run_id}") print(f" Timestamp: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(sep) print(f"\n CONFIGURACIÓN") print(f" RPS objetivo : {target_rps}") print(f" Duración objetivo : {duration:.0f} s") print(f" Duración real : {actual_elapsed:.2f} s") print(f"\n THROUGHPUT") print(f" Requests enviados : {total:>8,}") print(f" RPS real : {real_rps:>8.2f}") print(f" Llegaron al server : {reached_server:>8,} ({reached_server/total*100:.1f}%)") print(f" HTTP 200 : {ok_200:>8,} ({success_rate:.1f}%)") print(f"\n LATENCIA (ms) — sobre {len(latencies):,} respuestas con tiempo medido") print(f" Mínimo : {min_ms:>8.0f} ms") print(f" Mediana (p50) : {median_ms:>8.0f} ms") print(f" Promedio : {avg_ms:>8.0f} ms") print(f" p90 : {p90_ms:>8.0f} ms") print(f" p95 : {p95_ms:>8.0f} ms") print(f" p99 : {p99_ms:>8.0f} ms") print(f" Máximo : {max_ms:>8.0f} ms") print(f" Desv. estándar : {stdev_ms:>8.0f} ms") print(f"\n DISTRIBUCIÓN HTTP STATUS") for status, count in sorted(by_status.items(), key=lambda x: (x[0] or 0)): label = str(status) if status else "sin-resp" pct = count / total * 100 bar = "█" * int(pct / 2) print(f" {label:>6} : {count:>6,} ({pct:5.1f}%) {bar}") print(f"\n DISTRIBUCIÓN API STATUS (campo 'Status' en body JSON)") for api_st, count in sorted(by_api_status.items(), key=lambda x: (x[0] or "")): label = str(api_st) if api_st else "sin-body" pct = count / total * 100 bar = "█" * int(pct / 2) print(f" {label:>10} : {count:>6,} ({pct:5.1f}%) {bar}") print(f"\n ERRORES") print(f" Errores de red : {len(errors_net):>6,} ({error_rate:.1f}%)") print(f" Timeouts : {len(timeouts):>6,}") if errors_net and len(errors_net) <= 5: for r in errors_net[:5]: print(f" seq={r.seq}: {r.error}") # Rate limiting hit hits_429 = by_status.get(429, 0) if hits_429: print(f"\n RATE LIMIT (429) : {hits_429:,} requests bloqueados") print(f"\n BASE DE DATOS") print(f" Registros guardados: {total:,} en load_test_results") print(f" DB path : {db_path}") print(f"\n KPIs CLAVE") if p99_ms > 0 and p99_ms < 500: kpi_latencia = "EXCELENTE (<500ms p99)" elif p99_ms < 1000: kpi_latencia = "BUENO (<1s p99)" elif p99_ms < 3000: kpi_latencia = "ACEPTABLE (<3s p99)" else: kpi_latencia = f"ALTO ({p99_ms:.0f}ms p99)" kpi_throughput = f"{real_rps:.1f} rps reales / {target_rps} objetivo" kpi_reach = f"{reached_server/total*100:.1f}% de requests llegaron al servidor" print(f" Latencia p99 : {kpi_latencia}") print(f" Throughput : {kpi_throughput}") print(f" Alcance : {kpi_reach}") print(f"\n{sep}\n") # ─── Main ───────────────────────────────────────────────────────────────────── async def main(): parser = argparse.ArgumentParser(description="Stress test 200 RPS face-match-node") parser.add_argument("--rps", type=int, default=200, help="Requests por segundo (default: 200)") parser.add_argument("--duration", type=int, default=60, help="Duración en segundos (default: 60)") parser.add_argument("--url", type=str, default="http://localhost:3100", help="URL base del servidor") parser.add_argument("--apikey", type=str, default=None, help="API key (default: lee APIKEY_MASTER del .env)") parser.add_argument("--nationalid", type=str, default="22222222", help="DNI de prueba (default: 22222222)") parser.add_argument("--gender", type=str, default="M", help="Género de prueba M|F (default: M)") parser.add_argument("--db", type=str, default=None, help="Ruta a la DB SQLite (default: auto-detecta)") parser.add_argument("--concurrency",type=int, default=400, help="Requests concurrentes máximos (default: 400)") args = parser.parse_args() # Detectar apikey desde .env si no se pasó apikey = args.apikey if not apikey: env_path = Path(__file__).parent / ".env" if env_path.exists(): for line in env_path.read_text().splitlines(): if line.startswith("APIKEY_MASTER="): apikey = line.split("=", 1)[1].strip() break if not apikey: apikey = "fide-dev-local-test-key-2026" # Detectar DB path db_path = args.db if not db_path: env_path = Path(__file__).parent / ".env" if env_path.exists(): for line in env_path.read_text().splitlines(): if line.startswith("DB_PATH="): db_path = line.split("=", 1)[1].strip() break if not db_path: db_path = str(Path(__file__).parent / "data" / "fide-facematch.db") # Resolver ruta relativa respecto al directorio del script if not os.path.isabs(db_path): db_path = str(Path(__file__).parent / db_path) endpoint = f"{args.url.rstrip('/')}/v3/face-match/" run_id = f"lt_{int(time.time())}" print(f"\n{'='*65}") print(f" STRESS TEST — face-match-node") print(f" RPS objetivo : {args.rps}") print(f" Duración : {args.duration}s") print(f" Endpoint : {endpoint}") print(f" API Key : {apikey[:8]}... (primeros 8 chars)") print(f" Run ID : {run_id}") print(f" DB : {db_path}") print(f" Concurrencia : {args.concurrency} slots simultáneos") print(f"{'='*65}\n") # Pre-crear la tabla en SQLite antes de empezar try: conn = sqlite3.connect(db_path, timeout=30) conn.executescript(CREATE_TABLE_SQL) conn.commit() conn.close() print(f" [DB] Tabla load_test_results lista en {db_path}") except Exception as e: print(f" [DB] ADVERTENCIA: No se pudo pre-crear la tabla: {e}") results: list[ReqResult] = [] queue: asyncio.Queue = asyncio.Queue(maxsize=args.rps * 4) sem = asyncio.Semaphore(args.concurrency) # Configurar el connector aiohttp con pool grande connector = aiohttp.TCPConnector( limit=args.concurrency + 50, limit_per_host=args.concurrency + 50, keepalive_timeout=30, enable_cleanup_closed=True, ) print(f" Iniciando test... presionar Ctrl+C para interrumpir\n") start_wall = time.perf_counter() async with aiohttp.ClientSession(connector=connector) as session: # Lanzar N consumers concurrentes consumer_count = min(args.concurrency, args.rps * 2) consumers = [ asyncio.create_task( consumer(session, endpoint, apikey, args.nationalid, args.gender, queue, results, sem) ) for _ in range(consumer_count) ] # Productor de rate await rate_limiter(args.rps, args.duration, queue) # Esperar a que el queue se vacíe await queue.join() # Cancelar consumers restantes (que esperan None del queue) for c in consumers: c.cancel() await asyncio.gather(*consumers, return_exceptions=True) actual_elapsed = time.perf_counter() - start_wall # Progress reportado total_sent = len(results) print(f"\n Test completado: {total_sent:,} requests en {actual_elapsed:.2f}s " f"({total_sent/actual_elapsed:.1f} rps reales)") # Persistir a SQLite en un loop sincrónico (evita bloquear el event loop) print(f" Guardando {total_sent:,} registros en SQLite...") try: loop = asyncio.get_event_loop() await loop.run_in_executor(None, persist_results_sync, db_path, run_id, results) print(f" Registros guardados correctamente.") except Exception as e: print(f" [DB] Error al guardar: {e}") # Emitir resumen print_summary(results, run_id, args.rps, args.duration, actual_elapsed, db_path) if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: print("\n\n [INTERRUMPIDO] El test fue detenido manualmente.\n") sys.exit(0)