Spaces:
Paused
Paused
| import asyncio | |
| import httpx | |
| import sys | |
| BASE_URL = "http://localhost:8000" | |
| async def run_e2e(): | |
| """E2E validation of all quick wins""" | |
| async with httpx.AsyncClient(timeout=120) as client: | |
| # 1. Health check | |
| print("1. Health check...") | |
| r = await client.get(f"{BASE_URL}/api/health") | |
| assert r.status_code == 200, f"Health: {r.status_code}" | |
| print(" β Health OK") | |
| # 2. Login admin | |
| print("2. Login admin...") | |
| r = await client.post(f"{BASE_URL}/api/auth/jwt/login", data={ | |
| "username": "crowsistemas@proton.me", "password": "admin123456" | |
| }) | |
| assert r.status_code == 200, f"Login: {r.status_code} - {r.text}" | |
| token = r.json()["access_token"] | |
| headers = {"Authorization": f"Bearer {token}"} | |
| print(f" β Login OK (RS256 token)") | |
| # 3. Persona report (cached) | |
| print("3. Persona report (cached)...") | |
| r = await client.get(f"{BASE_URL}/api/reports/persona/20123456789", headers=headers) | |
| assert r.status_code == 200, f"Persona: {r.status_code} - {r.text}" | |
| print(f" β Persona OK") | |
| # 4. Empresa report | |
| print("4. Empresa report...") | |
| r = await client.get(f"{BASE_URL}/api/reports/empresa/30123456789", headers=headers) | |
| assert r.status_code == 200, f"Empresa: {r.status_code} - {r.text}" | |
| print(f" β Empresa OK") | |
| # 5. Vehiculo report (fix 1.4) | |
| print("5. Vehiculo report...") | |
| r = await client.get(f"{BASE_URL}/api/reports/vehiculo/ABC123", headers=headers) | |
| assert r.status_code == 200, f"Vehiculo: {r.status_code} - {r.text}" | |
| print(f" β Vehiculo OK (no AttributeError)") | |
| # 6. Logout (fix 1.1) | |
| print("6. Logout...") | |
| r = await client.post(f"{BASE_URL}/api/auth/jwt/logout", headers=headers) | |
| assert r.status_code in (200, 204), f"Logout: {r.status_code} - {r.text}" | |
| print(f" β Logout OK (no NameError)") | |
| # 7. BCRA scraper with TLS (fix 1.2) | |
| print("7. BCRA scraper TLS test...") | |
| from app.scrapers.bcra import BcraScraper | |
| scraper = BcraScraper() | |
| result = await scraper.safe_fetch('20123456789') | |
| assert isinstance(result, dict), "BCRA should return dict" | |
| assert "bcra_situacion_actual" in result | |
| print(f" β BCRA TLS OK") | |
| print("\nπ ALL E2E TESTS PASSED!") | |
| return True | |
| if __name__ == "__main__": | |
| try: | |
| asyncio.run(run_e2e()) | |
| except Exception as e: | |
| print(f"\nβ FAILED: {e}") | |
| import traceback | |
| traceback.print_exc() | |
| sys.exit(1) |