#!/usr/bin/env python3 """Pure's end-to-end checks: pure-serve, the HTTP host, running modules that aup builds (the sibling checkout, --aup-root). The core's and the store's own tests are core/test-module.c and store/test-tree-store.c. 1. Documents: pure-serve running the document editor written in aup (aup's examples/editor.aup) answers random requests exactly as the earlier document server running editor.pure, with one and with three connection threads, on persistent and pipelined connections too. 2. Lending: a read lends the stored extents and still carries its version while the document changes; lent answers count against the payload budget; a text view is lent while its answer waits and let go when it is read or its connection closes, the store keeping one mapping again. 3. Durability: concurrent edits share commits, and every acknowledged edit survives kill -9, with one and with three threads; so do the app's acknowledged posts (keyed lists, JSON bodies with groups). 4. JSON: request bodies are read in place, fields by path, and malformed or mistyped ones are refused with the field named. 5. Timers run on the owner and their beats are saved like events. 6. Links: sends are saved with their event and delivered in order and in batches to another app, which applies each once, across kill -9 of either side; webhooks receive them over HTTPS only from a trusted receiver; an API's answers become the app's events (--reply). """ import argparse import json import random import subprocess import sys import tempfile from pathlib import Path HERE = Path(__file__).resolve().parent ROOT = HERE.parent # The app that exercises documents: the editor written in aup, from aup's examples. EDITOR = ROOT.parent / 'aup' / 'examples' / 'editor.aup' def run(*args, stdin=None, check=True): return subprocess.run([str(a) for a in args], input=stdin, capture_output=True, text=True, check=check, timeout=60) def compile_module(aup, source_path, out): run(aup, 'build', source_path, '-o', out) return out.read_bytes() class HttpHost: """A Pure HTTP host on a fresh port; stops with its statistics.""" def __init__(self, args): import select self.process = subprocess.Popen([str(a) for a in args] + ['--port', '0'], stdout=subprocess.PIPE, stderr=subprocess.PIPE) ready, _, _ = select.select([self.process.stdout], [], [], 10) assert ready, 'HTTP host did not start' line = self.process.stdout.readline() assert line, self.process.communicate(timeout=5) self.port = json.loads(line)['port'] def request(self, method, key, body=b'', expected=None): import http.client headers = {'Content-Length': str(len(body))} if expected is not None: headers['If-None-Match' if expected == 0 else 'If-Match'] = '*' if expected == 0 else f'"{expected}"' connection = http.client.HTTPConnection('127.0.0.1', self.port, timeout=30) try: connection.request(method, f'/objects/{key}', body, headers) response = connection.getresponse() return response.status, response.getheader('ETag'), response.getheader('X-Pure-Error'), response.read() finally: connection.close() def stop(self): self.process.terminate() _, error = self.process.communicate(timeout=30) assert self.process.returncode == 0, error[-1000:] return [json.loads(line) for line in error.decode().splitlines() if line.startswith('{')] def server_parity(aup, pure, work, seeds, operations=150, threads=1): """The document server running editor.aup answers every request exactly as the server running editor.pure: status, revision, error and body.""" import struct module = work / 'editor-server.pmod' compile_module(aup, EDITOR, module) compared = 0 for seed in seeds: rng = random.Random(seed + 13000) old = HttpHost([pure / 'prototypes/connections/build/tree/pure-connections', '--program', pure / 'prototypes/connections/examples/editor.pure', '--store', work / f'http-old-{seed}-{threads}']) new = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / f'http-new-{seed}-{threads}', '--threads', str(threads)]) try: revision, size = {}, {} for _ in range(operations): key = rng.choice(['a', 'b', 'long-key_1']) current, length = revision.get(key, 0), size.get(key, 0) if rng.random() < 0.35: request = ('GET', key, b'', None) else: pick = rng.random() base = current if pick < 0.75 else max(current - 1, 0) if pick < 0.9 else current + 3 if rng.random() < 0.05: body = bytes(rng.randrange(256) for _ in range(rng.randint(0, 7))) # too short else: insert = bytes(rng.randrange(256) for _ in range(rng.choice([0, 1, 7, 300, 5000]))) body = struct.pack(' alone and held_again > alone and after_reading <= alone and after_leaving <= alone, \ (alone, held, after_reading, held_again, after_leaving) assert http(host.port, 'GET', '/views/page')[3] == b'c' * size finally: stats = host.stop() assert stats[0]['active'] == 0 and stats[0]['retained_payload_bytes'] == 0, stats return {'alone': alone, 'while_lent': max(held, held_again), 'after': after_leaving} def server_bursts(aup, pure, work, clients=8, edits=30, shards=1): """Concurrent edits share commits, and every acknowledged edit survives a kill -9 of the server.""" import http.client import signal import struct import threading import time module = work / 'editor-bursts.pmod' compile_module(aup, EDITOR, module) def load(host, rounds, acknowledged): def client(number): key, base = f'c{number}', acknowledged[number][1] for _ in range(rounds): try: status, tag, _, _ = host.request('PUT', key, struct.pack(' deadline: break time.sleep(0.1) assert (count, total) == (sent - 1, sent_total - 150), (count, total, sent, sent_total) assert sent_total >= sum(acknowledged) + 157 and sum(acknowledged) > 0, (sent_total, acknowledged) finally: stats = {} for name, process in hosts.items(): process.terminate() _, error = process.communicate(timeout=30) assert process.returncode == 0, error[-1000:] stats[name] = [json.loads(line) for line in error.decode().splitlines() if line.startswith('{"active"')][-1] return {'orders': sent, 'acknowledged_total': sum(acknowledged), 'delivered_after_restart': stats['a']['delivered'], 'attempts_after_restart': stats['a']['delivery_attempts']} def server_webhook(aup, pure, work): """An output wired to a third-party HTTPS endpoint (a Python server with a certificate of its own): five orders arrive in order as JSON with the sender's headers, one answered 503 at first arrives again, and none goes to a receiver whose certificate the sender does not trust.""" from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import ssl import threading import time module = work / 'hook-sender.pmod' module.with_suffix('.aup').write_text(LINK_SENDER) compile_module(aup, module.with_suffix('.aup'), module) for name in ('trusted', 'other'): subprocess.run(['openssl', 'req', '-x509', '-newkey', 'ec', '-pkeyopt', 'ec_paramgen_curve:P-256', '-nodes', '-keyout', work / f'{name}.key', '-out', work / f'{name}.pem', '-days', '1', '-subj', '/CN=127.0.0.1', '-addext', 'subjectAltName=IP:127.0.0.1'], check=True, capture_output=True) received, refused_once = [], set() class Hook(BaseHTTPRequestHandler): protocol_version = 'HTTP/1.1' def do_POST(self): body = self.rfile.read(int(self.headers['Content-Length'])) received.append((dict(self.headers), json.loads(body))) status = 200 if self.headers['X-Pure-Sequence'] == '2' and '2' not in refused_once: refused_once.add('2') status = 503 self.send_response(status) self.send_header('Content-Length', '0') self.end_headers() def log_message(self, *args): pass server = ThreadingHTTPServer(('127.0.0.1', 0), Hook) context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) context.load_cert_chain(work / 'trusted.pem', work / 'trusted.key') server.socket = context.wrap_socket(server.socket, server_side=True) threading.Thread(target=server.serve_forever, daemon=True).start() url = f'ordered=https://127.0.0.1:{server.server_address[1]}/hook' try: def buy(host, count): import http.client connection = http.client.HTTPConnection('127.0.0.1', host.port, timeout=10) data = json.dumps({'item': f'i{count}', 'count': count}).encode() connection.request('POST', '/inputs/buy', data, {'Content-Length': str(len(data)), 'Content-Type': 'application/json'}) assert connection.getresponse().status == 200 connection.close() host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / 'hook-a', '--send', url, '--send-ca', work / 'trusted.pem']) try: for count in range(1, 6): buy(host, count) deadline = time.time() + 20 while len({h['X-Pure-Sequence'] for h, _ in received}) < 5 and time.time() < deadline: time.sleep(0.05) finally: stats = host.stop()[-1] firsts = [] for headers, body in received: seq = int(headers['X-Pure-Sequence']) assert headers['X-Pure-Source'] == 'hook-sender.ordered' and headers['Content-Type'] == 'application/json' assert headers['Idempotency-Key'] == f'hook-sender.ordered-{seq}' and 'X-Pure-Count' not in headers assert body == {'item': f'i{seq}', 'count': seq, 'to': {'city': 'Seoul'}}, body if seq not in firsts: firsts.append(seq) assert firsts == [1, 2, 3, 4, 5] and len(received) == 6 and stats['delivered'] == 5, (firsts, len(received), stats) # A receiver the sender cannot trust gets nothing; the send waits in the outbox. before = len(received) host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / 'hook-b', '--send', url, '--send-ca', work / 'other.pem']) try: buy(host, 1) time.sleep(1) finally: host.process.terminate() _, error = host.process.communicate(timeout=30) assert len(received) == before and b'not trusted' in error, error[-500:] finally: server.shutdown() return {'hooks': len(received), 'delivered': stats['delivered']} PAY_SOURCE = """state sent = 0 state paid = 0 state declined = 0 state paid_total = 0 input buy = message { order: text amount: u32 } output charge = message { order: text amount: u32 } input charged = message { order: text amount: u32 status: text http_status: u32 } on buy { sent = sent + 1 send charge { order: buy.order, amount: buy.amount } } on charged { if charged.http_status == 200 { paid = paid + 1 paid_total = paid_total + charged.amount } else { declined = declined + 1 } } view sent_view = sent view paid_view = paid view declined_view = declined view paid_total_view = paid_total """ def server_answers(aup, pure, work): """An output wired to a payments API whose answers come back as events (--reply): the API answers 200 or 402 and gives a key's first answer to every request with that Idempotency-Key. While clients buy, the sender is killed (-9) and started again; in the end every saved order was answered exactly once: the app's paid and declined counts and paid total match the API's answers, one per key.""" from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import select import signal import socket import threading import time module = work / 'pay.pmod' module.with_suffix('.aup').write_text(PAY_SOURCE) compile_module(aup, module.with_suffix('.aup'), module) answers, lock = {}, threading.Lock() class Payments(BaseHTTPRequestHandler): protocol_version = 'HTTP/1.1' def do_POST(self): body = json.loads(self.rfile.read(int(self.headers['Content-Length']))) with lock: key = self.headers['Idempotency-Key'] if key not in answers: ok = body['amount'] <= 1000 answers[key] = (body, 200 if ok else 402, {'status': 'succeeded' if ok else 'declined', 'id': f'ch_{len(answers)}', 'x': [1, {}]}) _, status, answer = answers[key] data = json.dumps(answer).encode() self.send_response(status) self.send_header('Content-Type', 'application/json') self.send_header('Content-Length', str(len(data))) self.end_headers() self.wfile.write(data) def log_message(self, *args): pass server = ThreadingHTTPServer(('127.0.0.1', 0), Payments) server.handle_error = lambda request, address: None # a sender killed mid-request threading.Thread(target=server.serve_forever, daemon=True).start() with socket.socket() as probe: probe.bind(('127.0.0.1', 0)) port = probe.getsockname()[1] args = [pure / 'build/pure-serve', '--module', module, '--store', work / 'pay-store', '--port', port, '--send', f'charge=http://127.0.0.1:{server.server_address[1]}/charge', '--reply', 'charge=charged'] def start(): process = subprocess.Popen([str(a) for a in args], stdout=subprocess.PIPE, stderr=subprocess.PIPE) ready, _, _ = select.select([process.stdout], [], [], 10) assert ready and process.stdout.readline(), 'host did not start' return process def number(view): return int(http(port, 'GET', f'/views/{view}')[3]) host = start() stop = threading.Event() def client(which): import http.client rng = random.Random(which + 8800) while not stop.is_set(): data = json.dumps({'order': f'o{which}-{rng.randrange(10**6)}', 'amount': rng.randint(1, 1500)}).encode() try: connection = http.client.HTTPConnection('127.0.0.1', port, timeout=10) connection.request('POST', '/inputs/buy', data, {'Content-Length': str(len(data)), 'Content-Type': 'application/json'}) connection.getresponse().read() connection.close() except OSError: time.sleep(0.02) clients = [threading.Thread(target=client, args=(n,)) for n in range(3)] for thread in clients: thread.start() try: time.sleep(0.8) host.send_signal(signal.SIGKILL) host.wait() host = start() time.sleep(0.8) finally: stop.set() for thread in clients: thread.join() try: deadline = time.time() + 30 while number('paid_view') + number('declined_view') < number('sent_view') and time.time() < deadline: time.sleep(0.1) sent, paid, declined, total = (number(v) for v in ('sent_view', 'paid_view', 'declined_view', 'paid_total_view')) with lock: keys = list(answers.values()) assert paid + declined == sent == len(keys), (paid, declined, sent, len(keys)) assert paid == sum(1 for _, status, _ in keys if status == 200), (paid, len(keys)) assert total == sum(body['amount'] for body, status, _ in keys if status == 200) finally: host.terminate() _, error = host.communicate(timeout=30) server.shutdown() stats = [json.loads(line) for line in error.decode().splitlines() if line.startswith('{"active"')][-1] return {'orders': sent, 'paid': paid, 'declined': declined, 'answers_after_restart': stats['answers']} BEAT_SOURCE = """state beats = 0 state last = 0 state stamped = 0 input beat = every 50 ms input stamp = message { note: text } on beat { beats = beats + 1 last = now } on stamp { stamped = now } view beats_view = beats view last_view = last view stamped_view = stamped """ def server_timers(aup, pure, work): """A served app's timers tick on the server's clock and its events see the time they were applied, on one thread and with an owner; the ticks are saved, so a restarted server goes on from them.""" import time module, source = work / 'beat.pmod', work / 'beat.aup' source.write_text(BEAT_SOURCE) compile_module(aup, source, module) seen = {} for threads in (1, 3): store = work / f'beat-{threads}' beats = 0 for run_number in range(2): host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', store, '--threads', threads]) try: posted = time.time() * 1000 assert http(host.port, 'POST', '/inputs/stamp', b'x')[0] == 200 time.sleep(0.6) before = time.time() * 1000 read = lambda view: float(http(host.port, 'GET', f'/views/{view}')[3]) count, last, stamped = read('beats_view'), read('last_view'), read('stamped_view') after = time.time() * 1000 + 1 # the server's clock reads whole milliseconds assert count >= beats + 5, (threads, run_number, count, beats) assert before - 1000 < last <= after and posted - 50 <= stamped <= before, (last, stamped, posted, before) beats = count finally: stats = host.stop() assert stats[0]['ticks'] >= 5, stats seen[threads] = beats return seen def server_app(aup, pure, work, seeds, operations=200, threads=1): """An app served by its views and inputs, no routes written: random POSTs (a message with query fields and a body, a text, a key) and GETs (a number, a text, a table's pages) answer exactly as a Python model says, failures with their fault; what was acknowledged is there after the server stops, and after it is killed while clients post. The app's owner thread applies the inputs; connection threads (one or several) show views from the snapshot of its last save.""" import signal import threading import time module = work / 'served.pmod' source = work / 'served.aup' source.write_text(SERVED_SOURCE) compile_module(aup, source, module) names = ['ann', 'bob', 'zed', '가', 'a"b', 'back\\slash', 'tab\t', '', 'bob2'] compared = 0 for seed in seeds: rng, board, at = random.Random(seed + 9100), {}, 0 store = work / f'served-{seed}-{threads}' serve = [pure / 'build/pure-serve', '--module', module, '--threads', threads] host = HttpHost(serve + ['--store', store]) order = lambda: sorted(board, key=lambda name: name.encode()) def place(name): nonlocal at keys = order() if name not in board: position = sum(1 for k in keys if k.encode() < name.encode()) board[name] = 0 if len(board) > 1 and at >= position: at += 1 try: for _ in range(operations): roll, keys = rng.random(), order() if roll < 0.35: name, points = rng.choice(names), rng.randint(0, 9) if rng.random() < 0.05: request = ('POST', '/inputs/score', b'x') # no points want = (400, 'missing_field', b'') elif rng.random() < 0.05: request = ('POST', f'/inputs/score?points={points}', b'\xff') want = (422, 'encoding', b'') else: request = ('POST', f'/inputs/score?points={points}', name.encode()) place(name) board[name] += points at = order().index(name) want = (200, 'none', name.encode()) elif roll < 0.5: name = rng.choice(names) request = ('POST', '/inputs/join', name.encode()) if name in board: want = (409, 'exists', b'') else: place(name) at = order().index(name) want = (200, 'none', b'') elif roll < 0.58: request = ('POST', '/inputs/drop', b'') if not board: want = (404, 'not_found', b'') else: del board[keys[at]] at = min(at, max(len(board) - 1, 0)) want = (200, 'none', b'') elif roll < 0.68: request = ('GET', '/views/total', b'') want = (200, 'none', str(sum(board.values())).encode()) elif roll < 0.78: request = ('GET', '/views/current', b'') want = (200, 'none', keys[at].encode()) if board else (404, 'no_value', b'') elif roll < 0.96: first, count = rng.randint(0, 4), rng.choice([1, 2, 5, 100]) request = ('GET', f'/views/leaders?from={first}&count={count}', b'') shown = min(first, len(keys)) rows = [{'name': k, 'points': board[k]} for k in keys[shown:shown + count]] want = (200, 'none', {'rows': len(keys), 'from': shown, 'items': rows}) else: request, want = rng.choice([(('GET', '/views/nope', b''), (404, 'no_view', b'')), (('POST', '/inputs/nope', b''), (404, 'no_input', b'')), (('POST', '/inputs/join?x=1', b'q'), (400, 'query', b'')), (('GET', '/views/total?from=1', b''), (400, 'query', b''))]) status, error, kind, body = http(host.port, *request) if isinstance(want[2], dict): assert kind == 'application/json', kind body = json.loads(body) assert (status, error, body) == want, (seed, request, (status, error, body), want) compared += 1 finally: stats = host.stop() assert stats[0]['bursts'] <= stats[0]['requests'] and stats[0]['owner'], stats assert stats[0]['views_from_snapshots'] > 0, stats # A graceful stop kept every acknowledged event. host = HttpHost(serve + ['--store', store]) try: got = json.loads(http(host.port, 'GET', '/views/leaders')[3]) assert got['items'] == [{'name': k, 'points': board[k]} for k in order()], (seed, got) finally: host.stop() # Clients post while the server is killed: each acknowledged point is # stored. Readers on other connections meanwhile see every point # acknowledged before they asked, never a total going back, and never one # that was not saved: the total after the kill is at least any they saw. store = work / f'served-kill-{threads}' serve = [pure / 'build/pure-serve', '--module', module, '--threads', threads] host = HttpHost(serve + ['--store', store]) acknowledged = [0] * 6 seen = [0, 0] def client(number): for _ in range(10000): try: status, _, _, _ = http(host.port, 'POST', '/inputs/score?points=1', f'c{number}'.encode()) except OSError: return if status != 200: return acknowledged[number] += 1 def reader(number): for _ in range(100000): floor = sum(acknowledged) try: status, _, _, body = http(host.port, 'GET', '/views/total') except OSError: return if status != 200: return total = int(body) assert floor <= total and seen[number] <= total, (floor, seen[number], total) seen[number] = total clients = [threading.Thread(target=client, args=(n,)) for n in range(len(acknowledged))] clients += [threading.Thread(target=reader, args=(n,)) for n in range(len(seen))] for thread in clients: thread.start() time.sleep(0.8) host.process.send_signal(signal.SIGKILL) host.process.wait() for thread in clients: thread.join() host = HttpHost(serve + ['--store', store]) try: stored = {row['name']: row['points'] for row in json.loads(http(host.port, 'GET', '/views/leaders')[3])['items']} for number, count in enumerate(acknowledged): assert count > 0 and stored.get(f'c{number}', 0) in (count, count + 1), (number, count, stored) assert min(seen) > 0 and max(seen) <= sum(stored.values()), (seen, stored) finally: host.stop() return compared, sum(acknowledged) def main(): global EDITOR parser = argparse.ArgumentParser() parser.add_argument('--aup', type=Path, required=True, help='the aup compiler') parser.add_argument('--aup-root', type=Path, default=ROOT.parent / 'aup', help='its checkout, for examples/editor.aup') parser.add_argument('--seeds', type=int, default=4) args = parser.parse_args() aup, pure = args.aup.resolve(), ROOT EDITOR = args.aup_root.resolve() / 'examples' / 'editor.aup' report = {} with tempfile.TemporaryDirectory(prefix='pure-check-') as directory: work = Path(directory) report['server_parity_requests'] = sum(server_parity(aup, pure, work, range(args.seeds), threads=threads) for threads in (1, 3)) report['server_pipelined_answers'] = server_pipelining(aup, pure, work) report['server_lent_read_bytes'] = server_lent_reads(aup, pure, work) report['server_lent_budget_bytes'] = server_lent_budget(aup, pure, work) report['server_lent_views'] = [server_lent_views(aup, pure, work, threads) for threads in (1, 3)] report['server_bursts'] = server_bursts(aup, pure, work) report['server_bursts_three_threads'] = server_bursts(aup, pure, work, shards=3) report['server_app_requests'], report['server_app_acknowledged_before_kill'] = \ server_app(aup, pure, work, range(args.seeds)) report['server_app_owner_requests'], report['server_app_owner_acknowledged_before_kill'] = \ server_app(aup, pure, work, range(args.seeds), threads=3) report['server_json_requests'] = server_json(aup, pure, work) report['server_timer_beats'] = server_timers(aup, pure, work) report['server_links'] = server_links(aup, pure, work) report['server_webhook'] = server_webhook(aup, pure, work) report['server_answers'] = server_answers(aup, pure, work) print(json.dumps(report)) if __name__ == '__main__': sys.exit(main())