#!/usr/bin/env python3 """End-to-end check of INBOX push on the VM: a real headless Google Chrome (--headless=new, a private profile) opens inbox.html from a local copy against a scratch inbox server (its own token + VAPID key, real sending), turns notifications ON through the page's bell sheet, gets a real push service subscription, and receives real pushes (SEND TEST and a coordinator post through the CLI). Also takes stills of the notification sheet. Python stdlib only (a tiny CDP client). Never touches the real inbox store. python3 _tools/inbox_push_e2e.py OUTDIR (run on the VM; stills go to OUTDIR/*.png; exit 1 on a failure) """ import base64, json, os, shutil, socket, struct, subprocess, sys, tempfile, time from urllib.request import urlopen, Request TOOLS = os.path.dirname(os.path.abspath(__file__)) FEEDBACK = os.path.dirname(TOOLS) SERVER = os.path.join(TOOLS, 'inbox_server.py') CHROME = shutil.which('google-chrome') or shutil.which('google-chrome-stable') IPHONE_UA = ('Mozilla/5.0 (iPhone; CPU iPhone OS 18_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) ' 'Version/18.1 Mobile/15E148 Safari/604.1') ROWS = [] def row(name, ok, info=''): ROWS.append((name, ok)) print('%-4s %-60s %s' % ('PASS' if ok else 'FAIL', name, info), flush=True) def free_port(): s = socket.socket() s.bind(('127.0.0.1', 0)) p = s.getsockname()[1] s.close() return p class WS: """A minimal websocket client (text frames, client masking), enough for the DevTools protocol.""" def __init__(self, url): hostport, path = url[len('ws://'):].split('/', 1) host, port = hostport.split(':') self.s = socket.create_connection((host, int(port)), timeout=60) key = base64.b64encode(os.urandom(16)).decode() self.s.sendall(('GET /%s HTTP/1.1\r\nHost: %s\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n' 'Sec-WebSocket-Key: %s\r\nSec-WebSocket-Version: 13\r\n\r\n' % (path, hostport, key)).encode()) buf = b'' while b'\r\n\r\n' not in buf: buf += self.s.recv(4096) assert b' 101 ' in buf.split(b'\r\n')[0], buf[:200] self.rest = buf.split(b'\r\n\r\n', 1)[1] self.n, self.events = 0, [] def _read(self, n): while len(self.rest) < n: c = self.s.recv(1 << 16) if not c: raise EOFError self.rest += c out, self.rest = self.rest[:n], self.rest[n:] return out def recv(self): data = b'' while True: b0, b1 = self._read(2) ln = b1 & 0x7F if ln == 126: ln = struct.unpack('!H', self._read(2))[0] elif ln == 127: ln = struct.unpack('!Q', self._read(8))[0] data += self._read(ln) if b0 & 0x80: return json.loads(data) def send(self, obj): p = json.dumps(obj).encode() h = bytes([0x81]) if len(p) < 126: h += bytes([0x80 | len(p)]) elif len(p) < 65536: h += bytes([0x80 | 126]) + struct.pack('!H', len(p)) else: h += bytes([0x80 | 127]) + struct.pack('!Q', len(p)) m = os.urandom(4) self.s.sendall(h + m + bytes(c ^ m[i % 4] for i, c in enumerate(p))) def call(self, method, params=None, session=None): self.n += 1 msg = {'id': self.n, 'method': method, 'params': params or {}} if session: msg['sessionId'] = session self.send(msg) while True: r = self.recv() if r.get('id') == self.n: if 'error' in r: raise RuntimeError('%s: %s' % (method, r['error'])) return r.get('result', {}) self.events.append(r) def main(): out = sys.argv[1] if len(sys.argv) > 1 else '.' os.makedirs(out, exist_ok=True) work = tempfile.mkdtemp(prefix='inbox-e2e-') store, www, prof = (os.path.join(work, x) for x in ('store', 'www', 'profile')) for x in (store, www, os.path.join(www, 'img')): os.makedirs(x) for f in ('inbox.html', 'inbox-sw.js', 'inbox-manifest.json'): shutil.copy(os.path.join(FEEDBACK, f), www) for f in ('inbox-icon-180.png', 'inbox-icon-192.png', 'inbox-icon-512.png'): shutil.copy(os.path.join(FEEDBACK, 'img', f), os.path.join(www, 'img')) tok = base64.urlsafe_b64encode(os.urandom(12)).decode().rstrip('=') with open(os.path.join(store, 'token'), 'w') as f: f.write(tok) env = dict(os.environ, INBOX_DIR=store, INBOX_PUSH_GAP='5') env.pop('INBOX_PUSH_FAKE', None) subprocess.run([sys.executable, SERVER, 'push-init'], env=env, check=True, capture_output=True) api, web, cdp = free_port(), free_port(), free_port() procs = [subprocess.Popen([sys.executable, SERVER, 'serve', str(api)], env=env, stderr=subprocess.DEVNULL), subprocess.Popen([sys.executable, '-m', 'http.server', str(web), '--bind', '127.0.0.1', '--directory', www], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)] chrome = subprocess.Popen([CHROME, '--headless=new', '--user-data-dir=' + prof, '--remote-debugging-port=%d' % cdp, '--no-first-run', '--no-default-browser-check', '--disable-gpu', '--hide-scrollbars', '--window-size=390,844'] + (['--no-sandbox'] if os.geteuid() == 0 else []) + ['about:blank'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) procs.append(chrome) origin = 'http://127.0.0.1:%d' % web page_url = '%s/inbox.html?api=http://127.0.0.1:%d/api/inbox' % (origin, api) def cli(cmd, body): return subprocess.run([sys.executable, SERVER, cmd], input=json.dumps(body), env=env, capture_output=True, text=True) try: for _ in range(100): try: ver = json.load(urlopen('http://127.0.0.1:%d/json/version' % cdp, timeout=2)) break except OSError: time.sleep(0.2) b = WS(ver['webSocketDebuggerUrl']) row('headless Chrome up', True, ver.get('Browser', '')) def new_page(ua=None, standalone=False): tid = b.call('Target.createTarget', {'url': 'about:blank'})['targetId'] sid = b.call('Target.attachToTarget', {'targetId': tid, 'flatten': True})['sessionId'] b.call('Page.enable', session=sid) b.call('Runtime.enable', session=sid) b.call('Emulation.setDeviceMetricsOverride', {'width': 390, 'height': 844, 'deviceScaleFactor': 2, 'mobile': True}, session=sid) if ua: b.call('Emulation.setUserAgentOverride', {'userAgent': ua, 'platform': 'iPhone'}, session=sid) src = "localStorage.setItem('ix.inbox.key', %s);" % json.dumps(tok) if standalone: src += "Object.defineProperty(navigator, 'standalone', {get: () => true});" src += "window.__msgs = []; if (navigator.serviceWorker) navigator.serviceWorker.addEventListener('message', (e) => window.__msgs.push(e.data));" b.call('Page.addScriptToEvaluateOnNewDocument', {'source': src}, session=sid) b.call('Page.navigate', {'url': page_url}, session=sid) return sid def ev(sid, expr, wait=True): r = b.call('Runtime.evaluate', {'expression': expr, 'awaitPromise': wait, 'returnByValue': True}, session=sid) if 'exceptionDetails' in r: raise RuntimeError(r['exceptionDetails'].get('exception', {}).get('description', 'js error')) return r.get('result', {}).get('value') def until(sid, expr, t=30): end = time.time() + t while time.time() < end: try: v = ev(sid, expr) if v: return v except RuntimeError: pass time.sleep(0.4) return None def shot(sid, name): time.sleep(0.6) png = base64.b64decode(b.call('Page.captureScreenshot', {'format': 'png'}, session=sid)['data']) with open(os.path.join(out, name), 'wb') as f: f.write(png) # 1. an iPhone Safari tab: the Add to Home Screen steps s1 = new_page(IPHONE_UA) until(s1, "!!document.querySelector('.livedot.on') && document.fonts.status === 'loaded'") ev(s1, "document.getElementById('bellBtn').click()") st = until(s1, "document.querySelector('.pplate') && document.querySelector('.pplate').className") row('iPhone Safari tab -> Add to Home Screen steps', st == 'pplate install' and ev(s1, "document.querySelectorAll('.psteps li').length") == 4, st or '') shot(s1, 'push-1-iphone-safari-install.png') # 2. the iPhone Home Screen app before Allow: ENABLE NOTIFICATIONS s2 = new_page(IPHONE_UA, standalone=True) until(s2, "!!document.querySelector('.livedot.on')") until(s2, "document.fonts.status === 'loaded'") ev(s2, "document.getElementById('bellBtn').click()") st = until(s2, "document.querySelector('.pplate') && document.querySelector('.pplate').className") row('Home Screen app, not allowed yet -> ENABLE NOTIFICATIONS', st == 'pplate off' and bool(ev(s2, "!!document.querySelector('[data-p=enable]')")), st or '') shot(s2, 'push-2-iphone-app-enable.png') shot_hdr = ev(s2, "(() => { const r = document.querySelector('.hdr').getBoundingClientRect(); return [r.width, r.height]; })()") # 3. allowed + ON: a real push service subscription b.call('Browser.grantPermissions', {'origin': origin, 'permissions': ['notifications']}) ev(s2, "document.querySelector('[data-p=enable]').click()") st = until(s2, "document.querySelector('.pplate.on') ? 'on' : (document.querySelector('.pnote.err') || {}).textContent", t=45) row('tap ENABLE -> subscribed (state ON)', st == 'on', str(st)) ps = json.loads(subprocess.run([sys.executable, SERVER, 'push-status'], env=env, capture_output=True, text=True).stdout) row('the server stored the device', ps.get('devices') == 1, ', '.join(x['host'] for x in ps.get('subs', []))) shot(s2, 'push-3-on.png') # 4. SEND TEST arrives (the worker shows it + tells the page) ev(s2, "document.querySelector('[data-p=test]').click()") got = until(s2, "window.__msgs.some((m) => m.type === 'inbox-push' && m.data.lane === 'test')", t=60) n = ev(s2, "navigator.serviceWorker.getRegistration('./inbox.html').then((r) => r.getNotifications()).then((l) => l.map((x) => x.title))") row('SEND TEST: a real push arrives, notification shown', bool(got) and 'INBOX · test' in (n or []), json.dumps(n)) shot(s2, 'push-4-test-arrived.png') # 5. a coordinator post (the CLI) pushes: title, body, url, the page refreshes itself ev(s2, "window.__msgs = []") r = cli('post', {'to': 'live', 'topic': 'E2E push', 'text': 'Season 1 rewards are ready for your review on the shop page.'}) got = until(s2, "window.__msgs.find((m) => m.type === 'inbox-push' && m.data.lane === 'live')", t=60) row('CLI post -> real push "LIVE · E2E push"', r.returncode == 0 and bool(got) and got['data']['title'] == 'LIVE · E2E push' and got['data']['url'] == 'inbox.html?topic=e2e-push', got and got['data']['title'] or 'none') seen = until(s2, "[...document.querySelectorAll('.t-item')].some((x) => x.textContent.includes('E2E push'))", t=15) row('the open page refreshed at once (topic listed)', bool(seen)) # 6. a status line does not push ev(s2, "window.__msgs = []") subprocess.run([sys.executable, SERVER, 'status', '--to', 'live'], input='checking the shop', env=env, capture_output=True, text=True) time.sleep(8) row('status line -> no push', ev(s2, "window.__msgs.length") == 0) # 7. a tap on a notification (the worker's inbox-open message) jumps to the topic + opens the questions cli('ask', {'to': 'kills', 'topic': 'E2E push', 'text': 'Keep the bell in the header?', 'choices': ['Yes', 'No']}) until(s2, "window.__msgs.some((m) => m.type === 'inbox-push' && m.data.lane === 'kills')", t=60) ev(s2, "document.getElementById('pushClose').click(); navigator.serviceWorker.dispatchEvent(new MessageEvent('message', " "{data: {type: 'inbox-open', url: location.origin + '/inbox.html?topic=e2e-push&qs=1'}}))") ok = until(s2, "document.getElementById('topicTitle').textContent === 'E2E push' && !document.getElementById('sheet').hidden", t=15) row('notification tap -> that topic + the questions sheet', bool(ok)) # 8. opening inbox.html?topic=ID (a tap with the app closed) starts in that topic and tidies the address b.call('Page.navigate', {'url': page_url + '&topic=e2e-push'}, session=s2) ok = until(s2, "document.getElementById('topicTitle').textContent === 'E2E push' && !location.search.includes('topic=')", t=20) row('inbox.html?topic=ID opens that topic', bool(ok)) # 9. OFF unsubscribes here and on the server ev(s2, "document.getElementById('bellBtn').click()") until(s2, "!!document.querySelector('.pplate.on')", t=20) ev(s2, "document.querySelector('[data-p=off]').click()") st = until(s2, "document.querySelector('.pplate.off') && navigator.serviceWorker.getRegistration('./inbox.html')" ".then((r) => r.pushManager.getSubscription()).then((s) => !s)", t=20) ps = json.loads(subprocess.run([sys.executable, SERVER, 'push-status'], env=env, capture_output=True, text=True).stdout) row('OFF -> unsubscribed in the browser and on the server', bool(st) and ps.get('devices') == 0) shot(s2, 'push-5-off.png') # 10. desktop Chrome (no phone UA) + blocked b.call('Browser.resetPermissions', {}) b.call('Browser.setPermission', {'origin': origin, 'permission': {'name': 'notifications'}, 'setting': 'denied'}) s3 = new_page() until(s3, "!!document.querySelector('.livedot.on')") ev(s3, "document.getElementById('bellBtn').click()") st = until(s3, "document.querySelector('.pplate') && document.querySelector('.pplate').className") row('permission denied -> BLOCKED with the unblock steps', st == 'pplate blocked', st or '') shot(s3, 'push-6-blocked.png') print('header size', shot_hdr) finally: for p in procs: p.terminate() for p in procs: try: p.wait(10) except subprocess.TimeoutExpired: p.kill() shutil.rmtree(work, ignore_errors=True) bad = [n for n, ok in ROWS if not ok] print('%d/%d passed' % (len(ROWS) - len(bad), len(ROWS))) sys.exit(1 if bad else 0) if __name__ == '__main__': main()