58 lines
3.0 KiB
Python
58 lines
3.0 KiB
Python
import datetime
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import uuid
|
|
|
|
repo = Path(__file__).resolve().parents[2]
|
|
trial = repo / '.local/native-qa/contextual-source-20260911-v2/lifeline'
|
|
base = repo / 'harness/fixtures/staged-source-authoring/base'
|
|
mode = sys.argv[1]
|
|
assert mode in ('before-stop', 'after-restore')
|
|
expected = {p.name: p.read_text(encoding='utf-8') for p in base.glob('*.arcadia')}
|
|
assert len(expected) == 2
|
|
matches = []
|
|
for p in (trial / 'profile/local/ArcadiaStudio/recovery').glob('*.json'):
|
|
raw = p.read_bytes()
|
|
j = json.loads(raw)
|
|
draft = (j.get('behavior') or {}).get('draft') or {}
|
|
source = (draft.get('source') or {}).get('editing') or {}
|
|
if (draft.get('name') == ' Retain parent 雨 ' and source.get('name') == ' Retain source Δ '
|
|
and (mode == 'before-stop' or j.get('origin') is None)):
|
|
matches.append((p, raw, j, draft, source))
|
|
assert len(matches) == 1, f'Expected one completed matching recovery record, got {len(matches)}'
|
|
p, raw, j, draft, source = matches[0]
|
|
assert j['accepted'] == {'entry': 'project.arcadia', 'files': expected}
|
|
assert j['workspace']['entry'] == 'project.arcadia'
|
|
assert j['workspace']['applied'] is True
|
|
assert j['workspace']['files'] == {k: [v, v] for k, v in expected.items()}
|
|
assert set(j['workspace']['reachable']) == set(expected)
|
|
assert source['kind'] == 'Activity' and source['layer'] == 'Oa' and source['parent'] is None
|
|
assert source['document'] == 'd1510000-0000-4000-8000-000000000001'
|
|
assert source['scope'] == {'Lifeline': {'owner': 'd1510000-0000-4000-8000-000000000002', 'viewpoint': 'OperationalActivity'}}
|
|
assert draft['tool'] == 'Lifeline' and draft['source']['staged'] is None
|
|
assert draft['owner'] == 'd1510000-0000-4000-8000-000000000002'
|
|
ids = [draft['invocation'], source['id'], source['token'], *source['identities']]
|
|
assert len(ids) == 4 and len(set(ids)) == 4
|
|
assert all(uuid.UUID(i).version == 4 for i in ids)
|
|
if mode == 'before-stop':
|
|
assert {p.name: p.read_text(encoding='utf-8') for p in (trial / 'work').glob('*.arcadia')} == expected
|
|
else:
|
|
before = json.loads((trial / 'recovery-evidence/before-stop.json').read_bytes())
|
|
assert j['accepted'] == before['accepted']
|
|
assert j['behavior'] == before['behavior'], 'Authored pending behavior/identities changed on restore'
|
|
assert j['origin'] is None, 'Recovered copy must not retain writable source origin'
|
|
out = trial / 'recovery-evidence'
|
|
out.mkdir(exist_ok=True)
|
|
with (out / f'{mode}.json').open('xb') as f:
|
|
f.write(raw)
|
|
result = {'atUtc': datetime.datetime.now(datetime.timezone.utc).isoformat(), 'mode': mode,
|
|
'source': str(p), 'sha256': hashlib.sha256(raw).hexdigest(), 'reservedIds': ids,
|
|
'acceptedBaselineFiles': len(expected), 'rawParentAndChildNames': 'exact',
|
|
'kindAndRoot': 'Activity/Oa/root', 'result': 'PASS'}
|
|
with (out / f'{mode}-result.json').open('x', encoding='utf-8') as f:
|
|
json.dump(result, f, ensure_ascii=False, indent=2)
|
|
f.write('\n')
|
|
print(json.dumps(result, ensure_ascii=False))
|