71 lines
2.7 KiB
Python
71 lines
2.7 KiB
Python
"""uptime-kuma — recipe-specific functional test (Phase 2 P3).
|
|
|
|
uptime-kuma's defining feature is **real-time monitoring** delivered over Socket.IO. The
|
|
recipe routes /socket.io/* to the app. A working uptime-kuma must complete the Engine.IO v4
|
|
polling handshake at `/socket.io/?EIO=4&transport=polling`:
|
|
- HTTP 200
|
|
- Body starts with `0{` — the Engine.IO `open` packet (type 0) + JSON metadata (sid, upgrades,
|
|
pingInterval, pingTimeout)
|
|
|
|
Non-vacuous: a working HTTP front-end with a wedged Socket.IO backend returns 404 or 502 here.
|
|
A misrouted nginx returns 404. Only a correctly-wired uptime-kuma + Engine.IO listener responds
|
|
with the Engine.IO open packet.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import ssl
|
|
import sys
|
|
import urllib.request
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "runner"))
|
|
from harness import http as harness_http # noqa: E402
|
|
|
|
|
|
def test_socketio_polling_handshake(live_app):
|
|
"""GET /socket.io/?EIO=4&transport=polling → 200, body starts with '0{' + valid JSON."""
|
|
url = f"https://{live_app}/socket.io/?EIO=4&transport=polling"
|
|
|
|
ctx = ssl.create_default_context()
|
|
ctx.check_hostname = False
|
|
ctx.verify_mode = ssl.CERT_NONE
|
|
|
|
def _fetch():
|
|
req = urllib.request.Request(url, method="GET")
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=15, context=ctx) as r:
|
|
if r.status != 200:
|
|
return None
|
|
body = r.read().decode(errors="replace")
|
|
except Exception: # noqa: BLE001
|
|
return None
|
|
if not body.startswith("0"):
|
|
return None
|
|
# Body shape: `<length>:<packet>` or just the packet on EIO v4. We accept either.
|
|
# Locate the leading `0{` (Engine.IO open packet)
|
|
idx = body.find("0{")
|
|
if idx < 0:
|
|
return None
|
|
json_str = body[idx + 1 :] # skip the leading "0"
|
|
# The JSON may be followed by other Engine.IO frames; take everything to the last '}'
|
|
end = json_str.rfind("}")
|
|
if end < 0:
|
|
return None
|
|
try:
|
|
parsed = json.loads(json_str[: end + 1])
|
|
except (json.JSONDecodeError, ValueError):
|
|
return None
|
|
# Engine.IO v4 'open' packet carries sid, upgrades, pingInterval, pingTimeout
|
|
if not isinstance(parsed, dict) or "sid" not in parsed:
|
|
return None
|
|
return parsed
|
|
|
|
info = harness_http.assert_converges(
|
|
_fetch, f"GET {url} returns Engine.IO open packet", max_wait=60, interval=3
|
|
)
|
|
# Standard Engine.IO open-packet fields
|
|
assert "sid" in info, f"no sid in handshake: {info!r}"
|
|
assert "pingInterval" in info, f"no pingInterval in handshake: {info!r}"
|