diff options
| -rw-r--r-- | Makefile | 4 | ||||
| -rw-r--r-- | tests/check.py | 1175 |
2 files changed, 4 insertions, 1175 deletions
@@ -16,6 +16,6 @@ $(BUILD)/aup: compiler/main.c compiler/compile.c compiler/aup.h | $(BUILD) $(CC) $(CPPFLAGS) $(CFLAGS) compiler/main.c compiler/compile.c $(LDFLAGS) $(LDLIBS) -o $@ check: $(BUILD)/aup - $(MAKE) -C $(PURE) build/pure-run build/pure-edit-module build/pure-serve - $(MAKE) -C $(PURE)/prototypes/connections BUILD=build/tree STORE=store-tree.c build/tree/pure-edit build/tree/pure-connections + $(MAKE) -C $(PURE) build/pure-run build/pure-edit-module + $(MAKE) -C $(PURE)/prototypes/connections BUILD=build/tree STORE=store-tree.c build/tree/pure-edit python3 tests/check.py --aup $(BUILD)/aup --pure-run $(PURE)/build/pure-run --pure $(PURE) diff --git a/tests/check.py b/tests/check.py index 1666ed7..98d6e37 100644 --- a/tests/check.py +++ b/tests/check.py @@ -1,5 +1,6 @@ #!/usr/bin/env python3 -"""aup v0a checks. +"""aup v0a checks. (Pure's HTTP host is checked in Pure: its tests/check.py +builds its apps with this compiler.) 1. Every example compiles, and Pure's verifier accepts the module. 2. The same source always yields the same bytes. @@ -14,12 +15,6 @@ mark exactly as a reference Python model under random typing (ASCII, Korean, emoji, combining marks, NUL, newlines), moves and deletions; string escapes and number text follow their stated rules. -9. Server: the HTTP document server running editor.aup answers random - requests exactly as the server running editor.pure, on persistent and - pipelined connections too; a read lends the stored extents and still - carries its version while the document changes, and lent answers count - against the payload budget; concurrent edits share commits, and every - acknowledged edit survives kill -9. 8. Stored state: the list app and the text editor run on a tree store, restarting at random points (every other run saving in bursts), and still match their reference models. @@ -760,1155 +755,6 @@ def stored_runs(aup, pure_run, work, seeds, steps=240): return compared, restarts -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, ROOT / 'examples/editor.aup', 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('<II', rng.randint(0, length + 2), rng.choice([0, 1, 3, length])) + insert - request = ('PUT', key, body, base) - theirs, ours = old.request(*request), new.request(*request) - assert ours == theirs, (seed, request[:2], request[3], ours[:3], theirs[:3]) - if theirs[0] == 200 and theirs[1]: - revision[key] = int(theirs[1].strip('"')) - if request[0] == 'GET': - size[key] = len(theirs[3]) - else: - size[key] = len(old.request('GET', key)[3]) - assert new.request('GET', key) == old.request('GET', key) - compared += 1 - finally: - old.stop() - new.stop() - return compared - - -def server_pipelining(aup, pure, work): - """One connection, many requests: a keep-alive client and a pipelined - burst (all requests in one write, one split across writes) get the same - answers, in order, as separate connections would.""" - import http.client - import socket - import struct - module = work / 'editor-pipeline.pmod' - compile_module(aup, ROOT / 'examples/editor.aup', module) - host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / 'pipeline']) - try: - connection = http.client.HTTPConnection('127.0.0.1', host.port, timeout=30) - answers = [] - for body, headers in [(struct.pack('<II', 0, 0) + b'hello', {'If-None-Match': '*'}), - (struct.pack('<II', 5, 0) + b'!', {'If-Match': '"1"'}), - (struct.pack('<II', 0, 0) + b'x', {'If-Match': '"1"'})]: - connection.request('PUT', '/objects/p', body, headers) - response = connection.getresponse() - answers.append((response.status, response.getheader('ETag'), response.read())) - connection.request('GET', '/objects/p') - response = connection.getresponse() - answers.append((response.status, response.getheader('ETag'), response.read())) - connection.close() - assert answers == [(200, '"1"', b'hello'), (200, '"2"', b'!'), (412, '"2"', b''), (200, '"2"', b'hello!')], answers - - def request(method, key, body=b'', condition=''): - return (f'{method} /objects/{key} HTTP/1.1\r\nHost: x\r\nContent-Length: {len(body)}\r\n{condition}\r\n' - .encode() + body) - pipeline = (request('GET', 'p') + request('PUT', 'q', struct.pack('<II', 0, 0) + b'q1', 'If-None-Match: *\r\n') + - request('GET', 'q') + request('GET', 'missing') + - request('PUT', 'q', struct.pack('<II', 2, 0) + b'!', 'If-Match: "1"\r\n') + request('GET', 'q')) - expected = [(200, b'hello!'), (200, b'q1'), (200, b'q1'), (404, b''), (200, b'!'), (200, b'q1!')] - for split in (len(pipeline), 7, 61): # one write, and small writes that cut heads and bodies - with socket.create_connection(('127.0.0.1', host.port), timeout=30) as sock: - for at in range(0, len(pipeline), split): - sock.sendall(pipeline[at:at + split]) - reader = sock.makefile('rb') - for status, body in expected: - code = int(reader.readline().split()[1]) - length = 0 - for line in iter(reader.readline, b'\r\n'): - name, _, value = line.decode().partition(':') - if name.lower() == 'content-length': - length = int(value) - assert (code, reader.read(length)) == (status, body), (split, code, status, body) - # The next round starts from the same documents. - host.request('PUT', 'q', struct.pack('<II', 0, 3) + b'q1', int(host.request('GET', 'q')[1].strip('"'))) - expected = [(200, b'hello!'), (412, b''), (200, b'q1'), (404, b''), (412, b''), (200, b'q1')] - pipeline = (request('GET', 'p') + request('PUT', 'q', struct.pack('<II', 0, 0) + b'q1', 'If-None-Match: *\r\n') + - request('GET', 'q') + request('GET', 'missing') + - request('PUT', 'q', struct.pack('<II', 2, 0) + b'!', 'If-Match: "1"\r\n') + request('GET', 'q')) - finally: - stats = host.stop() - assert stats[0]['active'] == 0 and stats[0]['retained_payload_bytes'] == 0, stats - return len(answers) + 3 * len(expected) - - -def server_lent_reads(aup, pure, work): - """A read lends the stored document's extents instead of copying them: a - 1 MiB answer sent slowly still carries the version read while the - document is replaced five times meanwhile.""" - import socket - import struct - module = work / 'editor-lent.pmod' - compile_module(aup, ROOT / 'examples/editor.aup', module) - host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / 'lent']) - big = bytes(range(256)) * 4096 - try: - assert host.request('PUT', 'big', struct.pack('<II', 0, 0) + big, 0)[0] == 200 - with socket.create_connection(('127.0.0.1', host.port), timeout=30) as sock: - sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4096) - sock.sendall(b'GET /objects/big HTTP/1.1\r\nHost: x\r\nContent-Length: 0\r\n\r\n') - data = sock.recv(100) - for revision in range(1, 6): - edit = struct.pack('<II', 0, len(big) if revision == 1 else 1) + b'Z' - assert host.request('PUT', 'big', edit, revision)[0] == 200 - while b'\r\n\r\n' not in data or len(data) < data.index(b'\r\n\r\n') + 4 + len(big): - chunk = sock.recv(65536) - assert chunk, len(data) - data += chunk - assert data[data.index(b'\r\n\r\n') + 4:] == big - assert host.request('GET', 'big') == (200, '"6"', 'none', b'Z') - finally: - stats = host.stop() - assert stats[0]['active'] == 0 and stats[0]['retained_payload_bytes'] == 0, stats - return len(big) - - -def server_lent_budget(aup, pure, work): - """Lent extents count against the payload budget while an answer waits - on its socket. Two readers pipeline 1 MiB reads without reading: the - kernel takes what it can and each reader leaves one answer waiting. A - third request finds no room (503 payload_budget) until a reader goes.""" - import socket - import struct - import time - module = work / 'editor-budget.pmod' - compile_module(aup, ROOT / 'examples/editor.aup', module) - big = bytes(range(256)) * 4096 - budget = 2 * len(big) + len(big) // 2 - host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / 'lent-budget', - '--budget', budget]) - readers = [] - try: - assert host.request('PUT', 'big', struct.pack('<II', 0, 0) + big, 0)[0] == 200 - for _ in range(2): - sock = socket.create_connection(('127.0.0.1', host.port), timeout=30) - sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4096) - sock.sendall(b'GET /objects/big HTTP/1.1\r\nHost: x\r\nContent-Length: 0\r\n\r\n' * 32) - readers.append(sock) - time.sleep(0.3) - refused = host.request('GET', 'big') - assert refused[0] == 503 and refused[2] == 'payload_budget', refused[:3] - readers.pop().close() - for _ in range(100): - answered = host.request('GET', 'big') - if answered[0] == 200: - break - time.sleep(0.05) - assert answered[0] == 200 and answered[3] == big, answered[:3] - finally: - for sock in readers: - sock.close() - stats = host.stop() - assert stats[0]['active'] == 0 and stats[0]['retained_payload_bytes'] == 0, stats - return budget - - -NOTE_SOURCE = """state note: text -input write = message { - text: text -} -on write { note = write.text } -view page = note -view size = note.bytes -""" - - -def server_lent_views(aup, pure, work, threads=1): - """A view that shows a text state as it is lends the state's stored - extents. Answers pipelined on one connection and left waiting hold the - store mapping they were read from while the note is replaced: each comes - out whole in one version. When a reader goes without reading, the - mappings its answers held are unmapped and nothing stays charged. With - several threads the owner lends and a connection thread gives back.""" - import socket - import time - module, source = work / 'note.pmod', work / 'note.aup' - source.write_text(NOTE_SOURCE) - compile_module(aup, source, module) - store = work / f'lent-views-{threads}' - host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', store, '--threads', threads]) - - def write(text): - body = json.dumps({'text': text}).encode() - import http.client - connection = http.client.HTTPConnection('127.0.0.1', host.port, timeout=30) - connection.request('POST', '/inputs/write', body, {'Content-Type': 'application/json'}) - assert connection.getresponse().status == 200 - connection.close() - - def mappings(): - # A compaction replaces the file: a mapping of the old one shows as "(deleted)". - path = str((store / 'pure.tree').resolve()) - return sum(1 for line in open(f'/proc/{host.process.pid}/maps') if path in line) - - def settle(to): - for _ in range(100): # an answer's ticket is released on its thread - if mappings() <= to: return mappings() - time.sleep(0.02) - return mappings() - - def pipeline(count): - # Far more than the kernel's buffers take (a few MiB): an answer waits, lent. - sock = socket.create_connection(('127.0.0.1', host.port), timeout=30) - sock.sendall(b'GET /views/page HTTP/1.1\r\nHost: x\r\n\r\n' * count) - time.sleep(0.3) - return sock - - size = 1 << 20 - try: - write('x' * size) - assert http(host.port, 'GET', '/views/size')[3] == str(size).encode() - alone = mappings() - # Whole versions, however the note changes while answers wait. - sock = pipeline(32) - for round_ in range(5): - write(chr(ord('A') + round_) * size) - held = mappings() - data = b'' - while data.count(b'HTTP/1.1 200') < 32 or len(data) < 32 * size: - chunk = sock.recv(1 << 20) - assert chunk, len(data) - data += chunk - sock.close() - answers = [part.split(b'\r\n\r\n', 1) for part in data.split(b'HTTP/1.1 200 OK\r\n')[1:]] - bodies = [body[:size] for _, body in answers] - assert all(b'text/plain' in head for head, _ in answers) and bodies[0] == b'x' * size, [b[:1] for b in bodies] - assert all(len(set(body)) == 1 and len(body) == size for body in bodies), [len(b) for b in bodies] - after_reading = settle(alone) - # A reader that goes without reading gives back what its answers held. - sock = pipeline(32) - for round_ in range(3): - write(chr(ord('a') + round_) * size) - held_again = mappings() - sock.close() - after_leaving = settle(alone) - assert held > 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, ROOT / 'examples/editor.aup', 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('<II', 0, 0) + b'x', base) - except (OSError, http.client.HTTPException): # the server died mid-answer: not acknowledged - return - if status != 200: - return - base = int(tag.strip('"')) - acknowledged[number] = (acknowledged[number][0] + 1, base) - threads = [threading.Thread(target=client, args=(n,)) for n in range(clients)] - start = time.perf_counter() - for thread in threads: - thread.start() - return threads, start - - timings = {} - for name, args in [('program', [pure / 'prototypes/connections/build/tree/pure-connections', '--program', pure / 'prototypes/connections/examples/editor.pure']), - ('module', [pure / 'build/pure-serve', '--module', module, '--threads', str(shards)])]: - host = HttpHost(args + ['--store', work / f'bursts-{name}-{shards}']) - acknowledged = [(0, 0)] * clients - threads, start = load(host, edits, acknowledged) - for thread in threads: - thread.join() - timings[name] = time.perf_counter() - start - assert all(count == edits for count, _ in acknowledged), acknowledged - stats = host.stop() - if name == 'module': - bursts = stats[0]['bursts'] - assert stats[0]['burst_events'] == clients * edits and (clients == 1 or bursts < clients * edits), stats - # Kill the module server while clients edit; what it acknowledged is stored. - store = work / f'bursts-module-{shards}' - host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', store, '--threads', str(shards)]) - acknowledged = [(edits, edits)] * clients - threads, _ = load(host, 1000, acknowledged) - time.sleep(0.5) - host.process.send_signal(signal.SIGKILL) - host.process.wait() - for thread in threads: - thread.join() - host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', store, '--threads', str(shards)]) - try: - for number, (count, base) in enumerate(acknowledged): - status, tag, _, body = host.request('GET', f'c{number}') - stored = int(tag.strip('"')) - assert status == 200 and stored in (base, base + 1) and len(body) == stored, (number, count, base, tag) - finally: - host.stop() - return {'writes': clients * edits, 'bursts': bursts, 'program_seconds': round(timings['program'], 3), - 'module_seconds': round(timings['module'], 3), - 'acknowledged_before_kill': sum(count for count, _ in acknowledged) - clients * edits} - - -SERVED_SOURCE = """record Entry { - name: text - points = 0 -} -state board: list of Entry by name -state at: index of board -input score = message { - points: u32 - name: bytes -} -input join = text -input drop = key Delete -on score { - if !board.has(text(score.name)) { board.add(Entry { name: text(score.name) }) } - board[text(score.name)].points = board[text(score.name)].points + score.points - at = board[text(score.name)] - reply score.name -} -on join { - board.add(Entry { name: join }) - at = board[join] -} -on drop.down { board.remove(at) } -view total = sum(e.points for e in board) -view current = at.name -view leaders = each e in board { - name = e.name - points = e.points -} -""" - - -def http(port, method, path, body=b''): - """One request on its own connection: status, X-Pure-Error, content type, body.""" - import http.client - connection = http.client.HTTPConnection('127.0.0.1', port, timeout=30) - try: - connection.request(method, path, body, {'Content-Length': str(len(body))} if method != 'GET' else {}) - response = connection.getresponse() - return response.status, response.getheader('X-Pure-Error'), response.getheader('Content-Type'), response.read() - finally: - connection.close() - - -JSON_SOURCE = """record Note { - title: text - body: text - points = 0 -} -state notes: list of Note by title -state counted = 0 -input add = message { - title: text - points: number - body: text -} -input count = message { - n: u32 -} -record Shipment { - title: text - city: text - note: text - zip = 0 -} -state shipped: list of Shipment -input ship = message { - title: text - to: { - city: text - zip: u32 - door: { - note: text - } - } -} -on add { - notes.add(Note { title: add.title, body: add.body, points: add.points }) - reply add.title -} -on count { counted = counted + count.n } -on ship { - shipped.append(Shipment { title: ship.title, city: ship.to.city, note: ship.to.door.note, zip: ship.to.zip }) - reply ship.to.door.note -} -view counted_view = counted -view all = each n in notes { - title = n.title - body = n.body - points = n.points -} -view shipments = each s in shipped { - title = s.title - city = s.city - note = s.note - zip = s.zip -} -""" - - -def server_json(aup, pure, work, rounds=300): - """Messages of text and number fields posted as JSON objects: strings - come out as Python's json module reads them (escapes, control bytes, - Korean, surrogate pairs), numbers as it reads them; a missing, unknown, - doubled or mistyped field, broken JSON and a lone surrogate answer 400.""" - module, source = work / 'json.pmod', work / 'json.aup' - source.write_text(JSON_SOURCE) - compile_module(aup, source, module) - host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / 'json-store']) - rng = random.Random(77) - pieces = ['a', 'Z', ' ', '"', '\\', '/', '\n', '\t', '\x01', '\x1f', 'é', '가', '😀', '\u2028', 'x' * 20] - notes = {} - - def post(path, body, json_body=True): - import http.client - connection = http.client.HTTPConnection('127.0.0.1', host.port, timeout=30) - try: - headers = {'Content-Length': str(len(body))} - if json_body: headers['Content-Type'] = 'application/json' - connection.request('POST', path, body, headers) - response = connection.getresponse() - return response.status, response.getheader('X-Pure-Error'), response.read() - finally: - connection.close() - - try: - for _ in range(rounds): - title = ''.join(rng.choice(pieces) for _ in range(rng.randint(0, 6))) - body = ''.join(rng.choice(pieces) for _ in range(rng.randint(0, 4))) - points = rng.choice([0, 1, -2.5, 1e3, 3.25e-2, 12345678.5]) - members = [('title', title), ('points', points), ('body', body)] - rng.shuffle(members) - text = json.dumps(dict(members), ensure_ascii=rng.random() < 0.5) - text = text.replace(': ', ':' + ' ' * rng.randint(0, 2)).replace(', ', ',\n' if rng.random() < 0.3 else ', ') - got = post('/inputs/add', text.encode()) - if title in notes: - assert got[:2] == (409, 'exists'), (text, got) - continue - assert got == (200, 'none', title.encode()), (text, got) - notes[title] = (body, points) - rows = json.loads(http(host.port, 'GET', '/views/all')[3])['items'] - want = [{'title': t, 'body': notes[t][0], 'points': notes[t][1]} for t in sorted(notes, key=lambda t: t.encode())] - assert rows == want, [(a, b) for a, b in zip(rows, want) if a != b][:2] - # The query gives fixed fields; a body without JSON fills a message's one variable field only. - assert post('/inputs/count?n=3', b'', json_body=False)[:2] == (200, 'none') - assert post('/inputs/count', b'{"n": 4}')[:2] == (200, 'none') - assert http(host.port, 'GET', '/views/counted_view')[3] == b'7' - assert post('/inputs/add?points=2', b'{"title": "q-from-query", "body": ""}')[:2] == (200, 'none') - wrong = [ - (b'{"title": "m", "points": 1}', 'missing_field'), - (b'{"title": "m", "points": 1, "body": "", "extra": 1}', 'json_field'), - (b'{"title": "m", "title": "n", "points": 1, "body": ""}', 'json_field'), - (b'{"title": 1, "points": 1, "body": ""}', 'json_type'), - (b'{"title": "m", "points": "1", "body": ""}', 'json_type'), - (b'{"title": "m", "points": 1e999, "body": ""}', 'json_type'), - (b'{"title": "m", "points": 01, "body": ""}', 'json'), - (b'{"title": "m", "points": NaN, "body": ""}', 'json'), - (b'{"title": "m", "points": 1, "body": "",}', 'json'), - (b'{"title": "\\ud800", "points": 1, "body": ""}', 'json'), - (b'{"title": "\xff", "points": 1, "body": ""}', 'json'), - (b'{"title": "m", "points": 1, "body": ""} x', 'json'), - (b'[1]', 'json'), - (b'{"title": "m\n", "points": 1, "body": ""}', 'json'), # a raw newline in a string - ] - for body, error in wrong: - assert post('/inputs/add', body)[:2] == (400, error), (body, post('/inputs/add', body)) - assert post('/inputs/add?points=2', b'{"title": "m", "points": 1, "body": ""}')[:2] == (400, 'json_field') - assert post('/inputs/add', b'plain', json_body=False)[:2] == (400, 'json') # two text fields need JSON - assert post('/inputs/count', b'{"n": -1}')[:2] == (400, 'json_type') - assert post('/inputs/count', b'{"n": 4294967296}')[:2] == (400, 'json_type') - # Groups: objects inside the body, members in any order, strings - # read where they are (escaped ones unescaped over themselves). - shipments = [] - for _ in range(rounds // 3): - title, city, note = (''.join(rng.choice(pieces) for _ in range(rng.randint(0, 5))) for _ in range(3)) - zip_code = rng.randint(0, 99999) - door = [('note', note)] - to = [('city', city), ('zip', zip_code), ('door', dict(door))] - rng.shuffle(to) - members = [('title', title), ('to', dict(to))] - rng.shuffle(members) - text = json.dumps(dict(members), ensure_ascii=rng.random() < 0.5) - got = post('/inputs/ship', text.encode()) - assert got == (200, 'none', note.encode()), (text, got) - shipments.append({'title': title, 'city': city, 'note': note, 'zip': zip_code}) - # A group's fixed field from the query, by its path. - assert post('/inputs/ship?to.zip=7', b'{"title": "q", "to": {"city": "c", "door": {"note": "n"}}}')[:2] == \ - (200, 'none') - shipments.append({'title': 'q', 'city': 'c', 'note': 'n', 'zip': 7}) - rows = json.loads(http(host.port, 'GET', f'/views/shipments?count={len(shipments)}')[3])['items'] - assert rows == shipments, [(a, b) for a, b in zip(rows, shipments) if a != b][:2] - grouped = '{"title": "t", "to": {"city": "c", "zip": 1, "door": {"note": "n"}}}' - wrong_groups = [ - (grouped.replace('"zip": 1', '"zip": 1, "lat": 2'), 'json_field'), # unknown member of a group - (grouped.replace('"note": "n"', '"note": "n", "bell": {"x": 1}'), 'json_field'), - (grouped.replace('"door": {"note": "n"}', '"door": "n"'), 'json_type'), # a group as a value - (grouped.replace('"title": "t"', '"title": {"a": "t"}'), 'json_type'), # a field as a group - (grouped.replace('"door": {"note": "n"}', '"door": {}'), 'missing_field'), - (grouped.replace('"zip": 1', '"zip": 1, "zip": 2'), 'json_field'), - (grouped.replace('"zip": 1', '"zip": [1]'), 'json_type'), - (grouped.replace('"zip": 1', '"zip": null'), 'json_type'), - (grouped.replace('"note": "n"', '"note": "n\\ud800"'), 'json'), - (grouped.replace('"door": {"note": "n"}}', '"door": {"note": "n"}'), 'json'), # unclosed - ] - for body, error in wrong_groups: - assert post('/inputs/ship', body.encode())[:2] == (400, error), (body, post('/inputs/ship', body.encode())) - assert post('/inputs/ship?to.zip=7', grouped.encode())[:2] == (400, 'json_field') # given twice - finally: - host.stop() - return rounds + len(wrong) + rounds // 3 + len(wrong_groups) - - -LINK_SENDER = """state n = 0 -state total = 0 -input buy = message { - item: text - count: u32 -} -output ordered = message { - item: text - count: u32 - to: { - city: text - } -} -on buy { - n = n + 1 - total = total + buy.count - send ordered { item: buy.item, count: buy.count, to: { city: "Seoul" } } -} -view sent = n -view sent_total = total -""" - -LINK_RECEIVER = """state stock = 0 -state orders = 0 -input order = message { - item: text - count: u32 - to: { - city: text - } -} -on order { - require order.count < 100 - stock = stock + order.count - orders = orders + 1 -} -view total = stock -view count = orders -""" - - -def server_links(aup, pure, work): - """Two served apps, one's output wired to the other's input over HTTP - (--send). While four clients buy from the sender, the receiver and then - the sender are killed (-9) and started again on their stores. In the end - the receiver has applied every order the sender saved exactly once (its - total and count equal the sender's), a refused order (count 150) was - delivered once and did not hold back the order after it, and every - acknowledged order is among the saved ones.""" - import select - import signal - import socket - import threading - import time - sender_module, receiver_module = work / 'link-sender.pmod', work / 'link-receiver.pmod' - for module, text in ((sender_module, LINK_SENDER), (receiver_module, LINK_RECEIVER)): - source = module.with_suffix('.aup') - source.write_text(text) - compile_module(aup, source, module) - - def free_port(): - with socket.socket() as probe: - probe.bind(('127.0.0.1', 0)) - return probe.getsockname()[1] - - def start(args, port): - process = subprocess.Popen([str(a) for a in args] + ['--port', str(port)], 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 post(port, path, body): - import http.client - connection = http.client.HTTPConnection('127.0.0.1', port, timeout=10) - try: - data = json.dumps(body).encode() - connection.request('POST', path, data, {'Content-Length': str(len(data)), 'Content-Type': 'application/json'}) - response = connection.getresponse() - response.read() - return response.status - finally: - connection.close() - - def number(port, view): - return int(http(port, 'GET', f'/views/{view}')[3]) - - port_a, port_b = free_port(), free_port() - receiver = [pure / 'build/pure-serve', '--module', receiver_module, '--store', work / 'link-b', '--threads', '2'] - sender = [pure / 'build/pure-serve', '--module', sender_module, '--store', work / 'link-a', - '--send', f'ordered=http://127.0.0.1:{port_b}/inputs/order'] - hosts = {'b': start(receiver, port_b), 'a': start(sender, port_a)} - acknowledged = [0] * 4 - stop = threading.Event() - - def client(which): - rng = random.Random(which + 7700) - while not stop.is_set(): - count = rng.randint(1, 20) - try: - status = post(port_a, '/inputs/buy', {'item': f'c{which}', 'count': count}) - except OSError: - time.sleep(0.02) - continue - if status == 200: - acknowledged[which] += count - else: - time.sleep(0.02) - - clients = [threading.Thread(target=client, args=(n,)) for n in range(len(acknowledged))] - for thread in clients: - thread.start() - try: - for name, args, port in (('b', receiver, port_b), ('a', sender, port_a)): - time.sleep(0.6) - hosts[name].send_signal(signal.SIGKILL) - hosts[name].wait() - time.sleep(0.3) - hosts[name] = start(args, port) - time.sleep(0.6) - finally: - stop.set() - for thread in clients: - thread.join() - try: - assert post(port_a, '/inputs/buy', {'item': 'big', 'count': 150}) == 200 - assert post(port_a, '/inputs/buy', {'item': 'after', 'count': 7}) == 200 - deadline = time.time() + 30 - while True: - sent, sent_total = number(port_a, 'sent'), number(port_a, 'sent_total') - count, total = number(port_b, 'count'), number(port_b, 'total') - if (count, total) == (sent - 1, sent_total - 150) or time.time() > 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) - - FOLDS_SOURCE = """record Task { title: text done = 0 @@ -2397,23 +1243,6 @@ def main(): report['library_programs'], report['library_diagnostics'] = libraries(aup, pure_run, work) report['editor_parity_operations'], report['editor_parity_outcomes'] = \ editor_parity(aup, args.pure.resolve(), work, range(args.editor_seeds)) - report['server_parity_requests'] = sum(server_parity(aup, args.pure.resolve(), work, range(args.editor_seeds), - threads=threads) for threads in (1, 3)) - report['server_pipelined_answers'] = server_pipelining(aup, args.pure.resolve(), work) - report['server_lent_read_bytes'] = server_lent_reads(aup, args.pure.resolve(), work) - report['server_lent_budget_bytes'] = server_lent_budget(aup, args.pure.resolve(), work) - report['server_lent_views'] = [server_lent_views(aup, args.pure.resolve(), work, threads) for threads in (1, 3)] - report['server_bursts'] = server_bursts(aup, args.pure.resolve(), work) - report['server_bursts_three_threads'] = server_bursts(aup, args.pure.resolve(), work, shards=3) - report['server_app_requests'], report['server_app_acknowledged_before_kill'] = \ - server_app(aup, args.pure.resolve(), work, range(args.editor_seeds)) - report['server_app_owner_requests'], report['server_app_owner_acknowledged_before_kill'] = \ - server_app(aup, args.pure.resolve(), work, range(args.editor_seeds), threads=3) - report['server_json_requests'] = server_json(aup, args.pure.resolve(), work) - report['server_timer_beats'] = server_timers(aup, args.pure.resolve(), work) - report['server_links'] = server_links(aup, args.pure.resolve(), work) - report['server_webhook'] = server_webhook(aup, args.pure.resolve(), work) - report['server_answers'] = server_answers(aup, args.pure.resolve(), work) print(json.dumps(report)) |