208 lines
10 KiB
Python
208 lines
10 KiB
Python
"""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()
|