codex-bot
0fc6c68a99
Issue persistent signed anonymous IDs and accept cookie/token-bound events. Resolve verified email, phone, OAuth, and namespaced third-party IDs into transactionally merged master identities with historical event aliases. Protect shared-device account switching, tenant isolation, and event retries. Store sparse canonical profiles as PostgreSQL JSONB or local SQLite JSON. Support an optional versioned Cassandra wide-column projection through a transactional outbox, retry worker, and immediate canonical read fallback. Document deployment settings and provide opt-in database integration tests. Validation: 43 local tests passed; 6 external database tests skipped. Ruff checks, formatting, and offline uv lock verification passed.
70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
"""Exercise a running local backend without printing service/anonymous tokens."""
|
|
|
|
import argparse
|
|
import http.cookiejar
|
|
import json
|
|
import urllib.request
|
|
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):
|
|
headers = {"Content-Type": "application/json"}
|
|
if trusted:
|
|
headers["Authorization"] = "Bearer " + settings.service_token
|
|
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", {})
|
|
api("/v1/events", {"name": "product.view", "properties": {"sku": "demo-running-shoe"}})
|
|
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,
|
|
)
|
|
)
|