From c08c570d465ffa22c772c2c4d0910eb8a17a379e Mon Sep 17 00:00:00 2001 From: eunhhu Date: Sun, 27 Sep 2026 13:48:55 +0100 Subject: Check pure-serve in Pure, not in aup The HTTP host's end-to-end checks (documents against the earlier server, lending, kill -9 durability, JSON, timers, links, webhooks and answers) sat in aup's tests, so a change to the host was verified from the toolchain. They now live in tests/check.py; make check builds their apps with the sibling aup and runs them after the unit tests, which stay beside their parts. Design 009 states the rule for all three repositories. --- tests/check.py | 1231 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 1231 insertions(+) create mode 100644 tests/check.py (limited to 'tests/check.py') diff --git a/tests/check.py b/tests/check.py new file mode 100644 index 0000000..8352578 --- /dev/null +++ b/tests/check.py @@ -0,0 +1,1231 @@ +#!/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()) -- cgit v1.2.3