96 lines
3.0 KiB
Python
96 lines
3.0 KiB
Python
"""Exercise a running local backend without printing service/anonymous tokens."""
|
|
|
|
import argparse
|
|
import http.cookiejar
|
|
import json
|
|
import urllib.request
|
|
from datetime import UTC, datetime
|
|
from uuid import uuid4
|
|
|
|
from adcloud_server.settings import Settings
|
|
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--base-url", default="http://127.0.0.1:8000")
|
|
args = parser.parse_args()
|
|
settings = Settings.from_env()
|
|
if not settings.service_token:
|
|
raise SystemExit("Configure SERVICE_TOKEN in .env first")
|
|
opener = urllib.request.build_opener(
|
|
urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar())
|
|
)
|
|
|
|
|
|
def api(path, body=None, trusted=False, method=None, extra_headers=None):
|
|
headers = {"Content-Type": "application/json"}
|
|
if trusted:
|
|
headers["Authorization"] = "Bearer " + settings.service_token
|
|
headers.update(extra_headers or {})
|
|
request = urllib.request.Request(
|
|
args.base_url.rstrip("/") + path,
|
|
data=None if body is None else json.dumps(body).encode(),
|
|
headers=headers,
|
|
method=method,
|
|
)
|
|
with opener.open(request, timeout=10) as response:
|
|
return json.load(response)
|
|
|
|
|
|
anon = api("/v1/identity/anonymous?rotate=true", {})
|
|
app_id = "demo-" + uuid4().hex
|
|
application = api(
|
|
"/v1/registry/apps", {"app_id": app_id, "name": "Identity demo"}, trusted=True
|
|
)
|
|
registration = "/v1/registry/apps/" + app_id + "/events/product.view/versions/1"
|
|
management = {"X-Adcloud-Management-Key": application["management_key"]}
|
|
api(
|
|
registration,
|
|
{"standard_event": "commerce.product.view", "description": "Demo product view"},
|
|
method="PUT",
|
|
extra_headers=management,
|
|
)
|
|
api(registration + "/publish", {}, extra_headers=management)
|
|
api(
|
|
"/v1/events",
|
|
{
|
|
"name": "product.view",
|
|
"app_id": app_id,
|
|
"event_version": 1,
|
|
"event_id": str(uuid4()),
|
|
"occurred_at": datetime.now(UTC).isoformat(),
|
|
"properties": {"product_id": "demo-running-shoe"},
|
|
},
|
|
extra_headers={"X-Adcloud-Key": application["collection_key"]},
|
|
)
|
|
api(
|
|
"/v1/users/" + anon["master_id"],
|
|
{"attributes": {"interest": "running", "unused": None}, "tags": ["runner"]},
|
|
trusted=True,
|
|
method="PATCH",
|
|
)
|
|
# Synthetic verified identity for this local demo, not an OTP implementation.
|
|
stitched = api(
|
|
"/v1/identity/stitch",
|
|
{
|
|
"anonymous_token": anon["anonymous_token"],
|
|
"identifiers": [
|
|
{"kind": "email", "value": "demo-" + uuid4().hex + "@example.com"}
|
|
],
|
|
},
|
|
trusted=True,
|
|
)
|
|
profile = api("/v1/users/" + stitched["master_id"], trusted=True)
|
|
assert profile["attributes"] == {"interest": "running"}
|
|
assert profile["tags"] == ["runner"] and profile["event_count"] == 1
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"master_id": profile["master_id"],
|
|
"attributes": profile["attributes"],
|
|
"tags": profile["tags"],
|
|
"event_count": profile["event_count"],
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
)
|