|
@@ -80,6 +80,22 @@ Plan 10 phase 3 (capability awareness + UI):
|
|
|
|
|
|
|
|
Prints PASS/FAIL lines Statum-style; exits non-zero on any failure.
|
|
Prints PASS/FAIL lines Statum-style; exits non-zero on any failure.
|
|
|
|
|
|
|
|
|
|
+Plan 11 task D+E (UI + notifications), the `push` shard:
|
|
|
|
|
+
|
|
|
|
|
+- task page: progress feed above the status card, status + description
|
|
|
|
|
+ in a 50/50 grid, edit expander under the history expander, plan
|
|
|
|
|
+ behind an expander, answered blocks behind an expander; queue screen:
|
|
|
|
|
+ New Task expander above the filters; dashboard waiting-on-drafts
|
|
|
|
|
+ block with links to where Enqueue lives (pages shard)
|
|
|
|
|
+- real Web Push over VAPID: keypair generated at startup into
|
|
|
|
|
+ web-config.json, GET /sw.js service worker with click-through,
|
|
|
|
|
+ subscribe/unsubscribe + per-task Notify me actions, block events and
|
|
|
|
|
+ waiting-on-drafts pushed globally, first start/failure/completion to
|
|
|
|
|
+ opted-in users, best-effort queue with dispatcher-tick retry, 410/404
|
|
|
|
|
+ endpoint pruning — payloads verified protocol-level (RFC 8291
|
|
|
|
|
+ decryption + ES256 JWT signature) because the harness owns the
|
|
|
|
|
+ browser-side keys
|
|
|
|
|
+
|
|
|
Speed pass (plan 11, task 0):
|
|
Speed pass (plan 11, task 0):
|
|
|
|
|
|
|
|
- SHARDING: the scenario groups are partitioned into seven independent
|
|
- SHARDING: the scenario groups are partitioned into seven independent
|
|
@@ -119,6 +135,17 @@ import urllib.parse
|
|
|
import urllib.request
|
|
import urllib.request
|
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
|
|
|
|
|
|
+# Web Push protocol verification (plan 11 task E): the test harness
|
|
|
|
|
+# owns the browser-side keys, so it can do everything the push service
|
|
|
|
|
+# does — decrypt the RFC 8291 body and verify the ES256 VAPID JWT.
|
|
|
|
|
+import base64
|
|
|
|
|
+from cryptography.hazmat.primitives import hashes
|
|
|
|
|
+from cryptography.hazmat.primitives.asymmetric import ec
|
|
|
|
|
+from cryptography.hazmat.primitives.asymmetric.utils import encode_dss_signature
|
|
|
|
|
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
|
|
|
|
+from cryptography.hazmat.primitives.kdf.hkdf import HKDF
|
|
|
|
|
+from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
|
|
|
|
|
+
|
|
|
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
|
APP = os.path.abspath(sys.argv[1]) if len(sys.argv) > 1 else os.path.join(REPO, "builddir", "clanker-saddle")
|
|
APP = os.path.abspath(sys.argv[1]) if len(sys.argv) > 1 else os.path.join(REPO, "builddir", "clanker-saddle")
|
|
|
|
|
|
|
@@ -945,6 +972,129 @@ def dump_app_stacks(pid):
|
|
|
continue
|
|
continue
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
+# =============================================================================
|
|
|
|
|
+# Fake push service + Web Push protocol helpers (plan 11 task E)
|
|
|
|
|
+# =============================================================================
|
|
|
|
|
+
|
|
|
|
|
+def b64url(raw):
|
|
|
|
|
+ return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def b64url_dec(text):
|
|
|
|
|
+ return base64.urlsafe_b64decode(text + "=" * ((4 - len(text) % 4) % 4))
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+class FakePushService:
|
|
|
|
|
+ """In-process Web Push endpoint server: every subscription the tests
|
|
|
|
|
+ register points at a path here (a distinct "browser endpoint").
|
|
|
|
|
+ Responses are scriptable per path — a fixed status (the 410-prune
|
|
|
|
|
+ fixture) or N transient 500s (the retry-next-tick fixture)."""
|
|
|
|
|
+
|
|
|
|
|
+ def __init__(self):
|
|
|
|
|
+ self.requests = [] # [{"path", "headers", "body", "at"}]
|
|
|
|
|
+ self.fail_next = {} # path -> remaining 500s
|
|
|
|
|
+ self.status_for = {} # path -> fixed status
|
|
|
|
|
+ self.lock = threading.Lock()
|
|
|
|
|
+ self.port = None
|
|
|
|
|
+ self.server = None
|
|
|
|
|
+
|
|
|
|
|
+ def serve(self):
|
|
|
|
|
+ service = self
|
|
|
|
|
+
|
|
|
|
|
+ class Handler(BaseHTTPRequestHandler):
|
|
|
|
|
+ protocol_version = "HTTP/1.1"
|
|
|
|
|
+
|
|
|
|
|
+ def log_message(self, fmt, *args):
|
|
|
|
|
+ pass
|
|
|
|
|
+
|
|
|
|
|
+ def do_POST(self):
|
|
|
|
|
+ length = int(self.headers.get("Content-Length") or 0)
|
|
|
|
|
+ body = self.rfile.read(length)
|
|
|
|
|
+ status = 201
|
|
|
|
|
+ if self.path in service.status_for:
|
|
|
|
|
+ status = service.status_for[self.path]
|
|
|
|
|
+ elif service.fail_next.get(self.path, 0) > 0:
|
|
|
|
|
+ service.fail_next[self.path] -= 1
|
|
|
|
|
+ status = 500
|
|
|
|
|
+ with service.lock:
|
|
|
|
|
+ service.requests.append({
|
|
|
|
|
+ "path": self.path,
|
|
|
|
|
+ "headers": {key.lower(): value
|
|
|
|
|
+ for key, value in self.headers.items()},
|
|
|
|
|
+ "body": body,
|
|
|
|
|
+ "at": time.monotonic(),
|
|
|
|
|
+ })
|
|
|
|
|
+ payload = b"{}"
|
|
|
|
|
+ self.send_response(status)
|
|
|
|
|
+ self.send_header("Content-Type", "application/json")
|
|
|
|
|
+ self.send_header("Content-Length", str(len(payload)))
|
|
|
|
|
+ self.end_headers()
|
|
|
|
|
+ self.wfile.write(payload)
|
|
|
|
|
+
|
|
|
|
|
+ self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
|
|
|
+ self.port = self.server.server_address[1]
|
|
|
|
|
+ threading.Thread(target=self.server.serve_forever, daemon=True).start()
|
|
|
|
|
+ return self.server
|
|
|
|
|
+
|
|
|
|
|
+ def url(self, path):
|
|
|
|
|
+ return "http://127.0.0.1:%d%s" % (self.port, path)
|
|
|
|
|
+
|
|
|
|
|
+ def hits(self, prefix):
|
|
|
|
|
+ with self.lock:
|
|
|
|
|
+ return [request for request in self.requests
|
|
|
|
|
+ if request["path"].startswith(prefix)]
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def decrypt_webpush(request, client_private_key, client_public, auth_secret):
|
|
|
|
|
+ """RFC 8291 decryption of one captured push: returns
|
|
|
|
|
+ (record_size, payload dict) — the payload builder's exact output."""
|
|
|
|
|
+ body = request["body"]
|
|
|
|
|
+ salt = body[:16]
|
|
|
|
|
+ record_size = int.from_bytes(body[16:20], "big")
|
|
|
|
|
+ key_id_len = body[20]
|
|
|
|
|
+ eph_public = body[21:21 + key_id_len]
|
|
|
|
|
+ ciphertext = body[21 + key_id_len:]
|
|
|
|
|
+ eph_point = ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256R1(),
|
|
|
|
|
+ eph_public)
|
|
|
|
|
+ shared = client_private_key.exchange(ec.ECDH(), eph_point)
|
|
|
|
|
+ info = b"WebPush: info\x00" + eph_public + client_public
|
|
|
|
|
+ cek = HKDF(algorithm=hashes.SHA256(), length=16, salt=salt,
|
|
|
|
|
+ info=info).derive(shared + auth_secret)
|
|
|
|
|
+ nonce = HKDF(algorithm=hashes.SHA256(), length=12, salt=salt,
|
|
|
|
|
+ info=info).derive(shared + auth_secret)
|
|
|
|
|
+ plaintext = AESGCM(cek).decrypt(nonce, ciphertext, None)
|
|
|
|
|
+ return record_size, json.loads(plaintext[:-1]) # strip the 0x02 delimiter
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def verify_vapid(request, vapid_public, vapid_private):
|
|
|
|
|
+ """Verifies the ES256 VAPID JWT on one captured push; returns its
|
|
|
|
|
+ claims dict (aud/exp/sub) — raises on a bad signature."""
|
|
|
|
|
+ header = request["headers"].get("authorization", "")
|
|
|
|
|
+ assert header.startswith("vapid t="), "no vapid authorization header"
|
|
|
|
|
+ token_part = header[len("vapid t="):].split(",")[0]
|
|
|
|
|
+ sent_key = header.split("k=")[1]
|
|
|
|
|
+ head_b64, claims_b64, sig_b64 = token_part.split(".")
|
|
|
|
|
+ signature = b64url_dec(sig_b64)
|
|
|
|
|
+ key = ec.derive_private_key(
|
|
|
|
|
+ int.from_bytes(b64url_dec(vapid_private), "big"), ec.SECP256R1())
|
|
|
|
|
+ key.public_key().verify(
|
|
|
|
|
+ encode_dss_signature(int.from_bytes(signature[:32], "big"),
|
|
|
|
|
+ int.from_bytes(signature[32:], "big")),
|
|
|
|
|
+ (head_b64 + "." + claims_b64).encode(),
|
|
|
|
|
+ ec.ECDSA(hashes.SHA256()))
|
|
|
|
|
+ claims = json.loads(b64url_dec(claims_b64))
|
|
|
|
|
+ assert sent_key == vapid_public, "the k= key is not the app's VAPID key"
|
|
|
|
|
+ return claims
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def action_notify(result):
|
|
|
|
|
+ """The message of the first notify directive an action returned."""
|
|
|
|
|
+ for directive in result:
|
|
|
|
|
+ if directive.get("type") == "notify":
|
|
|
|
|
+ return directive.get("message", "")
|
|
|
|
|
+ return None
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
# =============================================================================
|
|
# =============================================================================
|
|
|
# Shard orchestration
|
|
# Shard orchestration
|
|
|
# =============================================================================
|
|
# =============================================================================
|
|
@@ -4360,8 +4510,308 @@ def bundle_answers(ctx):
|
|
|
check(True, "fleet re-asks after the restart (sleeping re-established)")
|
|
check(True, "fleet re-asks after the restart (sleeping re-established)")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
+# =============================================================================
|
|
|
|
|
+# Bundle: push — plan 11 task E: real Web Push over VAPID. VAPID
|
|
|
|
|
+# persistence, the service worker, subscription CRUD + validation,
|
|
|
|
|
+# notify-me prefs, the block/lifecycle dispatch matrix (global vs
|
|
|
|
|
+# opted-in), 410 pruning, retry-next-tick delivery and protocol-level
|
|
|
|
|
+# payload verification (RFC 8291 decryption + ES256 JWT) against the
|
|
|
|
|
+# fake push service.
|
|
|
|
|
+# =============================================================================
|
|
|
|
|
+
|
|
|
|
|
+def bundle_push(ctx):
|
|
|
|
|
+ auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token
|
|
|
|
|
+
|
|
|
|
|
+ def subscription_count(endpoint):
|
|
|
|
|
+ conn = db()
|
|
|
|
|
+ try:
|
|
|
|
|
+ return conn.execute(
|
|
|
|
|
+ "select count(*) from push_subscriptions where endpoint = ?",
|
|
|
|
|
+ (endpoint,)).fetchone()[0]
|
|
|
|
|
+ finally:
|
|
|
|
|
+ conn.close()
|
|
|
|
|
+
|
|
|
|
|
+ print("== task E: service worker asset ==")
|
|
|
|
|
+ status, sw = http("GET", "/sw.js", timeout=15)
|
|
|
|
|
+ check(status == 200 and "addEventListener('push'" in sw
|
|
|
|
|
+ and "notificationclick" in sw and "clients.openWindow" in sw,
|
|
|
|
|
+ "the service worker serves at the origin root with push + click routing")
|
|
|
|
|
+
|
|
|
|
|
+ print("== task E: VAPID keypair generation + persistence ==")
|
|
|
|
|
+ config = json.load(open(os.path.join(WORK, "web-config.json")))
|
|
|
|
|
+ vapid = config.get("vapid") or {}
|
|
|
|
|
+ check(bool(vapid.get("public_key")) and bool(vapid.get("private_key")),
|
|
|
|
|
+ "the VAPID keypair persists into web-config.json's vapid section")
|
|
|
|
|
+ vapid_private_key = ec.derive_private_key(
|
|
|
|
|
+ int.from_bytes(b64url_dec(vapid["private_key"]), "big"), ec.SECP256R1())
|
|
|
|
|
+ derived_public = vapid_private_key.public_key().public_bytes(
|
|
|
|
|
+ Encoding.X962, PublicFormat.UncompressedPoint)
|
|
|
|
|
+ check(b64url_dec(vapid["public_key"]) == derived_public,
|
|
|
|
|
+ "the persisted public key matches the private scalar (a valid P-256 pair)")
|
|
|
|
|
+ dash = slot_public(auth, "/", "dashboard")
|
|
|
|
|
+ check(dash.get("push_vapid_key") == vapid["public_key"],
|
|
|
|
|
+ "the dashboard exposes the same VAPID key to subscribers")
|
|
|
|
|
+ check(dash.get("push_subscribed") is False and bool(dash.get("push_subscribe")),
|
|
|
|
|
+ "the bell starts unsubscribed and carries the subscribe action")
|
|
|
|
|
+
|
|
|
|
|
+ print("== task E: subscription CRUD + auth validation ==")
|
|
|
|
|
+ browser = ec.generate_private_key(ec.SECP256R1())
|
|
|
|
|
+ browser_public = browser.public_key().public_bytes(
|
|
|
|
|
+ Encoding.X962, PublicFormat.UncompressedPoint)
|
|
|
|
|
+ browser_auth = os.urandom(16)
|
|
|
|
|
+ subscribe_fields = {"p256dh": b64url(browser_public), "auth": b64url(browser_auth)}
|
|
|
|
|
+ valid_endpoint = "http://127.0.0.1:1/x"
|
|
|
|
|
+ check("absolute http(s)" in (action_notify(act(auth, dash["push_subscribe"],
|
|
|
|
|
+ {"endpoint": "ftp://example.com/push", **subscribe_fields})) or ""),
|
|
|
|
|
+ "a non-http endpoint is rejected")
|
|
|
|
|
+ check("p256dh" in (action_notify(act(auth, dash["push_subscribe"],
|
|
|
|
|
+ {"endpoint": valid_endpoint, "p256dh": b64url(os.urandom(64)),
|
|
|
|
|
+ "auth": subscribe_fields["auth"]})) or ""),
|
|
|
|
|
+ "a malformed p256dh point is rejected")
|
|
|
|
|
+ check("auth" in (action_notify(act(auth, dash["push_subscribe"],
|
|
|
|
|
+ {"endpoint": valid_endpoint, "p256dh": subscribe_fields["p256dh"],
|
|
|
|
|
+ "auth": b64url(os.urandom(15))})) or ""),
|
|
|
|
|
+ "a malformed auth secret is rejected")
|
|
|
|
|
+
|
|
|
|
|
+ push = FakePushService()
|
|
|
|
|
+ push.serve()
|
|
|
|
|
+ endpoint_main = push.url("/main")
|
|
|
|
|
+ act(auth, dash["push_subscribe"],
|
|
|
|
|
+ {"endpoint": endpoint_main, **subscribe_fields})
|
|
|
|
|
+ conn = db()
|
|
|
|
|
+ rows = conn.execute("select user_id, p256dh, auth from push_subscriptions"
|
|
|
|
|
+ " where endpoint = ?", (endpoint_main,)).fetchall()
|
|
|
|
|
+ admin_id = conn.execute("select id from spry_users where username = 'admin'"
|
|
|
|
|
+ ).fetchone()[0]
|
|
|
|
|
+ conn.close()
|
|
|
|
|
+ check(len(rows) == 1 and rows[0][0] == admin_id,
|
|
|
|
|
+ "a valid subscription stores against the signed-in user")
|
|
|
|
|
+ # Re-subscribing the same endpoint replaces (never duplicates).
|
|
|
|
|
+ rotated_auth = os.urandom(16)
|
|
|
|
|
+ act(auth, dash["push_subscribe"],
|
|
|
|
|
+ {"endpoint": endpoint_main, "p256dh": subscribe_fields["p256dh"],
|
|
|
|
|
+ "auth": b64url(rotated_auth)})
|
|
|
|
|
+ conn = db()
|
|
|
|
|
+ rows = conn.execute("select auth from push_subscriptions where endpoint = ?",
|
|
|
|
|
+ (endpoint_main,)).fetchall()
|
|
|
|
|
+ conn.close()
|
|
|
|
|
+ check(len(rows) == 1 and rows[0][0] == b64url(rotated_auth),
|
|
|
|
|
+ "re-subscribing an endpoint replaces its keys in place")
|
|
|
|
|
+
|
|
|
|
|
+ print("== task E: notify-me pref persistence ==")
|
|
|
|
|
+ act(auth, slot_public(auth, "/tasks", "tasks")["create"],
|
|
|
|
|
+ {"project_id": "1", "title": "Push pref task", "priority": "5"})
|
|
|
|
|
+ pref_task = task_id_of("Push pref task")
|
|
|
|
|
+ act(auth, task_state(auth, pref_task)["toggle_notify"], {})
|
|
|
|
|
+ conn = db()
|
|
|
|
|
+ check(conn.execute("select count(*) from task_notify_prefs where task_id = ?"
|
|
|
|
|
+ " and user_id = ?", (pref_task, admin_id)).fetchone()[0] == 1,
|
|
|
|
|
+ "Notify me persists a task_notify_prefs row for the user")
|
|
|
|
|
+ conn.close()
|
|
|
|
|
+ check(task_state(auth, pref_task).get("notify_me") is True,
|
|
|
|
|
+ "the task slot reflects notify_me after the toggle")
|
|
|
|
|
+ act(auth, task_state(auth, pref_task)["toggle_notify"], {})
|
|
|
|
|
+ conn = db()
|
|
|
|
|
+ check(conn.execute("select count(*) from task_notify_prefs where task_id = ?",
|
|
|
|
|
+ (pref_task,)).fetchone()[0] == 0,
|
|
|
|
|
+ "Don't notify me removes the pref row")
|
|
|
|
|
+ conn.close()
|
|
|
|
|
+ check(task_state(auth, pref_task).get("notify_me") is False,
|
|
|
|
|
+ "the task slot reflects the opt-out")
|
|
|
|
|
+
|
|
|
|
|
+ print("== task E: dispatch matrix — global blocks vs opted-in lifecycle ==")
|
|
|
|
|
+ # Two "browsers" for admin (/main, /gone) and one for another user
|
|
|
|
|
+ # (id 999, rows seeded directly — push targeting is per user).
|
|
|
|
|
+ endpoint_gone = push.url("/gone")
|
|
|
|
|
+ push.status_for["/gone"] = 410
|
|
|
|
|
+ act(auth, dash["push_subscribe"],
|
|
|
|
|
+ {"endpoint": endpoint_gone, **subscribe_fields})
|
|
|
|
|
+ other_browser = ec.generate_private_key(ec.SECP256R1())
|
|
|
|
|
+ other_public = other_browser.public_key().public_bytes(
|
|
|
|
|
+ Encoding.X962, PublicFormat.UncompressedPoint)
|
|
|
|
|
+ other_auth = os.urandom(16)
|
|
|
|
|
+ endpoint_other = push.url("/other")
|
|
|
|
|
+ conn = db()
|
|
|
|
|
+ conn.execute("insert into push_subscriptions (user_id, endpoint, p256dh, auth,"
|
|
|
|
|
+ " created_at) values (999, ?, ?, ?, strftime('%s','now'))",
|
|
|
|
|
+ (endpoint_other, b64url(other_public), b64url(other_auth)))
|
|
|
|
|
+ conn.commit()
|
|
|
|
|
+ conn.close()
|
|
|
|
|
+
|
|
|
|
|
+ # Task 1: admin opted in via the UI toggle; the other user is not.
|
|
|
|
|
+ # Baselines are captured BEFORE the enqueue: the first-start push
|
|
|
|
|
+ # fires at the allocation, the instant the dispatch lands.
|
|
|
|
|
+ act(auth, slot_public(auth, "/tasks", "tasks")["create"],
|
|
|
|
|
+ {"project_id": "1", "title": "Push lifecycle task", "priority": "5"})
|
|
|
|
|
+ task1 = task_id_of("Push lifecycle task")
|
|
|
|
|
+ act(auth, task_state(auth, task1)["toggle_notify"], {})
|
|
|
|
|
+ main_before = len(push.hits("/main"))
|
|
|
|
|
+ other_before = len(push.hits("/other"))
|
|
|
|
|
+ gone_before = len(push.hits("/gone"))
|
|
|
|
|
+ enqueue_task(auth, task1)
|
|
|
|
|
+ check(wait_for(lambda: fake.running_task == task1 or False, timeout=30,
|
|
|
|
|
+ what="task 1 assignment") is not None,
|
|
|
|
|
+ "task 1 dispatches to the fake daemon")
|
|
|
|
|
+ task1_session = fake.session_id
|
|
|
|
|
+ check(wait_for(lambda: len(push.hits("/main")) > main_before, timeout=15,
|
|
|
|
|
+ what="first-start push to the opted-in admin") is not None,
|
|
|
|
|
+ "first start notifies the opted-in user's subscription")
|
|
|
|
|
+ check(wait_for(lambda: len(push.hits("/gone")) > gone_before, timeout=15,
|
|
|
|
|
+ what="first-start push to the gone endpoint") is not None,
|
|
|
|
|
+ "first start reaches every subscription of the opted-in user")
|
|
|
|
|
+ check(quiet_window(quiet_seconds(),
|
|
|
|
|
+ lambda: len(push.hits("/other")) == other_before,
|
|
|
|
|
+ "no first-start push to the not-opted-in user"),
|
|
|
|
|
+ "first start never reaches users without the notify pref")
|
|
|
|
|
+ check(wait_task_done(auth, task1, timeout=60) is not None,
|
|
|
|
|
+ "task 1 completes")
|
|
|
|
|
+ _, done_payload = decrypt_webpush(push.hits("/main")[-1], browser,
|
|
|
|
|
+ browser_public, rotated_auth)
|
|
|
|
|
+ check(done_payload["title"] == "Push lifecycle task — completed"
|
|
|
|
|
+ and done_payload["url"] == "/tasks/%d" % task1,
|
|
|
|
|
+ "completion notifies the opted-in user (decrypted payload)")
|
|
|
|
|
+ check(len(push.hits("/other")) == other_before,
|
|
|
|
|
+ "completion stays with the opted-in user only")
|
|
|
|
|
+ check(wait_for(lambda: subscription_count(endpoint_gone) == 0, timeout=15,
|
|
|
|
|
+ what="410 endpoint pruned") is not None,
|
|
|
|
|
+ "a 410 Gone endpoint is pruned from push_subscriptions")
|
|
|
|
|
+
|
|
|
|
|
+ # Block events go to EVERY push-subscribed user globally.
|
|
|
|
|
+ main_before = len(push.hits("/main"))
|
|
|
|
|
+ other_before = len(push.hits("/other"))
|
|
|
|
|
+ saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
|
|
|
|
|
+ {"type": "block", "session_id": task1_session, "data": {
|
|
|
|
|
+ "question": "Global question for everyone?", "type": "question"}})
|
|
|
|
|
+ check(wait_for(lambda: len(push.hits("/main")) > main_before
|
|
|
|
|
+ and len(push.hits("/other")) > other_before, timeout=15,
|
|
|
|
|
+ what="global block push") is not None,
|
|
|
|
|
+ "a raised block notifies every push-subscribed user globally")
|
|
|
|
|
+ block_hit = push.hits("/other")[-1]
|
|
|
|
|
+ _, payload = decrypt_webpush(block_hit, other_browser, other_public, other_auth)
|
|
|
|
|
+ check(payload["title"] == "Push lifecycle task — question raised"
|
|
|
|
|
+ and payload["body"] == "Global question for everyone?"
|
|
|
|
|
+ and payload["url"] == "/tasks/%d" % task1,
|
|
|
|
|
+ "the block payload carries task title + kind, question and click URL (%s)"
|
|
|
|
|
+ % payload)
|
|
|
|
|
+ ciphertext_len = len(block_hit["body"]) - (21 + 65)
|
|
|
|
|
+ envelope = block_hit["body"]
|
|
|
|
|
+ record_size = int.from_bytes(envelope[16:20], "big")
|
|
|
|
|
+ # rs = plaintext + delimiter (1) + tag (16) == ciphertext-with-tag len.
|
|
|
|
|
+ check(record_size == ciphertext_len,
|
|
|
|
|
+ "the RFC 8291 envelope record size covers payload + delimiter + tag")
|
|
|
|
|
+ claims = verify_vapid(block_hit, vapid["public_key"], vapid["private_key"])
|
|
|
|
|
+ check(claims.get("aud") == "http://127.0.0.1:%d" % push.port
|
|
|
|
|
+ and claims.get("sub", "").startswith("mailto:")
|
|
|
|
|
+ and claims.get("exp", 0) > time.time(),
|
|
|
|
|
+ "the VAPID JWT verifies: aud = the endpoint origin, mailto sub, future exp")
|
|
|
|
|
+ check(block_hit["headers"].get("ttl") == "60"
|
|
|
|
|
+ and block_hit["headers"].get("content-type") == "application/octet-stream",
|
|
|
|
|
+ "the push POST carries TTL and the encrypted content type")
|
|
|
|
|
+
|
|
|
|
|
+ # Task 2: only the OTHER user opted in — start + failure reach them
|
|
|
|
|
+ # alone. Baseline again before the enqueue (first start fires at
|
|
|
|
|
+ # the allocation).
|
|
|
|
|
+ fake.fail_next = True
|
|
|
|
|
+ act(auth, slot_public(auth, "/tasks", "tasks")["create"],
|
|
|
|
|
+ {"project_id": "1", "title": "Push failure task", "priority": "5"})
|
|
|
|
|
+ task2 = task_id_of("Push failure task")
|
|
|
|
|
+ conn = db()
|
|
|
|
|
+ conn.execute("insert into task_notify_prefs (task_id, user_id) values (?, 999)",
|
|
|
|
|
+ (task2,))
|
|
|
|
|
+ conn.commit()
|
|
|
|
|
+ conn.close()
|
|
|
|
|
+ main_before2 = len(push.hits("/main"))
|
|
|
|
|
+ other_before2 = len(push.hits("/other"))
|
|
|
|
|
+ enqueue_task(auth, task2)
|
|
|
|
|
+ check(wait_for(lambda: fake.running_task == task2 or False, timeout=30,
|
|
|
|
|
+ what="task 2 assignment") is not None,
|
|
|
|
|
+ "task 2 dispatches (scripted to fail)")
|
|
|
|
|
+ check(wait_for(lambda: task_state(auth, task2)["status"] == "failed"
|
|
|
|
|
+ or False, timeout=30, what="task 2 failed") is not None,
|
|
|
|
|
+ "task 2 fails")
|
|
|
|
|
+ check(wait_for(lambda: len(push.hits("/other")) >= other_before2 + 2,
|
|
|
|
|
+ timeout=15, what="opted-in user's started+failed pushes"),
|
|
|
|
|
+ "the opted-in user gets first start AND failure for the task")
|
|
|
|
|
+ started_hit = [hit for hit in push.hits("/other")[other_before2:]][0]
|
|
|
|
|
+ _, started_payload = decrypt_webpush(started_hit, other_browser,
|
|
|
|
|
+ other_public, other_auth)
|
|
|
|
|
+ check(started_payload["title"] == "Push failure task — started",
|
|
|
|
|
+ "the first-start payload names task + kind (%s)" % started_payload)
|
|
|
|
|
+ _, failure_payload = decrypt_webpush(push.hits("/other")[-1], other_browser,
|
|
|
|
|
+ other_public, other_auth)
|
|
|
|
|
+ check(failure_payload["title"] == "Push failure task — failed"
|
|
|
|
|
+ and failure_payload["url"] == "/tasks/%d" % task2,
|
|
|
|
|
+ "the failure payload names the task and links its page")
|
|
|
|
|
+ check(len(push.hits("/main")) == main_before2,
|
|
|
|
|
+ "an agent failure never reaches users without the pref")
|
|
|
|
|
+
|
|
|
|
|
+ print("== task E: retry next tick (best-effort delivery) ==")
|
|
|
|
|
+ endpoint_flaky = push.url("/flaky")
|
|
|
|
|
+ act(auth, dash["push_subscribe"],
|
|
|
|
|
+ {"endpoint": endpoint_flaky, **subscribe_fields})
|
|
|
|
|
+ push.fail_next["/flaky"] = 1
|
|
|
|
|
+ flaky_before = len(push.hits("/flaky"))
|
|
|
|
|
+ saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
|
|
|
|
|
+ {"type": "block", "session_id": task1_session, "data": {
|
|
|
|
|
+ "question": "Flaky delivery question?", "type": "question"}})
|
|
|
|
|
+ check(wait_for(lambda: len(push.hits("/flaky")) >= flaky_before + 2,
|
|
|
|
|
+ timeout=20, what="flaky endpoint retried") is not None,
|
|
|
|
|
+ "a failed send retries on the next dispatcher tick and lands")
|
|
|
|
|
+
|
|
|
|
|
+ print("== task E: waiting-on-drafts notifies like a block event ==")
|
|
|
|
|
+ fake.suspend_on_title = "Push parked task"
|
|
|
|
|
+ act(auth, slot_public(auth, "/tasks", "tasks")["create"],
|
|
|
|
|
+ {"project_id": "1", "title": "Push parked task", "priority": "5"})
|
|
|
|
|
+ task3 = task_id_of("Push parked task")
|
|
|
|
|
+ enqueue_task(auth, task3)
|
|
|
|
|
+ check(wait_for(lambda: fake.running_task == task3 or False, timeout=30,
|
|
|
|
|
+ what="task 3 assignment") is not None,
|
|
|
|
|
+ "task 3 dispatches (scripted to suspend)")
|
|
|
|
|
+ conn = db()
|
|
|
|
|
+ conn.execute("insert into tasks (project_id, title, status, priority,"
|
|
|
|
|
+ " parent_task_id, created_at)"
|
|
|
|
|
+ " values (1, 'Push draft subtask', 'draft', 5, ?,"
|
|
|
|
|
+ " strftime('%s','now'))", (task3,))
|
|
|
|
|
+ conn.commit()
|
|
|
|
|
+ conn.close()
|
|
|
|
|
+ main_before = len(push.hits("/main"))
|
|
|
|
|
+ other_before = len(push.hits("/other"))
|
|
|
|
|
+ check(wait_for(lambda: task_state(auth, task3)["status"] == "suspended"
|
|
|
|
|
+ or False, timeout=30, what="task 3 suspended") is not None,
|
|
|
|
|
+ "task 3 parks (suspended)")
|
|
|
|
|
+ check(wait_for(lambda: len(push.hits("/main")) > main_before
|
|
|
|
|
+ and len(push.hits("/other")) > other_before, timeout=15,
|
|
|
|
|
+ what="waiting-on-drafts global push") is not None,
|
|
|
|
|
+ "the waiting-on-drafts condition notifies globally like a block event")
|
|
|
|
|
+ _, waiting_payload = decrypt_webpush(push.hits("/main")[-1], browser,
|
|
|
|
|
+ browser_public, rotated_auth)
|
|
|
|
|
+ check(waiting_payload["title"]
|
|
|
|
|
+ == "Push parked task — waiting on subtasks to be enqueued"
|
|
|
|
|
+ and "still in draft" in waiting_payload["body"]
|
|
|
|
|
+ and waiting_payload["url"] == "/tasks/%d" % task3,
|
|
|
|
|
+ "the waiting-on-drafts payload carries the clear wording + link (%s)"
|
|
|
|
|
+ % waiting_payload)
|
|
|
|
|
+ # A subtask created under the ALREADY-parked parent re-arms the push.
|
|
|
|
|
+ main_before = len(push.hits("/main"))
|
|
|
|
|
+ saddle_api("POST", "/api/saddle/subtask", fake_token,
|
|
|
|
|
+ {"parent_task_id": task3, "title": "Parked-late child"})
|
|
|
|
|
+ check(wait_for(lambda: len(push.hits("/main")) > main_before, timeout=15,
|
|
|
|
|
+ what="waiting-on-drafts re-armed push") is not None,
|
|
|
|
|
+ "a draft subtask created under a parked parent re-arms the push")
|
|
|
|
|
+ check(task_state(auth, task_id_of("Parked-late child"))["status"] == "draft",
|
|
|
|
|
+ "the parked parent's child lands in draft (no auto-enqueue)")
|
|
|
|
|
+
|
|
|
|
|
+ print("== task E: unsubscribe removes the subscription ==")
|
|
|
|
|
+ dash = slot_public(auth, "/", "dashboard")
|
|
|
|
|
+ check(dash.get("push_subscribed") is True,
|
|
|
|
|
+ "the bell shows subscribed while endpoints exist")
|
|
|
|
|
+ act(auth, dash["push_unsubscribe"], {"endpoint": endpoint_main})
|
|
|
|
|
+ check(subscription_count(endpoint_main) == 0,
|
|
|
|
|
+ "unsubscribing removes the endpoint's row")
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
BUNDLES_ORDER = ("pages", "dispatch", "subtasks", "lineage",
|
|
BUNDLES_ORDER = ("pages", "dispatch", "subtasks", "lineage",
|
|
|
- "allocation", "lifecycle", "answers")
|
|
|
|
|
|
|
+ "allocation", "lifecycle", "answers", "push")
|
|
|
|
|
|
|
|
|
|
|
|
|
def run_bundle(name):
|
|
def run_bundle(name):
|