129 lines
4.8 KiB
Python
129 lines
4.8 KiB
Python
"""HTTP deployment checks using only synthetic data; never print credentials."""
|
|
|
|
import argparse
|
|
import http.cookiejar
|
|
import json
|
|
import os
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from dotenv import dotenv_values
|
|
|
|
|
|
def check(base_url: str, service_token: str, require_projection=False):
|
|
jar = http.cookiejar.CookieJar()
|
|
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
|
|
|
|
def api(path, body=None, *, trusted=False, method=None, expected=200):
|
|
headers = {"Content-Type": "application/json"}
|
|
if trusted:
|
|
headers["Authorization"] = "Bearer " + service_token
|
|
request = urllib.request.Request(
|
|
base_url.rstrip("/") + path,
|
|
data=None if body is None else json.dumps(body).encode(),
|
|
headers=headers,
|
|
method=method,
|
|
)
|
|
try:
|
|
with opener.open(request, timeout=10) as response:
|
|
assert response.status == expected, "Unexpected HTTP status"
|
|
return json.load(response)
|
|
except urllib.error.HTTPError as error:
|
|
if error.code == expected:
|
|
return None
|
|
raise RuntimeError(
|
|
f"HTTP smoke request failed: {path} HTTP {error.code}"
|
|
) from None
|
|
|
|
assert api("/health") == {"status": "ok"}
|
|
for module in ("events", "users", "ads", "identity"):
|
|
assert api(f"/v1/{module}/health") == {"status": "ok", "module": module}
|
|
assert "/v1/identity/stitch" in api("/openapi.json")["paths"]
|
|
anon = api("/v1/identity/anonymous", {})
|
|
assert any(cookie.name == "adcloud_anonymous" for cookie in jar), (
|
|
"Missing anonymous cookie"
|
|
)
|
|
event = api(
|
|
"/v1/events",
|
|
{
|
|
"event_id": str(uuid4()),
|
|
"name": "deployment.smoke",
|
|
"properties": {"source": "gitea-actions"},
|
|
},
|
|
)
|
|
assert event["anonymous_id"] == anon["anonymous_id"], "Anonymous cookie not reused"
|
|
if service_token:
|
|
api("/v1/users/" + anon["master_id"], expected=401)
|
|
api(
|
|
"/v1/users/" + anon["master_id"],
|
|
{"attributes": {"smoke": True}},
|
|
trusted=True,
|
|
method="PATCH",
|
|
)
|
|
linked = api(
|
|
"/v1/identity/stitch",
|
|
{
|
|
"anonymous_token": anon["anonymous_token"],
|
|
"identifiers": [
|
|
{"kind": "email", "value": "smoke-" + uuid4().hex + "@example.com"}
|
|
],
|
|
},
|
|
trusted=True,
|
|
)
|
|
profile = api("/v1/users/" + linked["master_id"], trusted=True)
|
|
assert profile["event_count"] == 1 and profile["attributes"]["smoke"] is True
|
|
history = api("/v1/events?master_id=" + linked["master_id"], trusted=True)
|
|
assert history["items"][0]["id"] == event["event_id"]
|
|
graph = api("/v1/identity/" + linked["master_id"] + "/graph", trusted=True)
|
|
assert len(graph["identifiers"]) == 1
|
|
if require_projection:
|
|
storage = api("/v1/identity/storage", trusted=True)
|
|
assert storage["graph_backend"] == "postgresql"
|
|
assert storage["canonical_profile_format"] == "jsonb"
|
|
assert storage["profile_backend"] == "cassandra"
|
|
for _ in range(60):
|
|
profile = api("/v1/users/" + linked["master_id"], trusted=True)
|
|
if profile["projection_status"] == "current":
|
|
break
|
|
time.sleep(1)
|
|
assert profile["projection_status"] == "current", (
|
|
"Worker projection not current"
|
|
)
|
|
assert profile["attributes"]["smoke"] is True
|
|
print(
|
|
"HTTP smoke passed: routes, anonymous cookie, events"
|
|
+ (", auth, stitching, profile and history" if service_token else "")
|
|
)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--base-url", default="http://127.0.0.1:8000")
|
|
parser.add_argument(
|
|
"--public-only",
|
|
action="store_true",
|
|
help="Do not read or send the service token",
|
|
)
|
|
parser.add_argument("--require-projection", action="store_true")
|
|
args = parser.parse_args()
|
|
token = ""
|
|
if not args.public_only:
|
|
stored = dotenv_values(
|
|
Path(os.getenv("IDENTITY_BOOTSTRAP_FILE", "/config/identity.env"))
|
|
)
|
|
token = os.getenv("SERVICE_TOKEN") or stored.get("SERVICE_TOKEN") or ""
|
|
if not token:
|
|
raise SystemExit(
|
|
"Configure SERVICE_TOKEN or mount the deployment config volume"
|
|
)
|
|
if args.require_projection and args.public_only:
|
|
parser.error("Projection checks require a service token")
|
|
check(args.base_url, token, args.require_projection)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|