"""Bounded real-HTTP checks of existing source7 authoring; no compiler or GUI.""" import copy import hashlib import json import os from pathlib import Path import queue import re import subprocess import tarfile import threading import time import urllib.error import urllib.request ROOT = Path(__file__).resolve().parents[3] OUT = Path(__file__).resolve().parent EXE = ROOT / '.local/target-gnullvm/debug/arcadia-server.exe' IDENTITY = re.compile(r' id "[0-9a-f-]{36}"') OA = '''arcadia 7; project distress "Distress response" { oa_actor vessel "Vessel in distress" {} oa_entity rescue "Rescue coordination" {} oa_activity report "Report distress" {} oa_activity assess "Assess incident" {} allocation vessel_reports "Crew reports" { source @vessel; target @report; } allocation rescue_assesses "Team assesses" { source @rescue; target @assess; } interaction distress_report "Distress report" { source @report; target @assess; } } ''' SCENARIO = '''arcadia 7; project response "Distress response" { oa_scenario assessment "Assess the distress report" { viewpoint OperationalActivity; } } ''' def save(name, value): path = OUT / name path.write_text(value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, indent=2) + '\n', encoding='utf-8') def digest(data): return hashlib.sha256(data).hexdigest() def provenance(): archive = ROOT / '.local/linux-qa/runtime/source7-20260907T082753190288Z.tar.gz' comparisons = [] with tarfile.open(archive) as source: for member in source.getmembers(): path = member.name.removeprefix('./') if member.isfile() and (path.startswith('crates/oa-core/') or path.startswith('crates/oa-server/')): current = ROOT / path old = source.extractfile(member).read() comparisons.append({'path': path, 'snapshot_sha256': digest(old), 'current_sha256': digest(current.read_bytes()) if current.is_file() else None}) assert comparisons, 'Snapshot paths were not recognized; do not infer equality' return {'executable': str(EXE), 'exe_sha256': digest(EXE.read_bytes()), 'exe_bytes': EXE.stat().st_size, 'snapshot': str(archive), 'snapshot_sha256': digest(archive.read_bytes()), 'source_comparisons': comparisons, 'help': subprocess.check_output([str(EXE), '--help'], text=True), 'version': subprocess.check_output([str(EXE), '--version'], text=True).strip()} def start_server(store): env = os.environ.copy() env.pop('ARCADIA_SERVER_TOKEN', None) proc = subprocess.Popen([str(EXE), '--bind', '127.0.0.1:0', '--data-dir', str(store)], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, env=env, creationflags=subprocess.CREATE_NO_WINDOW) lines = queue.Queue() threading.Thread(target=lambda: [lines.put(line) for line in proc.stderr], daemon=True).start() line = lines.get(timeout=15) match = re.search(r'listening on (127\.0\.0\.1:\d+)', line) if not match: proc.terminate() proc.wait(timeout=10) raise AssertionError(line) return proc, 'http://' + match[1], line.strip() requests = [] base = '' def request(method, path, payload=None, etag=None, expected=200): headers = {'Content-Type': 'application/json'} if etag: headers['If-Match'] = etag raw = None if payload is None else json.dumps(payload, ensure_ascii=False).encode('utf-8') started = time.perf_counter() try: response = urllib.request.urlopen(urllib.request.Request(base + path, data=raw, headers=headers, method=method), timeout=15) except urllib.error.HTTPError as error: response = error with response: data = response.read().decode('utf-8') body = json.loads(data) if data else None result = {'method': method, 'path': path, 'status': response.status, 'etag': response.headers.get('ETag'), 'location': response.headers.get('Location'), 'seconds': time.perf_counter() - started, 'body': body} requests.append(result) save('http-results.json', requests) assert result['status'] == expected, result return result def assert_semantics(model, kind): assert model['schema_version'] == 7 for key in ['diagrams', 'artifact_links', 'traces']: assert model[key] == {}, (key, model[key]) elements = {v['alias']: v for v in model['elements'].values()} relations = {v['alias']: v for v in model['relations'].values()} assert all(v['parent'] is None and v['description'] == '' for v in elements.values()) if kind == 'oa': expected = {'vessel': ('Actor', 'Vessel in distress'), 'rescue': ('Entity', 'Rescue coordination'), 'report': ('Activity', 'Report distress'), 'assess': ('Activity', 'Assess incident')} assert set(elements) == set(expected) for alias, (body, name) in expected.items(): assert elements[alias]['body'] == {'Oa': body} assert elements[alias]['name'] == name links = {'vessel_reports': ('Allocation', 'vessel', 'report'), 'rescue_assesses': ('Allocation', 'rescue', 'assess'), 'distress_report': ('Interaction', 'report', 'assess')} assert set(relations) == set(links) for alias, (body, source, target) in links.items(): fields = {'source': elements[source]['id'], 'target': elements[target]['id']} if body == 'Interaction': fields['items'] = [] assert relations[alias]['body'] == {body: fields} else: assert set(elements) == {'assessment'} and not relations assert elements['assessment']['body'] == {'Oa': {'Shared': {'Scenario': { 'viewpoint': 'OperationalActivity', 'context': None, 'precondition': '', 'postcondition': '', 'steps': []}}}} return elements def exercise(kind, source, name, replacement, alias): save(kind + '-input.arcadia', source) bundle = {'entry': 'project.arcadia', 'files': {'project.arcadia': source}} created = request('POST', '/api/v1/projects/import-source-project', bundle, expected=201) path = created['location'] assert path and path.startswith('/api/v1/projects/') accepted = request('GET', path + '/source-project') stable = accepted['body'] text = stable['files']['project.arcadia'] assert stable['entry'] == 'project.arcadia' and set(stable['files']) == {'project.arcadia'} assert IDENTITY.sub('', text) == source, 'Normalization changed more than UUID insertion' ids = IDENTITY.findall(text) assert len(ids) == (8 if kind == 'oa' else 2) and len(set(ids)) == len(ids) save(kind + '-stabilized.arcadia', text) initial_model = request('GET', path + '/model')['body'] elements = assert_semantics(initial_model, kind) save(kind + '-initial-model.json', initial_model) unchanged = request('PUT', path + '/source-project', stable, accepted['etag']) assert unchanged['body']['changed'] is False and unchanged['etag'] == accepted['etag'] assert unchanged['body']['source_project'] == stable changed = copy.deepcopy(stable) assert text.count('"' + name + '"') == 1 changed['files']['project.arcadia'] = text.replace('"' + name + '"', '"' + replacement + '"') updated = request('PUT', path + '/source-project', changed, accepted['etag']) assert updated['body']['changed'] is True and updated['etag'] != accepted['etag'] assert updated['body']['source_project'] == changed after = request('GET', path + '/model')['body'] expected_model = copy.deepcopy(initial_model) expected_model['elements'][elements[alias]['id']]['name'] = replacement assert after == expected_model, 'Name-only edit changed other semantic facts' assert IDENTITY.findall(changed['files']['project.arcadia']) == ids save(kind + '-renamed.arcadia', changed['files']['project.arcadia']) save(kind + '-renamed-model.json', after) stale = request('PUT', path + '/source-project', stable, accepted['etag'], expected=412) assert request('GET', path + '/source-project')['body'] == changed invalid = copy.deepcopy(changed) invalid['files']['project.arcadia'] += '\nunknown_declaration broken "Broken" {}\n' rejected = request('PUT', path + '/source-project', invalid, updated['etag'], expected=422) assert request('GET', path + '/source-project')['body'] == changed assert request('GET', path + '/model')['body'] == after return {'path': path, 'source': changed, 'model': after, 'ids': len(ids), 'initial_issues': created['body'], 'stale_status': stale['status'], 'invalid_status': rejected['status']} def main(): global base began = time.perf_counter() store = OUT / ('store-' + time.strftime('%Y%m%dT%H%M%SZ', time.gmtime())) assert not store.exists(), 'Never reuse an earlier experiment store' save('provenance.json', provenance()) proc = None try: proc, base, startup = start_server(store) save('server-start.txt', startup + '\n') results = [exercise('oa', OA, 'Assess incident', 'Assess incident carefully', 'assess'), exercise('scenario', SCENARIO, 'Assess the distress report', 'Assess the reported distress', 'assessment')] proc.terminate() proc.wait(timeout=10) proc, base, startup = start_server(store) save('server-restart.txt', startup + '\n') for item in results: assert request('GET', item['path'] + '/source-project')['body'] == item['source'] assert request('GET', item['path'] + '/model')['body'] == item['model'] save('summary.json', {'status': 'PASS', 'elapsed_seconds': time.perf_counter() - began, 'request_count': len(requests), 'cases': results, 'store': str(store), 'limitations': ['Existing preserved binary; no current rebuild', 'No GUI/Undo/performance acceptance', 'Process termination after synchronous requests; restart verifies persisted data']}) print(json.dumps({'status': 'PASS', 'requests': len(requests), 'seconds': time.perf_counter() - began})) except Exception as error: save('failure.json', {'type': type(error).__name__, 'message': str(error)}) raise finally: if proc is not None and proc.poll() is None: proc.terminate() proc.wait(timeout=10) if __name__ == '__main__': main()