更新
This commit is contained in:
@@ -0,0 +1,133 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Post-deploy smoke test for Heicode Manager. Runs entirely from this
|
||||
laptop against https://code.xinghanlab.com — no SSH required.
|
||||
|
||||
Checks:
|
||||
1. /api/status returns 200 with success:true (proves the binary
|
||||
is running and the DB is reachable).
|
||||
2. New build hash present (HTML title or static asset includes a
|
||||
fresh content hash, indirectly confirming the new image is live).
|
||||
3. /api/devices returns 401 when called without auth (proves
|
||||
the route is wired AND not accidentally public).
|
||||
4. /api/devices/pair returns 401 without auth (same reasoning).
|
||||
5. /v1/messages with a clearly-invalid V2 envelope returns 401
|
||||
AND carries X-Heicode-Auth-Error + X-Heicode-Server-Time headers
|
||||
— this is the V2 diagnostic-header contract we shipped this round.
|
||||
|
||||
Exit code: 0 == all green, 1 == any failure (caller escalates).
|
||||
"""
|
||||
|
||||
import sys
|
||||
import time
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
import json
|
||||
|
||||
BASE = "https://code.xinghanlab.com"
|
||||
SESSION_UA = "heicode-smoke/1.0"
|
||||
|
||||
|
||||
def http(method: str, path: str, headers: dict | None = None, body: bytes | None = None):
|
||||
"""Returns (status_code, headers_dict, body_text). Does NOT raise on 4xx/5xx."""
|
||||
req = urllib.request.Request(BASE + path, method=method, data=body)
|
||||
req.add_header("User-Agent", SESSION_UA)
|
||||
for k, v in (headers or {}).items():
|
||||
req.add_header(k, v)
|
||||
try:
|
||||
resp = urllib.request.urlopen(req, timeout=15)
|
||||
return resp.status, dict(resp.headers), resp.read().decode("utf-8", "replace")
|
||||
except urllib.error.HTTPError as e:
|
||||
return e.code, dict(e.headers), e.read().decode("utf-8", "replace")
|
||||
|
||||
|
||||
FAILS: list[str] = []
|
||||
|
||||
|
||||
def check(label: str, ok: bool, detail: str = "") -> None:
|
||||
mark = "OK " if ok else "FAIL"
|
||||
print(f"[{mark}] {label}" + (f" -- {detail}" if detail else ""))
|
||||
if not ok:
|
||||
FAILS.append(label)
|
||||
|
||||
|
||||
print(f"Smoke testing {BASE} ...\n")
|
||||
|
||||
# 1. /api/status reachable + healthy
|
||||
code, hdrs, body = http("GET", "/api/status")
|
||||
ok = code == 200 and '"success":true' in body
|
||||
check("/api/status returns 200 with success:true", ok,
|
||||
f"HTTP {code}, body starts with {body[:80]!r}")
|
||||
|
||||
# 2. /api/devices without auth → 401 (route exists + protected)
|
||||
code, hdrs, body = http("GET", "/api/devices/")
|
||||
ok = code in (401, 403)
|
||||
check("/api/devices unauthenticated → 401/403", ok, f"HTTP {code}")
|
||||
|
||||
# 3. /api/devices/pair without auth → 401
|
||||
code, hdrs, body = http("POST", "/api/devices/pair",
|
||||
headers={"Content-Type": "application/json"},
|
||||
body=b'{"device_id":"x","public_key":"","fingerprint":""}')
|
||||
ok = code in (401, 403)
|
||||
check("/api/devices/pair unauthenticated → 401/403", ok, f"HTTP {code}")
|
||||
|
||||
# 4. /v1/messages with bogus V2 envelope → 401 + diagnostic headers.
|
||||
# We construct a request that:
|
||||
# - sets Content-Encoding: heicode-aead-v1 (forces V2 dispatcher)
|
||||
# - sets junk for the 6 X-Heicode-* headers
|
||||
# - sends a too-short body (12 bytes — fails AEAD nonce check)
|
||||
# Expected: HTTP 401, X-Heicode-Auth-Error header present,
|
||||
# X-Heicode-Server-Time header present and parseable as int.
|
||||
bogus_headers = {
|
||||
"Content-Encoding": "heicode-aead-v1",
|
||||
"X-Heicode-Device-Id": "smoke-test-bogus",
|
||||
"X-Heicode-Timestamp": str(int(time.time() * 1000)),
|
||||
"X-Heicode-Nonce": "deadbeef" * 4,
|
||||
"X-Heicode-Fingerprint": "0" * 64,
|
||||
"X-Heicode-Eph-Pubkey": "QUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUFBQUE=",
|
||||
"X-Heicode-Signature": "BBBB",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
code, hdrs, body = http("POST", "/v1/messages",
|
||||
headers=bogus_headers,
|
||||
body=b"\x00" * 12)
|
||||
|
||||
ok_401 = code == 401
|
||||
check("/v1/messages with bogus V2 envelope → 401", ok_401, f"HTTP {code}")
|
||||
|
||||
# Header lookup is case-insensitive in HTTP but our http() dict isn't.
|
||||
# Build a lowercase view.
|
||||
hdrs_lc = {k.lower(): v for k, v in hdrs.items()}
|
||||
diag_code = hdrs_lc.get("x-heicode-auth-error", "")
|
||||
diag_time = hdrs_lc.get("x-heicode-server-time", "")
|
||||
check("V2 401 carries X-Heicode-Auth-Error header", bool(diag_code), f"value={diag_code!r}")
|
||||
|
||||
ok_time = False
|
||||
try:
|
||||
if diag_time and abs(int(diag_time) - int(time.time() * 1000)) < 60_000:
|
||||
ok_time = True
|
||||
except ValueError:
|
||||
pass
|
||||
check("V2 401 carries X-Heicode-Server-Time within 60s of local clock",
|
||||
ok_time, f"value={diag_time!r}")
|
||||
|
||||
# 5. Sanity: status data has live numbers (start_time freshly set if rebuild
|
||||
# happened just now).
|
||||
code, hdrs, body = http("GET", "/api/status")
|
||||
try:
|
||||
obj = json.loads(body)
|
||||
st = obj.get("data", {}).get("start_time", 0)
|
||||
age = int(time.time()) - st
|
||||
print(f"\nManager start_time = {st} (age {age}s)")
|
||||
check("Manager start_time looks recent (<30 min)", 0 <= age < 30 * 60,
|
||||
f"age={age}s")
|
||||
except Exception as e:
|
||||
check("Manager status JSON parseable", False, repr(e))
|
||||
|
||||
if FAILS:
|
||||
print(f"\n{len(FAILS)} smoke check(s) FAILED:")
|
||||
for f in FAILS:
|
||||
print(f" - {f}")
|
||||
sys.exit(1)
|
||||
|
||||
print("\nALL SMOKE CHECKS PASSED")
|
||||
Reference in New Issue
Block a user