#!/usr/bin/env python
"""Live check of the public-link door (erp_platform.public) on a real DataK3 bucket.
tests/test_public.py proves the door on SQLite. What only the engine can answer is checked here, through the
REAL public app (ASGI, in process) over a REAL engine (erp_platform.livecheck.engine — your CLI login):
1. migrate creates the platform's tables from platform.sql, and verifies them;
2. the table holds the token's sha256, never the token;
3. a GET is SEEN — the event row, and the projection UPDATE (coalesce + 1) the engine must accept;
4. N concurrent claims of one link's act (link.use) → exactly 1 wins — the plain-INSERT property, which on
DataK3 may be decided at COMMIT; N concurrent POSTs of the act through the app → the handler acts once;
5. revoke / reissue / expiry → the one 404, byte for byte the same as an unknown token;
6. TRACK: the pixel counts one open, the click goes to the stored destination only.
BUCKET=erp-platform-test DODIL_ORG=dodil python scripts/live_public_check.py
"""
import json
import os
import subprocess
import threading
from collections import Counter
from datetime import timedelta
from urllib.parse import unquote, urlparse
import anyio
import httpx
from sqlalchemy import select, text
from erp_platform import livecheck as lc, public
from erp_platform.bricks import Brick
from erp_platform.public import Page
N = int(os.environ.get("THREADS", "10"))
BASE = "https://links.check.example"
os.environ.setdefault("PUBLIC_LINK_BASE_URL", BASE)
os.environ.setdefault("PUBLIC_REDIRECT_HOSTS", "dodil.io")
TABLES = [public.links.name, public.events.name]
CHK = Brick("pubcheck")
ACTS: list = []
_lock = threading.Lock()
@CHK.public("SIGN", subject_kind="contract")
def page(link, req):
return Page(f"Contract {link.subject_id}", "<form method='post'><button>Sign</button></form>")
@page.act
def sign(link, req):
with _lock:
ACTS.append(link.link_id)
return Page("Signed", "<p>Thank you.</p>")
APP = public.app(CHK.public_handlers, name="pubcheck-public")
def call(method, path, **kw):
async def go():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=APP), base_url="http://door") as c:
return await c.request(method, path, **kw)
return anyio.run(go)
def path_of(url):
return url[len(BASE):]
def migrate_platform() -> int:
"""erp_platform.db reads the pg password from DODIL_TOKEN locally: take it from `dodil data connect`
(the CLI login) into this process's environment only — never printed."""
cmd = [lc.DODIL, "data", "connect", lc.BUCKET, "-o", "json"] + (["--org", lc.ORG] if lc.ORG else [])
u = urlparse(json.loads(subprocess.run(cmd, capture_output=True, text=True, check=True).stdout)["pg"])
os.environ["DODIL_TOKEN"] = unquote(u.password or "")
from erp_platform import migrate
return migrate.migrate_platform()
def main() -> int:
lc.drop_all(TABLES)
lc.check("migrate creates and verifies the platform's tables from platform.sql", migrate_platform() == 0)
eng = lc.engine(pool_size=N + 2)
public.configure(eng.begin)
link_id, url = CHK.links.issue("contract", "C-1|1", "Kim@Client.example", "SIGN", actor="live-check")
token = url.rsplit("/", 1)[1]
r = call("GET", path_of(url))
lc.check("the page resolves through the public app", r.status_code == 200 and "Contract C-1|1" in r.text, r.status_code)
with eng.connect() as c:
dump = repr([dict(x._mapping) for t in (public.links, public.events) for x in c.execute(select(t))])
row = dict(c.execute(select(public.links)).first()._mapping)
lc.check("the table holds the sha256, never the token",
token not in dump and token[:16] not in dump and row["token_hash"] == public.hash_token(token))
call("GET", path_of(url))
with eng.connect() as c:
row = dict(c.execute(select(public.links)).first()._mapping)
st = CHK.links.status("contract", "C-1|1")[0]
lc.check("seen: two events, and the projection UPDATE (coalesce + 1) lands on DataK3",
st["seen_count"] == 2 and row["seen_count"] == 2 and row["first_seen_at"] is not None, (st, row))
# 4a. N concurrent claims of one act.
link = public._resolve(token, None)
wins, gate = [], threading.Barrier(N)
def claim():
gate.wait()
try:
wins.append(link.use())
except Exception as exc: # noqa: BLE001 — counted, and a failure
wins.append(type(exc).__name__)
threads = [threading.Thread(target=claim) for _ in range(N)]
[t.start() for t in threads]
[t.join(120) for t in threads]
print(f" {N} concurrent link.use(): {dict(Counter(map(str, wins)))}")
lc.check(f"{N} concurrent claims of one act: exactly 1 wins", Counter(wins) == Counter({True: 1, False: N - 1}), wins)
# 4b. N concurrent POSTs of the act, on a fresh link, through the app.
_, url2 = CHK.links.issue("contract", "C-2|1", "lee@client.example", "SIGN")
codes, gate2 = [], threading.Barrier(N)
def post():
gate2.wait()
codes.append(call("POST", path_of(url2)).status_code)
threads = [threading.Thread(target=post) for _ in range(N)]
[t.start() for t in threads]
[t.join(120) for t in threads]
print(f" {N} concurrent POSTs of the act: statuses {dict(Counter(codes))}, handler ran {len(ACTS)}x")
lc.check(f"{N} concurrent POSTs: the handler acts once, the rest get the 404 page",
len(ACTS) == 1 and Counter(codes) == Counter({200: 1, 404: N - 1}), codes)
# 5. the one 404.
unknown = call("GET", "/p/" + "A" * 43)
rid, revoked = CHK.links.issue("contract", "C-3|1", "r@client.example", "SIGN")
CHK.links.revoke(rid, actor="live-check", reason="wrong person")
_, expired = CHK.links.issue("contract", "C-3|1", "e@client.example", "SIGN", expires_in=timedelta(seconds=-1))
old_id, old = CHK.links.issue("contract", "C-4|1", "o@client.example", "SIGN")
new_id, new = CHK.links.reissue(old_id, actor="live-check")
replies = [call("GET", path_of(u)) for u in (revoked, expired, old)]
def shape(x):
return x.status_code, x.content, sorted((k, v) for k, v in x.headers.items() if k != "x-request-id")
lc.check("revoked, expired and reissued-away are the same 404 as an unknown token",
unknown.status_code == 404 and all(shape(x) == shape(unknown) for x in replies),
[x.status_code for x in replies])
st4 = {s["link_id"]: s for s in CHK.links.status("contract", "C-4|1")}
lc.check("reissue: the new link works and names the old; the old is revoked",
call("GET", path_of(new)).status_code == 200 and st4[new_id]["reissued_from"] == old_id
and st4[old_id]["state"] == "revoked")
# 6. TRACK.
_, pixel = CHK.links.issue("mail", "M-1", "kim@client.example", "TRACK")
for _ in range(3):
call("GET", path_of(pixel))
_, click = CHK.links.issue("mail", "M-1", "kim@client.example", "TRACK", destination="https://dodil.io/x")
cr = call("GET", path_of(click) + "?url=https://evil.example/", follow_redirects=False)
sm = CHK.links.status("mail", "M-1")
with eng.connect() as c:
kinds = Counter(r[0] for r in c.execute(text("SELECT kind FROM erp_public_link_event")))
lc.check("TRACK: 3 pixel hits are one open; the click goes to the stored destination only",
kinds["opened"] == 1 and kinds["clicked"] == 1 and cr.status_code == 303
and cr.headers.get("location") == "https://dodil.io/x" and any(s["opened"] for s in sm), (kinds, cr.status_code))
eng.dispose()
public.configure(None)
return lc.finish(TABLES)
if __name__ == "__main__":
raise SystemExit(main())