Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 24 additions & 3 deletions src/universal_law_query/stub_verify.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,11 +24,32 @@ def load_candidates(root: Path|None=None)->list[dict]:
if data.get('scientific_status_authority') is not False: raise SystemExit('REFUSED: candidate bundle must deny scientific authority: '+bundle_path.name)
if bundle_path.name=='CANDIDATE_DOWNSTREAM_GATE_STUBS.json': _validate_downstream_snapshot(data)
rows.extend(data['artifacts'])
seen=set();out=[]
# The standalone RN stub is also present in a public bundle with a
# different commentary note. Notes are not consumed by the verifier.
# Every other field, including unknown future fields, must agree.
def verifying_fields(item):
return {name:value for name,value in item.items() if name!='note'}
def same_json_value(left, right):
# JSON booleans and numbers are distinct even though Python says
# True == 1 == 1.0. Compare types recursively, including future keys.
if type(left) is not type(right): return False
if isinstance(left, dict):
return left.keys()==right.keys() and all(
same_json_value(left[key],right[key]) for key in left)
if isinstance(left, list):
return len(left)==len(right) and all(
same_json_value(a,b) for a,b in zip(left,right))
return left==right
seen={};out=[]
for row in rows:
key=row['key']
if key in seen: continue
seen.add(key);out.append(row)
if key in seen:
# Refuse conflicting identities, access or scopes before any fetch;
# preserve the first source annotation for note-only repeats.
if not same_json_value(verifying_fields(row),verifying_fields(seen[key])):
raise SystemExit('REFUSED: conflicting candidate key: '+str(key))
continue
seen[key]=row;out.append(row)
return out

def validate_row(row:dict)->None:
Expand Down
105 changes: 105 additions & 0 deletions tests/test_stub_verify_snapshot.py
Original file line number Diff line number Diff line change
Expand Up @@ -150,6 +150,111 @@ def test_duplicate_key_does_not_hide_inconsistent_snapshot(self):
with self.assertRaisesRegex(SystemExit, 'REFUSED:.*math_tip'):
verifier.load_candidates(self.root)

def test_identical_candidate_duplicates_keep_first_row(self):
original = row('shared')
self.write_bundle(
{'scientific_status_authority': False,
'artifacts': [original, dict(original)]},
'CANDIDATE_PUBLIC_REPLAY_STUBS.json')
self.assertEqual(verifier.load_candidates(self.root), [original])

def test_note_only_duplicate_is_not_identity_conflict(self):
original = row('shared')
original['note'] = 'Historical standalone commentary; candidate only.'
second = dict(original)
second['note'] = 'Bundle commentary; candidate only.'
self.write_bundle(
{'scientific_status_authority': False,
'artifacts': [original, second]},
'CANDIDATE_PUBLIC_REPLAY_STUBS.json')
# The first annotation is kept; neither annotation is an acceptance.
self.assertEqual(verifier.load_candidates(self.root), [original])

def test_conflicting_candidate_duplicates_in_one_bundle_are_refused(self):
original = row('duplicate')
original['scope'] = 'same declared scope'
# Each record is individually well-formed; the conflict must not be
# ignored simply because a valid first occurrence has already been seen.
changes = (
('commit', OLDER), ('sha256', '0' * 64),
('bytes', len(PAYLOAD) + 1),
('path', 'frontiers/example/another.md'),
('visibility', 'private'),
('scope', 'conflicting scope'),
('unlisted_future_field', 'unexpected distinct value'))
for field, value in changes:
with self.subTest(field=field):
conflicting = dict(original)
conflicting[field] = value
self.write_bundle(
{'scientific_status_authority': False,
'artifacts': [original, conflicting]},
'CANDIDATE_PUBLIC_REPLAY_STUBS.json')
with self.assertRaisesRegex(
SystemExit, 'REFUSED: conflicting candidate key: duplicate'):
verifier.load_candidates(self.root)

def test_conflicting_single_and_bundle_duplicate_refused_before_fetch(self):
original = row('shared')
(self.portable / 'RN_FIXED_REMOTE_REPLAY_STUB.json').write_text(
json.dumps(original), encoding='utf-8')
conflicting = dict(original)
conflicting['sha256'] = '0' * 64
self.write_bundle(
{'scientific_status_authority': False,
'artifacts': [conflicting]},
'CANDIDATE_PUBLIC_REPLAY_STUBS.json')
fetch = mock.Mock()
with mock.patch.dict(os.environ, {'QUERY_STUB_VERIFY': '1'}), \
self.assertRaisesRegex(
SystemExit, 'REFUSED: conflicting candidate key: shared'):
verifier.main(
[], load_candidates_fn=lambda: verifier.load_candidates(self.root),
fetch_fn=fetch)
fetch.assert_not_called()

def test_duplicate_json_type_collisions_refused_before_fetch(self):
# JSON distinguishes true, 1 and 1.0; Python dict equality does not.
# Unknown nested future fields are identity-bearing, except 'note'.
changes = (
('integer_vs_boolean', 1, True),
('integer_vs_float', 1, 1.0),
('nested_integer_vs_boolean',
{'items': [{'ordinal': 1}]},
{'items': [{'ordinal': True}]}),
)
for name, left, right in changes:
with self.subTest(name=name):
original = row('typed')
original['unlisted_future_field'] = left
conflicting = dict(original)
conflicting['unlisted_future_field'] = right
self.write_bundle(
{'scientific_status_authority': False,
'artifacts': [original, conflicting]},
'CANDIDATE_PUBLIC_REPLAY_STUBS.json')
fetch = mock.Mock()
with mock.patch.dict(os.environ, {'QUERY_STUB_VERIFY': '1'}), \
self.assertRaisesRegex(
SystemExit, 'REFUSED: conflicting candidate key: typed'):
verifier.main(
[], load_candidates_fn=lambda: verifier.load_candidates(self.root),
fetch_fn=fetch)
fetch.assert_not_called()

def test_duplicate_bool_byte_count_does_not_equal_integer(self):
original = row('typed-size')
original['bytes'] = 1
conflicting = dict(original)
conflicting['bytes'] = True
self.write_bundle(
{'scientific_status_authority': False,
'artifacts': [original, conflicting]},
'CANDIDATE_PUBLIC_REPLAY_STUBS.json')
with self.assertRaisesRegex(
SystemExit, 'REFUSED: conflicting candidate key: typed-size'):
verifier.load_candidates(self.root)

def test_current_checkout_retains_ten_unique_candidate_keys(self):
keys = [item['key'] for item in verifier.load_candidates(ROOT)]
self.assertEqual(len(keys), 10)
Expand Down
Loading