49 lines
1.7 KiB
Python
49 lines
1.7 KiB
Python
"""Smoke стенда: бэк, логин, фронт, прокси /api и прокси WebSocket."""
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
import httpx
|
|
import websockets
|
|
|
|
API = os.environ.get("PERF_API", "http://localhost:8000")
|
|
WEB = os.environ.get("PERF_WEB", "http://localhost:8080")
|
|
|
|
|
|
def check(cond: bool, msg: str) -> None:
|
|
if not cond:
|
|
raise AssertionError(msg)
|
|
|
|
|
|
async def ws_check(token: str) -> None:
|
|
url = WEB.replace("http", "ws", 1) + "/api/v1/ws/export"
|
|
async with websockets.connect(url, open_timeout=10) as ws:
|
|
await ws.send(json.dumps({"token": token}))
|
|
msg = json.loads(await asyncio.wait_for(ws.recv(), timeout=10))
|
|
check(msg.get("event") == "exports", f"ws: неожиданное сообщение {str(msg)[:200]}")
|
|
|
|
|
|
def main() -> None:
|
|
with httpx.Client(timeout=15) as c:
|
|
r = c.get(f"{API}/healthz")
|
|
check(r.status_code == 200, f"healthz: {r.status_code}")
|
|
r = c.post(f"{API}/api/v1/auth/login", json={"username": "admin", "password": "admin"})
|
|
check(r.status_code == 200, f"login: {r.status_code} {r.text[:200]}")
|
|
token = r.json().get("access_token")
|
|
check(bool(token), "login: нет access_token")
|
|
r = c.get(f"{WEB}/")
|
|
check(r.status_code == 200 and "<script" in r.text, f"web /: {r.status_code}")
|
|
r = c.get(f"{WEB}/api/v1/users/me", headers={"Authorization": f"Bearer {token}"})
|
|
check(r.status_code == 200, f"proxy users/me: {r.status_code} {r.text[:200]}")
|
|
asyncio.run(ws_check(token))
|
|
print("SMOKE OK")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except Exception as e: # noqa: BLE001
|
|
print(f"SMOKE FAIL: {type(e).__name__}: {e}")
|
|
sys.exit(1)
|