138 lines
5.5 KiB
Python
138 lines
5.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Verify that a deployed dashboard is usable, not merely alive."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import http.cookiejar
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from dataclasses import dataclass
|
|
|
|
|
|
class DeploymentVerificationError(RuntimeError):
|
|
"""A public deployment failed one observable user-flow check."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Response:
|
|
status: int
|
|
url: str
|
|
content_type: str
|
|
body: bytes
|
|
|
|
|
|
def _request(opener: urllib.request.OpenerDirector, request: str | urllib.request.Request, label: str) -> Response:
|
|
try:
|
|
with opener.open(request, timeout=10) as response:
|
|
return Response(
|
|
status=response.status,
|
|
url=response.geturl(),
|
|
content_type=response.headers.get_content_type(),
|
|
body=response.read(),
|
|
)
|
|
except urllib.error.HTTPError as error:
|
|
raise DeploymentVerificationError(f"{label} returned HTTP {error.code}") from None
|
|
except (OSError, urllib.error.URLError, TimeoutError):
|
|
raise DeploymentVerificationError(f"{label} could not be reached") from None
|
|
|
|
|
|
def verify_deployment(base_url: str, access_token: str) -> dict[str, str]:
|
|
"""Exercise liveness, readiness, sign-in, PWA scope, and mobile Home."""
|
|
parsed = urllib.parse.urlsplit(base_url)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
|
raise DeploymentVerificationError("deployment URL must be absolute HTTP(S)")
|
|
if not access_token:
|
|
raise DeploymentVerificationError("operator access token is required")
|
|
base_url = base_url.rstrip("/") + "/"
|
|
base_path = urllib.parse.urlsplit(base_url).path
|
|
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()))
|
|
|
|
health = _request(opener, urllib.parse.urljoin(base_url, "healthz"), "liveness")
|
|
if health.status != 200:
|
|
raise DeploymentVerificationError(f"liveness returned HTTP {health.status}")
|
|
|
|
readiness = _request(opener, urllib.parse.urljoin(base_url, "readyz"), "readiness")
|
|
if readiness.status != 200:
|
|
raise DeploymentVerificationError(f"readiness returned HTTP {readiness.status}")
|
|
|
|
public_entry = _request(opener, base_url, "public entry")
|
|
public_text = public_entry.body.decode("utf-8", errors="replace")
|
|
if public_entry.content_type != "text/html" or 'name="access_token"' not in public_text:
|
|
raise DeploymentVerificationError("public entry did not render the operator sign-in form")
|
|
|
|
manifest_response = _request(
|
|
opener, urllib.parse.urljoin(base_url, "manifest.webmanifest"), "manifest"
|
|
)
|
|
try:
|
|
manifest = json.loads(manifest_response.body)
|
|
except (UnicodeDecodeError, json.JSONDecodeError):
|
|
raise DeploymentVerificationError("manifest did not return valid JSON") from None
|
|
manifest_scope = urllib.parse.urljoin(manifest_response.url, str(manifest.get("scope", "")))
|
|
manifest_start = urllib.parse.urljoin(manifest_response.url, str(manifest.get("start_url", "")))
|
|
expected = urllib.parse.urlsplit(base_url)
|
|
resolved_scope = urllib.parse.urlsplit(manifest_scope)
|
|
resolved_start = urllib.parse.urlsplit(manifest_start)
|
|
if (
|
|
(resolved_scope.scheme, resolved_scope.netloc, resolved_scope.path)
|
|
!= (expected.scheme, expected.netloc, expected.path)
|
|
or (resolved_start.scheme, resolved_start.netloc, resolved_start.path)
|
|
!= (expected.scheme, expected.netloc, expected.path)
|
|
):
|
|
raise DeploymentVerificationError("manifest scope or start URL escaped the deployment subpath")
|
|
|
|
encoded = json.dumps({
|
|
"access_token": access_token,
|
|
"device_label": "Deployment smoke verifier",
|
|
}).encode()
|
|
login_request = urllib.request.Request(
|
|
urllib.parse.urljoin(base_url, "api/v1/session"),
|
|
data=encoded,
|
|
headers={"Content-Type": "application/json", "Accept": "application/json"},
|
|
method="POST",
|
|
)
|
|
authenticated = _request(opener, login_request, "operator sign-in")
|
|
try:
|
|
authenticated_payload = json.loads(authenticated.body)
|
|
except (UnicodeDecodeError, json.JSONDecodeError):
|
|
raise DeploymentVerificationError("operator sign-in did not return valid JSON") from None
|
|
if authenticated_payload.get("authenticated") is not True:
|
|
raise DeploymentVerificationError("operator sign-in was not accepted")
|
|
|
|
home = _request(opener, base_url, "authenticated Home")
|
|
home_text = home.body.decode("utf-8", errors="replace")
|
|
if home.content_type != "text/html" or (
|
|
'id="mobile-task-dock"' not in home_text or 'id="new-issue"' not in home_text
|
|
):
|
|
raise DeploymentVerificationError("operator sign-in did not reach the mobile dashboard Home")
|
|
|
|
return {
|
|
"health": "ok",
|
|
"readiness": "ready",
|
|
"public_entry": "login",
|
|
"manifest_scope": base_path,
|
|
"authenticated_home": "mobile",
|
|
}
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("url", help="Dashboard URL including its public subpath")
|
|
args = parser.parse_args(argv)
|
|
token = os.environ.get("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "")
|
|
try:
|
|
result = verify_deployment(args.url, token)
|
|
except DeploymentVerificationError as error:
|
|
print(f"deployment verification failed: {error}", file=sys.stderr)
|
|
return 1
|
|
print(json.dumps(result, sort_keys=True))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|