File size: 3,022 Bytes
25ae7fe | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 | import asyncio
import httpx
import time
import sys
import os
# Set encoding for Windows CLI
if sys.platform == "win32":
import codecs
sys.stdout = codecs.getwriter("utf-8")(sys.stdout.detach())
async def test_system():
print("\n" + "="*50)
print("π MEIH SYSTEM HEALTH CHECK")
print("="*50 + "\n")
async with httpx.AsyncClient(timeout=30.0) as client:
# 1. Test FlareSolverr
print("π‘ Checking FlareSolverr...")
try:
resp = await client.get("http://localhost:8191/health")
if resp.status_code == 200:
print("β
FlareSolverr: ONLINE")
else:
print(f"β FlareSolverr: ERROR (Status {resp.status_code})")
except Exception as e:
print(f"β FlareSolverr: OFFLINE ({e})")
# 2. Test FastAPI Backend
print("\nβοΈ Checking FastAPI Backend...")
try:
resp = await client.get("http://localhost:8000/")
if resp.status_code == 200:
print("β
Backend: ONLINE")
data = resp.json()
print(f" Mirror Active: {data.get('active_mirror')}")
print(f" Engine Status: {data.get('engine_status')}")
else:
print(f"β Backend: ERROR (Status {resp.status_code})")
except Exception as e:
print(f"β Backend: OFFLINE ({e})")
# 3. Test Scrapper (Latest Movies)
print("\n㪠Testing Movie Scrapper (Live Fetch)...")
try:
start_time = time.time()
resp = await client.get("http://localhost:8000/latest")
duration = time.time() - start_time
if resp.status_code == 200:
items = resp.json()
print(f"β
Scrapper: SUCCESS")
print(f" Items Found: {len(items)}")
print(f" Time Taken: {duration:.2f}s")
if items:
print(f" Top Item: {items[0]['title']}")
else:
print(f"β Scrapper: FAILED (Status {resp.status_code})")
except Exception as e:
print(f"β Scrapper: ERROR ({e})")
# 4. Test Category (Fast Path)
print("\nπ Testing Category (Prefetch Integrity)...")
try:
start_time = time.time()
resp = await client.get("http://localhost:8000/category/arabic-movies")
duration = time.time() - start_time
if resp.status_code == 200:
items = resp.json()
print(f"β
Category Path: STABLE")
print(f" Items: {len(items)}")
print(f" Time Taken: {duration:.2f}s (Should be < 0.5s if cached)")
else:
print(f"β Category: FAILED")
except Exception as e:
print(f"β Category: ERROR ({e})")
print("\n" + "="*50)
print("β¨ ALL TESTS COMPLETED")
print("="*50 + "\n")
if __name__ == "__main__":
asyncio.run(test_system())
|