1510 lines
87 KiB
Python
1510 lines
87 KiB
Python
#!/usr/bin/env python3
|
|
"""Local semantic-checker tests; synthetic examples are never game evidence."""
|
|
import copy
|
|
import ast
|
|
import unittest
|
|
import sys
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import hashlib
|
|
import struct
|
|
import zlib
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
sys.dont_write_bytecode = True
|
|
import check_p23_life_observation_evidence as check
|
|
|
|
|
|
class ActorNonoverlapTests(unittest.TestCase):
|
|
def report(self, second_at):
|
|
return dict(actor_drawings=[
|
|
dict(key='guest/a', at=[0, 0], scale=1, mode='idle', frame=0, flip=False),
|
|
dict(key='staff/b', at=second_at, scale=1, mode='idle', frame=0, flip=False)], cabins=[])
|
|
|
|
def test_edge_contact_is_not_overlap(self):
|
|
for at in ([6, 0], [0, 12], [-6, 0], [0, -12]):
|
|
with self.subTest(at=at):
|
|
check.validate_actor_nonoverlap(self.report(at))
|
|
|
|
def test_overlap_ignores_asserted_verdicts_and_offscreen_labels(self):
|
|
report = self.report([5, 11])
|
|
for actor in report['actor_drawings']:
|
|
actor.update(mechanical_pass=True, eligibility='offscreen', rect=[999, 999, 1, 1])
|
|
with self.assertRaisesRegex(ValueError, 'actor-actor core overlap: guest/a / staff/b'):
|
|
check.validate_actor_nonoverlap(report)
|
|
|
|
def test_cabin_passenger_is_checked_against_floor_actor(self):
|
|
report = self.report([0, 0])
|
|
report['cabins'] = [dict(passengers=[report['actor_drawings'].pop()])]
|
|
with self.assertRaisesRegex(ValueError, 'actor-actor core overlap'):
|
|
check.validate_actor_nonoverlap(report)
|
|
|
|
def test_invalid_footprints_are_not_silently_skipped(self):
|
|
for scale in (0, -1, float('nan'), float('inf'), True):
|
|
report = self.report([100, 100])
|
|
report['actor_drawings'][0]['scale'] = scale
|
|
with self.subTest(scale=scale), self.assertRaisesRegex(ValueError, 'invalid actor footprint'):
|
|
check.validate_actor_nonoverlap(report)
|
|
|
|
def test_r13_false_row_11_overlaps_staff_row_29(self):
|
|
fixture = json.loads((check.ROOT / 'tests/fixtures/p23-03/r13_actor_overlap.json').read_text())
|
|
self.assertEqual(fixture['independent_judgment'], [11, False, False, False])
|
|
with self.assertRaisesRegex(ValueError,
|
|
'actor-actor core overlap: guest/guest-auto-00000041 / staff/staff_gina'):
|
|
check.validate_actor_nonoverlap(fixture['report_inputs'])
|
|
# Exercise the actual offline evidence gate, before any audit assertions.
|
|
with self.assertRaisesRegex(ValueError, 'actor-actor core overlap'):
|
|
check.validate_drawing_rows(fixture['report_inputs'], {'rows': []})
|
|
|
|
def test_overlapping_full_boxes_with_disjoint_cores_pass(self):
|
|
# Both 16x18 boxes overlap by 9x18, but their six-pixel cores do not.
|
|
check.validate_actor_nonoverlap(self.report([7, 0]))
|
|
|
|
def test_walk_bob_is_part_of_core(self):
|
|
report = self.report([0, 12])
|
|
report['actor_drawings'][1].update(mode='walk', frame=2)
|
|
with self.assertRaisesRegex(ValueError, 'actor-actor core overlap'):
|
|
check.validate_actor_nonoverlap(report)
|
|
report['actor_drawings'][1]['frame'] = 3
|
|
check.validate_actor_nonoverlap(report)
|
|
|
|
def test_flipped_core_and_lying_breathing_bounds(self):
|
|
actor = self.report([0, 0])['actor_drawings'][0]
|
|
actor.update(flip=True, scale=3)
|
|
self.assertEqual(check.actor_identity_core(actor), [12, 3, 18, 36])
|
|
actor.update(mode='standard_room_sleep', frame=2)
|
|
self.assertEqual(check.actor_identity_core(actor), [3, 30, 39, 18])
|
|
actor['frame'] = 5
|
|
self.assertEqual(check.actor_identity_core(actor), [3, 33, 39, 15])
|
|
actor.update(mode='collapsed', flip=False)
|
|
self.assertEqual(check.actor_identity_core(actor), [6, 33, 39, 15])
|
|
|
|
def test_missing_pose_is_rejected(self):
|
|
report = self.report([100, 0])
|
|
del report['actor_drawings'][0]['frame']
|
|
with self.assertRaisesRegex(ValueError, 'missing actor pose'):
|
|
check.validate_actor_nonoverlap(report)
|
|
|
|
def test_duplicate_actor_is_rejected(self):
|
|
report = self.report([100, 0])
|
|
report['actor_drawings'][1]['key'] = 'guest/a'
|
|
with self.assertRaisesRegex(ValueError, 'duplicate actor drawing'):
|
|
check.validate_actor_nonoverlap(report)
|
|
|
|
def test_forged_transit_flag_does_not_exempt_overlap(self):
|
|
report = self.report([0, 0])
|
|
report['actor_drawings'][1]['in_transit'] = True
|
|
with self.assertRaisesRegex(ValueError, 'actor-actor core overlap'):
|
|
check.validate_actor_nonoverlap(report)
|
|
|
|
def test_core_table_requires_exact_art_source(self):
|
|
with mock.patch.object(check, 'PIXEL_PEOPLE_CORE_SHA256', '0' * 64):
|
|
with self.assertRaisesRegex(ValueError, 'PixelPeople core source SHA256'):
|
|
check.validate_actor_nonoverlap(self.report([100, 0]))
|
|
|
|
def test_recorded_service_region_and_logical_cell_have_no_common_area(self):
|
|
# Diagnostic of the unresolved simultaneous-bounds requirement, not
|
|
# permission to hide these occupants or bypass containment checks.
|
|
fixture = json.loads((check.ROOT / 'tests/fixtures/p23-03/r13_service_region_gap.json').read_text())
|
|
for guest in fixture['guests']:
|
|
traffic = next(r for r in fixture['traffic'] if r['cell'] == guest['cell'])
|
|
x, y, w, h = traffic['rect']
|
|
cell = [x - 2, y - 2, w + 4, h + 4] # get_cell_rect(...).grow(-2)
|
|
pixel = cell[2] / 16
|
|
x, y, w, h = fixture['room_rects'][guest['target']]
|
|
region = [x + pixel, y + pixel, w - 2 * pixel, h - 2 * pixel]
|
|
with self.subTest(guest=guest['id']):
|
|
self.assertLessEqual(min(cell[0] + cell[2], region[0] + region[2])
|
|
- max(cell[0], region[0]), 0)
|
|
|
|
|
|
class LifeOverlayTests(unittest.TestCase):
|
|
def report(self):
|
|
return dict(observation_serial=7, overlay_consumed_serial=7,
|
|
overlay_passes=['traffic', 'actors'], overlay_child_indices=[0, 1, 2],
|
|
actor_drawings=[dict(key='guest/a', at=[20, 30], scale=3,
|
|
mode='idle', frame=0, flip=False)], cabins=[],
|
|
traffic=[dict(rect=[2, 2, 60, 92], band_rect=[9.2, 85.5, 45.6, 5],
|
|
traffic=dict(current_density=1, temporary_block=False))])
|
|
|
|
def test_front_and_back_valid_consumed_band(self):
|
|
for layer in (0, 1):
|
|
report = self.report()
|
|
report['layer'] = layer
|
|
check.validate_life_overlay(report)
|
|
|
|
def test_stale_serial_fails(self):
|
|
report = self.report()
|
|
report['overlay_consumed_serial'] = 6
|
|
with self.assertRaisesRegex(ValueError, 'stale LifeOverlay serial'):
|
|
check.validate_life_overlay(report)
|
|
|
|
def test_band_after_actors_fails(self):
|
|
report = self.report()
|
|
report['overlay_passes'].reverse()
|
|
with self.assertRaisesRegex(ValueError, 'LifeOverlay pass order'):
|
|
check.validate_life_overlay(report)
|
|
|
|
def test_overlay_below_inactive_art_fails(self):
|
|
report = self.report()
|
|
report['overlay_child_indices'] = [1, 0, 2]
|
|
with self.assertRaisesRegex(ValueError, 'LifeOverlay child order'):
|
|
check.validate_life_overlay(report)
|
|
|
|
def test_core_crossed_by_actual_band_fails(self):
|
|
report = self.report()
|
|
report['actor_drawings'][0]['at'][1] = 60
|
|
with self.assertRaisesRegex(ValueError, 'traffic band covers actor core'):
|
|
check.validate_life_overlay(report)
|
|
|
|
def test_forged_band_geometry_fails(self):
|
|
report = self.report()
|
|
report['traffic'][0]['band_rect'][1] = 50
|
|
with self.assertRaisesRegex(ValueError, 'traffic band rectangle'):
|
|
check.validate_life_overlay(report)
|
|
|
|
def test_report_cannot_excuse_wrong_source_order(self):
|
|
source = (check.ROOT / 'scenes/main/world_preview.gd').read_text()
|
|
# Claimed pass order stays valid; actual callback source is inverted.
|
|
broken = source.replace('_draw_congestion_overlay(_life_overlay)', 'SWAP')
|
|
broken = broken.replace('_draw_actors(_life_overlay)', '_draw_congestion_overlay(_life_overlay)')
|
|
broken = broken.replace('SWAP', '_draw_actors(_life_overlay)')
|
|
with mock.patch.object(Path, 'read_text', return_value=broken):
|
|
with self.assertRaisesRegex(ValueError, 'LifeOverlay source pass order'):
|
|
check.validate_life_overlay(self.report())
|
|
|
|
|
|
class SetupBudgetTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.source = (check.ROOT / 'tests/fixtures/p23-03/observation.gd').read_text()
|
|
|
|
def test_all_planned_spends_enter_common_needed_and_deltas(self):
|
|
self.assertEqual(check.check_setup_budget_source(self.source),
|
|
{'construction', 'signing', 'purchase'})
|
|
|
|
def test_missing_spend_category_fails_before_os_replay(self):
|
|
expressions = {
|
|
'construction': 'needed[key] += int(definition.construction_cost.get(key, 0))',
|
|
'signing': 'needed[&"gold"] += signing_total',
|
|
'purchase': 'needed[&"gold"] += purchase_total',
|
|
}
|
|
for category, expression in expressions.items():
|
|
with self.subTest(category=category), self.assertRaisesRegex(ValueError, category):
|
|
check.check_setup_budget_source(self.source.replace(expression, 'pass'))
|
|
|
|
def test_missing_delta_coverage_fails(self):
|
|
with self.assertRaisesRegex(ValueError, 'deltas'):
|
|
check.check_setup_budget_source(self.source.replace(
|
|
'deltas[key] = needed[key] - state.get_resource(key)', 'pass'))
|
|
|
|
|
|
class QueuedElevatorTests(unittest.TestCase):
|
|
def setUp(self):
|
|
fixture = json.loads((check.ROOT / 'tests/fixtures/p23-03/queued_elevator.json').read_text())
|
|
self.state = {'systems': {'runtime': {'elevators': fixture['saved']}}}
|
|
|
|
def test_queued_trip_without_cabin_seat_uses_config_shaft(self):
|
|
saved = self.state['systems']['runtime']['elevators']
|
|
self.assertNotIn('shaft_id', saved['shafts'][0])
|
|
self.assertEqual(saved['shafts'][0]['boarded'], [])
|
|
report = check.elevator_report(self.state)
|
|
self.assertEqual(report['shafts'][0]['waiting_count'], 1)
|
|
self.assertEqual(report['shafts'][0]['passenger_count'], 0)
|
|
self.assertAlmostEqual(report['shafts'][0]['visual_floor'], .9)
|
|
self.assertEqual(report['trips'], saved['trips'])
|
|
|
|
def test_tampered_shaft_attribution_fails(self):
|
|
self.state['systems']['runtime']['elevators']['trips'][0]['shaft_id'] = 'wrong.shaft'
|
|
with self.assertRaisesRegex(ValueError, 'shaft attribution'):
|
|
check.elevator_report(self.state)
|
|
|
|
def test_wrong_waiting_count_fails_whole_report_comparison(self):
|
|
expected = check.elevator_report(self.state)
|
|
tampered = copy.deepcopy(expected)
|
|
tampered['shafts'][0]['waiting_count'] = 0
|
|
directory = mock.MagicMock()
|
|
manifest = dict(label='queued-fixture', pid=123, state='state',
|
|
state_after='state', report='report')
|
|
directory.__truediv__.return_value.open.return_value = io.StringIO(json.dumps(manifest)+'\n')
|
|
# Exercise the actual stream comparison; isolate unrelated actor fields.
|
|
with mock.patch.object(check, 'read_object', side_effect=[self.state, self.state,
|
|
{'elevators': tampered}]), \
|
|
mock.patch.object(check, 'compare_actor_rows', return_value=[]):
|
|
with self.assertRaisesRegex(ValueError, 'typed elevator report mismatch'):
|
|
check.inspect_stream(directory)
|
|
|
|
def test_extra_and_missing_trip_keys_fail(self):
|
|
for mutation in ('extra', 'missing'):
|
|
with self.subTest(mutation=mutation):
|
|
state = copy.deepcopy(self.state)
|
|
trip = state['systems']['runtime']['elevators']['trips'][0]
|
|
if mutation == 'extra':
|
|
trip['unexpected'] = True
|
|
else:
|
|
del trip['shaft_id']
|
|
with self.assertRaisesRegex(ValueError, 'exact elevator trip keys'):
|
|
check.elevator_report(state)
|
|
|
|
|
|
class SnapshotCellTests(unittest.TestCase):
|
|
def test_layout_arrival_compares_serialized_cell_arrays(self):
|
|
source = (check.ROOT / 'tests/fixtures/p23-03/layout_comparison.gd').read_text()
|
|
serializer = (check.ROOT / 'scripts/guest/guest_runtime.gd').read_text()
|
|
self.assertIn('&"current_cell": _cell_to_array(current_cell)', serializer)
|
|
goal = check.re.search(r'const GOAL := Vector3i\(([^)]+)\)', source)
|
|
self.assertIsNotNone(goal)
|
|
coordinates = ast.literal_eval('[' + goal.group(1) + ']')
|
|
comparison = check.re.search(r'if guest\[&"current_cell"\] != (.+?) or finished\.has', source)
|
|
self.assertIsNotNone(comparison)
|
|
expression = comparison.group(1)
|
|
self.assertEqual(expression, '[GOAL.x, GOAL.y, GOAL.z]')
|
|
for axis, value in zip('xyz', coordinates):
|
|
expression = expression.replace('GOAL.' + axis, str(value))
|
|
target = ast.literal_eval(expression)
|
|
self.assertEqual(json.loads('[2, 1, 0]'), target)
|
|
for cell in ([2, 0, 0], [3, 1, 0], [2, 1, 1]):
|
|
self.assertNotEqual(cell, target)
|
|
|
|
|
|
class ImageTests(unittest.TestCase):
|
|
def test_collector_conversion_is_bound_to_each_cua_screenshot(self):
|
|
runner = check.module('p23_unit_collector', check.ROOT/'tools/run_p23_life_observation.py')
|
|
with tempfile.TemporaryDirectory(dir=check.ROOT/'tests/fixtures/p23-03') as temporary:
|
|
root = Path(temporary)
|
|
pair = root/'os-pair/A'
|
|
request_dir = pair/'writer-request-0001'
|
|
request_dir.mkdir(parents=True)
|
|
request = request_dir/'request.json'
|
|
request.write_text(json.dumps(dict(sequence=1, target_pid=123)))
|
|
raw = request_dir/'raw.jpg'
|
|
raw.write_bytes(b'\xff\xd8unit-only\xff\xd9')
|
|
shot = dict(path=str(raw.relative_to(root)), raw_sha256=check.sha(raw), tool_call_id='unit-shot-call')
|
|
receipt = request_dir/'cua.json'
|
|
receipt.write_text(json.dumps(dict(screenshots=[shot])))
|
|
ack = request_dir/'ack.json'
|
|
ack.write_text(json.dumps(dict(request_sha256=check.sha(request), input_receipt=str(receipt.relative_to(root)))))
|
|
(pair/'writer-requests.json').write_text(json.dumps(dict(requests=[str(ack.relative_to(root))])))
|
|
def convert(raw_path, png_path, cua, receipt_path):
|
|
self.assertEqual(cua, shot)
|
|
self.assertEqual(Path(raw_path).read_bytes(), b'\xff\xd8unit-only\xff\xd9')
|
|
Path(png_path).write_bytes(self.png())
|
|
Path(receipt_path).write_text('{}')
|
|
with mock.patch.object(runner, 'module', return_value=check), \
|
|
mock.patch.object(check, 'image_kind', return_value=('jpeg', 2, 2)), \
|
|
mock.patch.object(runner, 'convert_screenshot', side_effect=convert) as converter:
|
|
output = runner.collect_screenshots(root, pair, 'writer')
|
|
converter.assert_called_once()
|
|
rows = json.loads((root/output).read_text())
|
|
self.assertEqual(rows[0]['tool_call_id'], shot['tool_call_id'])
|
|
self.assertEqual(rows[0]['raw_sha256'], check.sha(raw))
|
|
self.assertEqual(rows[0]['png_sha256'], check.sha(root/rows[0]['png_path']))
|
|
self.assertTrue(rows[0]['conversion'])
|
|
|
|
@staticmethod
|
|
def png(mode=0, channels=3):
|
|
def chunk(kind, body):
|
|
return struct.pack('>I', len(body)) + kind + body + struct.pack('>I', zlib.crc32(kind + body))
|
|
rows = [bytes([10, 20, 30, 40][:channels] * 2),
|
|
bytes([50, 60, 70, 80][:channels] * 2)]
|
|
filtered = bytearray()
|
|
prior = bytes(2 * channels)
|
|
for row in rows:
|
|
filtered.append(mode)
|
|
for x, value in enumerate(row):
|
|
left, up, corner = (row[x-channels] if x >= channels else 0,
|
|
prior[x], prior[x-channels] if x >= channels else 0)
|
|
p = left + up - corner
|
|
distances = [abs(p-left), abs(p-up), abs(p-corner)]
|
|
predictors = [0, left, up, (left+up)//2,
|
|
[left, up, corner][distances.index(min(distances))]]
|
|
filtered.append((value-predictors[mode]) % 256)
|
|
prior = row
|
|
header = struct.pack('>IIBBBBB', 2, 2, 8, 2 if channels == 3 else 6, 0, 0, 0)
|
|
return (b'\x89PNG\r\n\x1a\n' + chunk(b'IHDR', header)
|
|
+ chunk(b'IDAT', zlib.compress(filtered)) + chunk(b'IEND', b''))
|
|
|
|
def test_all_png_filters_rgb_and_rgba(self):
|
|
for mode in range(5):
|
|
for channels in (3, 4):
|
|
with self.subTest(mode=mode, channels=channels):
|
|
w, h, rows = check.png_pixels(self.png(mode, channels))
|
|
self.assertEqual((w, h), (2, 2))
|
|
self.assertEqual(rows[0][0], tuple([10, 20, 30, 40][:channels]))
|
|
self.assertEqual(rows[1][1], tuple([50, 60, 70, 80][:channels]))
|
|
|
|
def test_crc_and_trailing_bytes_rejected(self):
|
|
data = bytearray(self.png())
|
|
data[29] ^= 1
|
|
with self.assertRaisesRegex(ValueError, 'CRC mismatch'):
|
|
check.png_pixels(bytes(data))
|
|
with self.assertRaisesRegex(ValueError, 'invalid IEND'):
|
|
check.png_pixels(self.png() + b'extra')
|
|
|
|
def conversion(self):
|
|
raw, png = b'\xff\xd8synthetic-unit-test\xff\xd9', self.png()
|
|
digest = lambda b: hashlib.sha256(b).hexdigest()
|
|
rec = dict(raw_path='/raw.png', raw_sha256=digest(raw), raw_kind='jpeg',
|
|
tool_call_id='unit-test-call', converter='/usr/bin/sips', converter_sha256='a'*64,
|
|
argv=['/usr/bin/sips', '-s', 'format', 'png', '/raw.png', '--out', '/decoded.png'],
|
|
exit_code=0, stdout_sha256=digest(b''), stderr_sha256=digest(b''),
|
|
png_path='/decoded.png', png_sha256=digest(png), width=2, height=2)
|
|
cua = dict(raw_sha256=digest(raw), tool_call_id='unit-test-call')
|
|
return rec, cua, {'/raw.png': raw, '/decoded.png': png}
|
|
|
|
def test_conversion_chain_and_specific_forgery_deficits(self):
|
|
rec, cua, files = self.conversion()
|
|
helper = mock.Mock()
|
|
helper.jpeg_dimensions.return_value = (2, 2)
|
|
with mock.patch.object(Path, 'read_bytes', lambda p: files[str(p)]), \
|
|
mock.patch.object(check, 'module', return_value=helper):
|
|
self.assertEqual(check.inspect_conversion(rec, cua, 'a'*64)[:2], (2, 2))
|
|
for key, value, deficit in [('converter_sha256', 'b'*64, 'converter SHA mismatch'),
|
|
('argv', [], 'converter argv mismatch'),
|
|
('raw_sha256', 'b'*64, 'CUA raw hash/call mismatch'),
|
|
('width', 3, 'conversion dimensions mismatch')]:
|
|
with self.subTest(key=key):
|
|
bad = dict(rec, **{key: value})
|
|
with self.assertRaisesRegex(ValueError, deficit):
|
|
check.inspect_conversion(bad, cua, 'a'*64)
|
|
with self.assertRaisesRegex(ValueError, 'JPEG screenshot requires conversion record'):
|
|
check.screenshot_pixels('/raw.png')
|
|
|
|
|
|
class RestAuditTests(unittest.TestCase):
|
|
def fixture(self):
|
|
member = dict(staff_id='s', state='OFF_DUTY', health=100, current_task_id='', current_cell=[0, 0, 0])
|
|
facilities = [dict(instance_id=fid, definition_id='staff_waiting_room', origin=[0, 0, 0],
|
|
occupied_cells=[], state='OPEN', suspension_reason='') for fid in ('a', 'b')]
|
|
runtime = dict(staff=dict(staff=[member]), facilities=dict(instances=facilities),
|
|
defense=dict(raids=[]), events=dict(actor_locks={}), elevators=dict(trips=[]),
|
|
path=dict(cell_traffic=[dict(cell=[x, 0, 0]) for x in (-1, 0, 1)]),
|
|
world=dict(bounds=dict(min_x=-1, max_x=1, min_y=0, max_y=0), occupancies=[]))
|
|
routes = [dict(cell=[x, 0, 0], path=[[0, 0, 0], [x, 0, 0]] if x else [[0, 0, 0]],
|
|
edges=[dict(**{'from': [0, 0, 0], 'to': [x, 0, 0]}, total_cost=0)] if x else [],
|
|
calculate_path_cost=0) for x in (-1, 0, 1)]
|
|
observed = []
|
|
for f in facilities:
|
|
observed.append({k: f[k] for k in ('instance_id', 'definition_id', 'origin', 'occupied_cells')})
|
|
observed[-1].update(operation_state='OPERATING', probes=[dict(cell=[x, 0, 0], has_cell=True) for x in (0, -1, 1)],
|
|
ordered=[[x, 0, 0] for x in (-1, 0, 1)], routes=copy.deepcopy(routes))
|
|
selected = dict(reachable=True, resting=True, cell=[-1, 0, 0], facility_id='a', label='待機室で休憩中')
|
|
audit = dict(path_service_sha256=check.sha(check.ROOT/'scripts/world/path_service.gd'),
|
|
staff=[dict(member, defense_memberships=[], actor_lock=False, in_transit=False,
|
|
facilities=observed, comparison_only=selected)])
|
|
report = dict(resting_staff={'s': 'a'}, actor_drawings=[], layer=0, carried_patients={}, rooms={},
|
|
viewport=[1440, 810], zoom=1, pan=[0, 0])
|
|
return dict(systems=dict(runtime=runtime)), report, audit
|
|
|
|
def test_selection_ties_and_nonmutation(self):
|
|
state, report, audit = self.fixture()
|
|
before = copy.deepcopy((state, report, audit))
|
|
self.assertEqual(check.compare_rest(state, report, audit)['s']['cell'], [-1, 0, 0])
|
|
self.assertEqual((state, report, audit), before)
|
|
|
|
def test_rest_forgeries(self):
|
|
mutations = {
|
|
'selection': lambda a: a['staff'][0]['comparison_only'].update(facility_id='b'),
|
|
'tie': lambda a: a['staff'][0]['comparison_only'].update(cell=[0, 0, 0]),
|
|
'lock': lambda a: a['staff'][0].update(actor_lock=True),
|
|
'transit': lambda a: a['staff'][0].update(in_transit=True),
|
|
'label': lambda a: a['staff'][0]['comparison_only'].update(label='宿舎で休憩中'),
|
|
'cost': lambda a: a['staff'][0]['facilities'][0]['routes'][0]['edges'][0].update(total_cost=9),
|
|
'facility_order': lambda a: a['staff'][0]['facilities'].reverse(),
|
|
'candidate_order': lambda a: a['staff'][0]['facilities'][0]['ordered'].reverse(),
|
|
'empty_edge': lambda a: a['staff'][0]['facilities'][0]['routes'][0]['edges'][0].clear(),
|
|
}
|
|
for name, mutate in mutations.items():
|
|
with self.subTest(name=name):
|
|
state, report, audit = self.fixture()
|
|
mutate(audit)
|
|
with self.assertRaises(ValueError):
|
|
check.compare_rest(state, report, audit)
|
|
|
|
def test_ineligible_and_transit_recomputed(self):
|
|
state, report, audit = self.fixture()
|
|
state['systems']['runtime']['events']['actor_locks']['s'] = 'event'
|
|
audit['staff'][0].update(actor_lock=True, comparison_only={})
|
|
report['resting_staff'] = {}
|
|
self.assertEqual(check.compare_rest(state, report, audit), {'s': {}})
|
|
state, report, audit = self.fixture()
|
|
state['systems']['runtime']['elevators']['trips'] = [dict(actor_id='staff:s')]
|
|
audit['staff'][0]['in_transit'] = True
|
|
audit['staff'][0]['comparison_only'].update(resting=False, label='休憩先へ移動中')
|
|
report['resting_staff'] = {}
|
|
self.assertFalse(check.compare_rest(state, report, audit)['s']['resting'])
|
|
|
|
|
|
class InfrastructureTests(unittest.TestCase):
|
|
def test_matrix_requires_each_condition_and_each_category(self):
|
|
executed = {c: {k: set(v) for k, v in categories.items()} for c, categories in check.REQUIRED_CHECKS.items()}
|
|
self.assertTrue(all(r['status'] == 'passed' for r in check.condition_matrix(executed)))
|
|
for condition in check.CONDITIONS:
|
|
for category in check.CATEGORIES:
|
|
with self.subTest(condition=condition, category=category):
|
|
broken = copy.deepcopy(executed)
|
|
broken[condition][category].clear()
|
|
matrix = check.condition_matrix(broken)
|
|
self.assertEqual(next(r for r in matrix if r['condition'] == condition)['status'], 'pending')
|
|
self.assertIn(category, check.BLOCKERS(broken)[condition])
|
|
|
|
def test_distinct_pid_ack_and_formal_hash_negatives(self):
|
|
members = {name: 'a'*64 for name in ('world.bin', 'manifest.json', 'thumbnail.png', 'journal.log')}
|
|
writer = dict(actual_pid=1)
|
|
reader = dict(actual_pid=2, formal_input={k: dict(sha256=v) for k, v in members.items()})
|
|
saved = dict(pid=1, members=members)
|
|
records = [dict(request=dict(action='load_menu', target_pid=2, sequence=1, requested_ticks_usec=10, requested_unix_ns=90),
|
|
ack=dict(target_pid=2, sequence=1, request_sha256='b'*64, timestamp_ns=110),
|
|
observed=dict(pid=2, sequence=1, request_sha256='b'*64, ack_observed_ticks_usec=20, ack_observed_unix_ns=120, status='手動セーブを読み込みました。'),
|
|
cua=dict(target_pid=2, method='native_os', tool_call_id='actual-call', timestamp_ns=100),
|
|
request_sha256='b'*64)]
|
|
check.inspect_resume(writer, reader, saved, records, members)
|
|
for change in ('pid', 'ack', 'time', 'hash', 'id'):
|
|
with self.subTest(change=change):
|
|
r, q = copy.deepcopy(reader), copy.deepcopy(records)
|
|
if change == 'pid': r['actual_pid'] = 1
|
|
if change == 'ack': q.clear()
|
|
if change == 'time': q[0]['observed']['ack_observed_ticks_usec'] = 9
|
|
if change == 'hash': r['formal_input']['world.bin']['sha256'] = 'c'*64
|
|
if change == 'id': q[0]['ack']['sequence'] = 2
|
|
with self.assertRaises(ValueError): check.inspect_resume(writer, r, saved, q, members)
|
|
|
|
def test_interpolation_history_tampering(self):
|
|
state = dict(systems=dict(runtime=dict(guests=dict(guests=[]), staff=dict(staff=[]),
|
|
elevators=dict(shafts=[], trips=[], current_minute=0))))
|
|
report = dict(animation_time=0.0, paused=True, layer=0, zoom=1, pan=[0,0], viewport=[100,100],
|
|
actor_tracks={}, elevator_tracks={}, actor_drawings=[], cabins=[])
|
|
audit = dict(frames=[dict(frame=1, delta=.01, scaled_delta=0, display=copy.deepcopy(report)),
|
|
dict(frame=2, delta=.01, scaled_delta=0, display=copy.deepcopy(report))])
|
|
check.compare_motion(state, report, audit)
|
|
for change in ('delta', 'frame', 'decreasing_frame', 'no_frames', 'one_frame', 'display', 'report_binding'):
|
|
with self.subTest(change=change):
|
|
bad = copy.deepcopy(audit)
|
|
if change == 'delta': bad['frames'][1]['scaled_delta'] = 1
|
|
if change == 'frame': bad['frames'][1]['frame'] = 1
|
|
if change == 'decreasing_frame': bad['frames'][1]['frame'] = 0
|
|
if change == 'no_frames': bad['frames'].clear()
|
|
if change == 'one_frame': bad['frames'].pop()
|
|
if change == 'report_binding': bad['frames'][1]['display']['zoom'] = 2
|
|
if change == 'display': bad['frames'][1]['display']['animation_time'] = 1
|
|
with self.assertRaises(ValueError): check.compare_motion(state, report, bad)
|
|
|
|
def test_verify_only_missing_index_is_a_concrete_deficit(self):
|
|
with mock.patch.dict(check.os.environ, {'BUILD_OUTPUT_DIR': '/missing-p23-proof'}, clear=True):
|
|
with contextlib.redirect_stdout(io.StringIO()) as output:
|
|
self.assertEqual(check.main(['--verify-only']), 1)
|
|
result = json.loads(output.getvalue())
|
|
self.assertEqual(result['integrity'], 'fail')
|
|
self.assertEqual(result['acceptance'], 'pending')
|
|
self.assertIn('index.json', result['deficits'][0])
|
|
self.assertNotIn('blocked', result['deficits'][0])
|
|
|
|
def test_verify_only_requires_build_output(self):
|
|
with mock.patch.dict(check.os.environ, {}, clear=True):
|
|
with contextlib.redirect_stderr(io.StringIO()) as output:
|
|
self.assertEqual(check.main(['--verify-only']), 2)
|
|
self.assertIn('BUILD_OUTPUT_DIR is required', output.getvalue())
|
|
|
|
def test_normative_plan_change_invalidates_but_progress_does_not(self):
|
|
root = check.ROOT
|
|
text = (root / check.PLAN).read_text()
|
|
original = check.input_map(root)
|
|
progress = text.splitlines(keepends=True)
|
|
progress[2] = 'Progress-only test mutation\n'
|
|
with mock.patch.object(Path, 'read_text', return_value=''.join(progress)):
|
|
self.assertEqual(check.input_map(root), original)
|
|
changed = text.replace('同需要の動線改善を観察できる。', 'altered normative acceptance')
|
|
self.assertNotEqual(changed, text)
|
|
with mock.patch.object(Path, 'read_text', return_value=changed):
|
|
self.assertNotEqual(check.input_map(root), original)
|
|
self.assertNotIn(check.PLAN, original)
|
|
self.assertIn('full_plan_sha256', check.input_provenance(root))
|
|
for path in (root / 'tests/fixtures/p23-03').rglob('*'):
|
|
if path.is_file():
|
|
self.assertIn(path.relative_to(root).as_posix(), original)
|
|
|
|
def test_os_process_guards_reject_engine_profile_and_source_drift(self):
|
|
source = {'source.gd': 'a' * 64}
|
|
receipt = dict(engine_ended=True, passed=True, exit_code=0, error=None,
|
|
guard_errors=[], cleanup_errors=[], default_before={}, default_after={},
|
|
stage_before={}, stage_after={}, source_map=source,
|
|
source_after={'source.gd': {'sha256': 'a' * 64}},
|
|
owned_cleanup=[{'engine_ended': True}], ordinary_owner_proof={'present': True})
|
|
receipt.update(spec_sha256=check.SPEC_SHA, inputs={'spec': 'a'*64}, inputs_after={'spec': 'a'*64},
|
|
environment={'P23_LAYOUT': 'A'}, environment_after={'P23_LAYOUT': 'A'}, actual_pid=123,
|
|
profile='/private/test', profile_name='test', profile_before={'existed': False, 'files': {}}, profile_after={})
|
|
for field in ('engine_sha256', 'settings_sha256', 'environment_sha256'):
|
|
receipt[field] = receipt[field+'_after'] = 'a'*64
|
|
check.inspect_os_receipt(receipt, source)
|
|
for field, value in [('engine_ended', False), ('default_after', {'changed': True}),
|
|
('source_after', {'source.gd': {'sha256': 'b' * 64}}),
|
|
('stage_after', {'changed': True}), ('owned_cleanup', []),
|
|
('environment_after', {}), ('profile_before', {}),
|
|
('engine_sha256_after', 'b'*64), ('spec_sha256', 'b'*64)]:
|
|
with self.subTest(field=field):
|
|
altered = copy.deepcopy(receipt)
|
|
altered[field] = value
|
|
with self.assertRaises(ValueError):
|
|
check.inspect_os_receipt(altered, source)
|
|
|
|
|
|
class SemanticTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.guest = dict(guest_id='g', state='QUEUING', current_cell=[1, 0, 0],
|
|
target_facility_id='f', current_service_id='', queue_request_id='q')
|
|
self.staff = dict(staff_id='s', state='WORKING', current_cell=[2, 0, 1],
|
|
current_task_id='t', assigned_facility_id='f', role_id='porter',
|
|
carried_inventory_id='cargo')
|
|
self.task = dict(task_id='t', state='WORKING', task_type='delivery',
|
|
assigned_staff_id='s', target_entity_id='request', rescue_facility_id='')
|
|
self.snapshot = {'systems': {'runtime': {'guests': {'guests': [self.guest]},
|
|
'staff': {'staff': [self.staff], 'tasks': [self.task]}}}}
|
|
self.report = dict(guests=[dict(id='g', state='QUEUING', cell=[1, 0, 0], target='f', service='', queue='q')],
|
|
staff=[dict(id='s', state='WORKING', cell=[2, 0, 1], task='t', facility='f',
|
|
role='porter', carried_inventory_id='cargo')],
|
|
tasks=[dict(id='t', state='WORKING', type='delivery', staff='s',
|
|
target='request', rescue_facility='')])
|
|
|
|
def test_real_schema_projection(self):
|
|
self.assertEqual(check.compare_actor_rows(self.snapshot, self.report), ['guests', 'staff', 'tasks'])
|
|
|
|
def test_changed_state_position_target_task_and_goods_are_rejected(self):
|
|
for category, field, replacement in [('guests', 'cell', [8, 0, 0]), ('guests', 'target', 'other'),
|
|
('guests', 'state', 'MOVING'), ('staff', 'task', ''),
|
|
('staff', 'carried_inventory_id', ''), ('tasks', 'staff', 'other')]:
|
|
with self.subTest(category=category, field=field):
|
|
report = copy.deepcopy(self.report)
|
|
report[category][0][field] = replacement
|
|
with self.assertRaises(ValueError):
|
|
check.compare_actor_rows(self.snapshot, report)
|
|
|
|
def test_missing_extra_and_boolean_only_rows_are_rejected(self):
|
|
for bad in [[], self.report['guests'] * 2, [True]]:
|
|
report = copy.deepcopy(self.report)
|
|
report['guests'] = bad
|
|
with self.assertRaises(ValueError):
|
|
check.compare_actor_rows(self.snapshot, report)
|
|
|
|
def test_unknown_enum_and_duplicate_id_are_rejected(self):
|
|
self.guest['state'] = 'FICTIONAL'
|
|
with self.assertRaises(ValueError):
|
|
check.compare_actor_rows(self.snapshot, self.report)
|
|
self.guest['state'] = 'QUEUING'
|
|
self.snapshot['systems']['runtime']['guests']['guests'].append(copy.deepcopy(self.guest))
|
|
with self.assertRaises(ValueError):
|
|
check.compare_actor_rows(self.snapshot, self.report)
|
|
|
|
def test_typed_rectangle_and_duplicate_vector_key(self):
|
|
self.assertEqual(check.decode(dict(type='Rect2', text='Rect2(1, 2, 16, 24)')), [1, 2, 16, 24])
|
|
with self.assertRaises(ValueError):
|
|
check.decode(dict(type='Rect2', text='Rect2(1, nan, 16, 24)'))
|
|
key = dict(type='Vector3i', text='Vector3i(1, 2, 0)')
|
|
value = dict(type='int', text='1')
|
|
with self.assertRaises(ValueError):
|
|
check.decode(dict(type='Dictionary', entries=[[key, value], [key, value]]))
|
|
|
|
def test_immutable_typed_input_is_not_modified(self):
|
|
before = copy.deepcopy(self.snapshot)
|
|
check.compare_actor_rows(self.snapshot, self.report)
|
|
self.assertEqual(before, self.snapshot)
|
|
|
|
|
|
class PackingTests(unittest.TestCase):
|
|
def test_scale_trial_rebuilds_vertical_groups(self):
|
|
members = self.members(2)
|
|
for member, y in zip(members, (118, 144)):
|
|
member['row'].update(center=[108, y], at=[84, y-15])
|
|
member['bounds'] = [80, 74, 20, 130]
|
|
# At p=3 the vertical intervals intersect; at p=2 they separate.
|
|
# Keeping the stale component would incorrectly force both down to p=1.
|
|
rows, badges = check.fitted_actor_layout(members, 4, 96)
|
|
self.assertEqual(badges, [])
|
|
self.assertEqual([rows['guest/'+str(i)]['scale'] for i in range(2)], [2, 2])
|
|
self.assertEqual([rows['guest/'+str(i)]['at'] for i in range(2)], [[76, 120], [76, 146]])
|
|
check.validate_actor_nonoverlap(dict(actor_drawings=list(rows.values()), cabins=[]))
|
|
|
|
def members(self, count=4):
|
|
# R13 cell dimensions and the actual _actor_offset sequence; no transit.
|
|
offsets = [0, 16, -16, 8, -8, 24, -24]
|
|
return [dict(row=dict(key='guest/'+str(i), center=[108+offsets[i % 7], 118-4*(i % 2)],
|
|
at=[84+offsets[i % 7], 103-4*(i % 2)], scale=3,
|
|
mode='idle', frame=0, flip=False, state=3, cell=[1, 1, 0]),
|
|
bounds=[80, 74, 56, 88], transit=False) for i in range(count)]
|
|
|
|
def test_four_movers_fit_at_two_not_three(self):
|
|
rows, badges = check.fitted_actor_layout(self.members(), 4, 96)
|
|
self.assertEqual(badges, [])
|
|
self.assertEqual([rows['guest/'+str(i)]['scale'] for i in range(4)], [2]*4)
|
|
self.assertEqual([rows['guest/'+str(i)]['at'] for i in range(4)],
|
|
[[88, 120], [112, 116], [76, 120], [100, 116]])
|
|
check.validate_actor_nonoverlap(dict(actor_drawings=list(rows.values()), cabins=[]))
|
|
|
|
def test_cross_kind_group_and_passenger(self):
|
|
members = self.members()
|
|
members[1]['row']['key'] = 'staff/worker'
|
|
members[2]['row']['key'] = 'guest/passenger'
|
|
members[2]['row']['scale'] = 2
|
|
members[2]['row']['at'] = [76, 120]
|
|
rows, badges = check.fitted_actor_layout(members, 4, 96)
|
|
self.assertEqual(badges, [])
|
|
self.assertEqual(len(rows), 4)
|
|
check.validate_actor_nonoverlap(dict(actor_drawings=list(rows.values()), cabins=[]))
|
|
|
|
def test_saturation_keeps_all_identities(self):
|
|
rows, badges = check.fitted_actor_layout(self.members(12), 4, 96)
|
|
self.assertEqual(len(rows), 9)
|
|
self.assertEqual(len(badges), 1)
|
|
self.assertEqual(len(badges[0]['members']), 12)
|
|
self.assertEqual(badges[0]['text'], '+3')
|
|
self.assertEqual(set(badges[0]['drawn_keys']), set(rows))
|
|
self.assertTrue(all(r['scale'] == 1 for r in rows.values()))
|
|
check.validate_actor_nonoverlap(dict(actor_drawings=list(rows.values()), cabins=[]))
|
|
|
|
def test_each_actor_keeps_own_bounds(self):
|
|
members = self.members(2)
|
|
members[0]['bounds'] = [80, 74, 12, 88]
|
|
members[1]['bounds'] = [124, 74, 12, 88]
|
|
rows, badges = check.fitted_actor_layout(members, 4, 96)
|
|
self.assertEqual(badges, [])
|
|
self.assertEqual(check.actor_identity_core(rows['guest/0'])[0:3:2], [80, 12])
|
|
self.assertEqual(check.actor_identity_core(rows['guest/1'])[0:3:2], [124, 12])
|
|
|
|
def test_receiving_cafeteria_uses_r13_room_not_access_cell(self):
|
|
fixture = json.loads((check.ROOT/'tests/fixtures/p23-03/r13_service_region_gap.json').read_text())
|
|
guest = fixture['guests'][0]
|
|
rect = fixture['room_rects'][guest['target']]
|
|
anchor = check.room_slots(rect, 4, 'cafeteria')[0]
|
|
center = [anchor[0], anchor[1]+12-48]
|
|
member = dict(row=dict(key='guest/'+guest['id'], center=center,
|
|
at=[center[0]-24, center[1]-15], scale=3,
|
|
mode='eat', frame=0, flip=False, state=5, cell=guest['cell']),
|
|
bounds=[rect[0]+4, rect[1]+4, rect[2]-8, rect[3]-8], transit=False)
|
|
rows, badges = check.fitted_actor_layout([member], 4, 96)
|
|
self.assertFalse(badges)
|
|
core = check.actor_identity_core(next(iter(rows.values())))
|
|
self.assertGreaterEqual(core[0], 1031)
|
|
self.assertEqual(next(iter(rows.values()))['cell'], [5, 0, 0])
|
|
|
|
|
|
class SpatialTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.runtime = dict(world=dict(bounds=dict(min_x=0, max_x=1, min_y=0, max_y=0),
|
|
walkable_cells=[[0, 0, 0], [1, 0, 0]], occupancies=[]),
|
|
path=dict(cell_traffic=[dict(cell=[0, 0, 0], base_capacity=2,
|
|
current_density=3, dynamic_cost=0, temporary_block=False)]),
|
|
facilities=dict(instances=[], requests=[]), guests=dict(guests=[]),
|
|
staff=dict(staff=[], tasks=[]), logistics_requests=[],
|
|
elevators=dict(current_minute=0, shafts=[], trips=[]))
|
|
self.snapshot = dict(systems=dict(runtime=self.runtime))
|
|
self.report = dict(viewport=[184, 164], zoom=1, pan=[0, 0], layer=0,
|
|
traffic=[dict(cell=[0, 0, 0], rect=[78, 72, 12, 20],
|
|
band_rect=[79.44, 90.4, 9.12, 3.2],
|
|
traffic=dict(base_capacity=2, current_density=3,
|
|
dynamic_cost=0, temporary_block=False))],
|
|
queue_positions={}, queue_hidden={}, landing_badges=[], rooms={},
|
|
treated_patients={}, carried_patients={}, resting_staff={},
|
|
goods=dict(requests=[], deliveries=[]), actor_drawings=[], cabins=[],
|
|
observation_serial=1, overlay_consumed_serial=1,
|
|
overlay_passes=['traffic', 'actors'], overlay_child_indices=[0, 1, 2])
|
|
|
|
def four_movers(self):
|
|
self.runtime['world'].update(bounds=dict(min_x=1, max_x=1, min_y=1, max_y=1), walkable_cells=[[1, 1, 0]])
|
|
self.report.update(viewport=[216, 236], paused=True, animation_time=0,
|
|
actor_tracks={}, boarded={}, actor_aggregations=[])
|
|
for i, (center, at) in enumerate(zip([[108,118], [124,114], [92,118], [116,114]],
|
|
[[88,120], [112,116], [76,120], [100,116]])):
|
|
key = 'guest/'+str(i)
|
|
self.runtime['guests']['guests'].append(dict(guest_id=str(i), current_cell=[1,1,0],
|
|
current_service_id='', target_facility_id='', state='MOVING'))
|
|
self.report['actor_tracks'][key] = dict(cell=[1,1,0], **{'from':[1,1,0]}, at=0, direction=1)
|
|
hashed = 5381
|
|
for char in key:
|
|
hashed = (hashed*33+ord(char)) & 0xffffffff
|
|
self.report['actor_drawings'].append(dict(key=key, center=center, at=at, scale=2,
|
|
cell=[1,1,0], mode='idle', frame=hashed%15, variant=hashed%15, flip=False,
|
|
staff=False, state=3, receiving=False, carrying=False, riding=False))
|
|
|
|
def test_four_movers_complete_geometry(self):
|
|
self.four_movers()
|
|
check.compare_actor_geometry(self.snapshot, self.report)
|
|
|
|
def moving_four(self):
|
|
self.four_movers()
|
|
self.report.update(paused=False, animation_time=.21, badges=[])
|
|
for row in self.report['actor_drawings']:
|
|
self.report['actor_tracks'][row['key']]['from'] = [0, 1, 0]
|
|
row['center'][0] -= 32
|
|
row.update(scale=3, mode='walk', frame=row['variant']+2)
|
|
row['at'] = [row['center'][0]-24, row['center'][1]-15]
|
|
|
|
def drawing_audit(self):
|
|
rows = []
|
|
for category, identifier, rect, source, text in check.drawing_rectangle_rows(self.report):
|
|
self.assertIsNone(text)
|
|
colors = list(dict.fromkeys(check.re.findall(r'Color\("([0-9a-fA-F]{6})"',
|
|
(check.ROOT/source).read_text())))
|
|
rows.append(dict(category=category, id=identifier, rect=rect,
|
|
style_source=source, style_colors=colors))
|
|
return dict(rows=rows)
|
|
|
|
def test_transit_raw_state_in_full_and_standalone_paths(self):
|
|
self.moving_four()
|
|
check.compare_actor_geometry(self.snapshot, self.report)
|
|
check.validate_drawing_rows(self.report, self.drawing_audit(), state=self.snapshot)
|
|
# Without authoritative context the overlapping moving cores fail closed.
|
|
with self.assertRaisesRegex(ValueError, 'core overlap'):
|
|
check.validate_drawing_rows(self.report, self.drawing_audit())
|
|
|
|
def test_transit_cannot_be_claimed_by_flags_or_invalid_tracks(self):
|
|
self.moving_four()
|
|
mutations = {
|
|
'paused': lambda r: r.update(paused=True),
|
|
'expired': lambda r: r.update(animation_time=.42),
|
|
'nonboolean_pause': lambda r: r.update(paused=0),
|
|
'future': lambda r: r['actor_tracks']['guest/0'].update(at=1),
|
|
'nonfinite': lambda r: r.update(animation_time=float('nan')),
|
|
'wrong_endpoint': lambda r: r['actor_tracks']['guest/0'].update(cell=[0, 1, 0]),
|
|
'teleport_origin': lambda r: r['actor_tracks']['guest/0'].update({'from': [-9, 1, 0]}),
|
|
}
|
|
for name, mutate in mutations.items():
|
|
report = copy.deepcopy(self.report)
|
|
mutate(report)
|
|
for row in report['actor_drawings']:
|
|
row.update(in_transit=True, region='transit')
|
|
with self.subTest(name=name), self.assertRaises(ValueError):
|
|
check.validate_drawing_rows(report, self.drawing_audit(), state=self.snapshot)
|
|
|
|
def test_staff_and_guests_share_complete_geometry_front_and_back(self):
|
|
self.four_movers()
|
|
self.runtime['guests']['guests'].pop()
|
|
self.runtime['staff']['staff'] = [dict(staff_id='3', current_cell=[1, 1, 0],
|
|
state='AVAILABLE', assigned_facility_id='', carried_inventory_id='', role_id='guard')]
|
|
self.report['actor_tracks']['staff/3'] = self.report['actor_tracks'].pop('guest/3')
|
|
row = self.report['actor_drawings'][-1]
|
|
hashed = 5381
|
|
for char in 'staff/3':
|
|
hashed = (hashed*33+ord(char)) & 0xffffffff
|
|
row.update(key='staff/3', staff=True, state=2, variant=hashed%15, frame=hashed%15)
|
|
for layer in (0, 1):
|
|
self.report['layer'] = layer
|
|
for member in self.runtime['guests']['guests'] + self.runtime['staff']['staff']:
|
|
member['current_cell'][2] = layer
|
|
for track in self.report['actor_tracks'].values():
|
|
track['cell'][2] = layer
|
|
track['from'][2] = layer
|
|
for drawing in self.report['actor_drawings']:
|
|
drawing['cell'][2] = layer
|
|
with self.subTest(layer=layer):
|
|
check.compare_actor_geometry(self.snapshot, self.report)
|
|
|
|
def test_spoofed_region_does_not_allow_core_outside_own_cell(self):
|
|
self.four_movers()
|
|
self.report['actor_drawings'][0].update(region='facility', in_transit=True, at=[500, 120])
|
|
with self.assertRaises(ValueError):
|
|
check.compare_actor_geometry(self.snapshot, self.report)
|
|
|
|
def test_aggregation_badge_is_required_in_drawing_audit(self):
|
|
self.four_movers()
|
|
self.report['badges'] = []
|
|
audit = self.drawing_audit()
|
|
self.report['actor_aggregations'] = [dict(at=[80, 74], text='+1',
|
|
members=[], drawn_keys=[])]
|
|
# This low-level test covers badge provenance only. Full geometry already
|
|
# rejects the unnecessary aggregate in test_fitted_geometry_rejects_tampering.
|
|
with self.assertRaisesRegex(ValueError, 'offline rectangle coverage'):
|
|
check.validate_drawing_rows(self.report, audit)
|
|
|
|
def test_low_world_pixel_complete_geometry_keeps_fractional_base(self):
|
|
self.four_movers()
|
|
self.runtime['guests']['guests'] = self.runtime['guests']['guests'][:1]
|
|
self.report.update(viewport=[156, 146], actor_drawings=self.report['actor_drawings'][:1])
|
|
self.report['actor_drawings'][0].update(center=[78, 73], at=[77, 72], scale=.1875)
|
|
check.compare_actor_geometry(self.snapshot, self.report)
|
|
|
|
def test_saturated_complete_geometry_keeps_every_identity(self):
|
|
self.four_movers()
|
|
prototype = copy.deepcopy(self.report['actor_drawings'][0])
|
|
self.runtime['guests']['guests'] = []
|
|
self.report['actor_drawings'] = []
|
|
# 10 upright cores need 60px at p=1, but the interval is only 56px.
|
|
# The first nine in preferred-x/key order occupy x82..136; guest/5
|
|
# cannot be retained. This expected layout is independent of the solver.
|
|
order = [6, 2, 9, 4, 0, 7, 3, 1, 8, 5]
|
|
identities = {}
|
|
for i in range(10):
|
|
key = 'guest/'+str(i)
|
|
cell = [1, 1, 0]
|
|
self.runtime['guests']['guests'].append(dict(guest_id=str(i), current_cell=cell,
|
|
current_service_id='', target_facility_id='', state='MOVING'))
|
|
self.report['actor_tracks'][key] = dict(cell=cell, **{'from': cell}, at=0, direction=1)
|
|
identities[i] = dict(key=key, id=str(i), cell=cell, state=3, mode='idle')
|
|
if i == 5:
|
|
continue
|
|
hashed = 5381
|
|
for char in key:
|
|
hashed = (hashed*33+ord(char)) & 0xffffffff
|
|
center = [108+[0,16,-16,8,-8,24,-24][i%7], 118-4*(i%2)]
|
|
self.report['actor_drawings'].append(dict(prototype, key=key, center=center,
|
|
at=[76+6*order.index(i), center[1]+19], scale=1, frame=hashed%15, variant=hashed%15))
|
|
self.report['actor_aggregations'] = [dict(members=[identities[i] for i in order],
|
|
drawn_keys=['guest/'+str(i) for i in order[:-1]], at=[80,74], text='+1')]
|
|
check.compare_actor_geometry(self.snapshot, self.report)
|
|
self.report['actor_aggregations'][0]['members'].pop()
|
|
with self.assertRaisesRegex(ValueError, 'aggregation/no-fit'):
|
|
check.compare_actor_geometry(self.snapshot, self.report)
|
|
|
|
def test_fitted_geometry_rejects_tampering(self):
|
|
self.four_movers()
|
|
for field, value in [('cell', [0,1,0]), ('state', 4), ('scale', 3), ('at', [999,120])]:
|
|
report = copy.deepcopy(self.report)
|
|
report['actor_drawings'][0][field] = value
|
|
with self.subTest(field=field), self.assertRaises(ValueError):
|
|
check.compare_actor_geometry(self.snapshot, report)
|
|
for mutation in ('missing', 'duplicate', 'aggregate'):
|
|
report = copy.deepcopy(self.report)
|
|
if mutation == 'missing':
|
|
report['actor_drawings'].pop()
|
|
elif mutation == 'duplicate':
|
|
report['actor_drawings'].append(copy.deepcopy(report['actor_drawings'][0]))
|
|
else:
|
|
report['actor_aggregations'] = [dict(drawn_keys=[], text='+4')]
|
|
with self.subTest(mutation=mutation), self.assertRaises(ValueError):
|
|
check.compare_actor_geometry(self.snapshot, report)
|
|
|
|
def test_density_and_pan_are_independently_recomputed(self):
|
|
check.compare_traffic(self.snapshot, self.report)
|
|
for key, value in [('current_density', 0), ('base_capacity', 9)]:
|
|
changed = copy.deepcopy(self.report)
|
|
changed['traffic'][0]['traffic'][key] = value
|
|
with self.assertRaises(ValueError):
|
|
check.compare_traffic(self.snapshot, changed)
|
|
self.report['pan'][0] = 20
|
|
with self.assertRaises(ValueError):
|
|
check.compare_traffic(self.snapshot, self.report)
|
|
|
|
def test_missing_queue_overflow_rejected(self):
|
|
saved = json.loads((check.ROOT / 'tests/fixtures/p23-03/queued_elevator.json').read_text())['saved']
|
|
self.runtime['elevators'] = saved
|
|
saved['trips'] = [dict(copy.deepcopy(saved['trips'][0]), actor_id='guest:g'+str(i),
|
|
request_id='trip'+str(i), enqueued_at=i) for i in range(3)]
|
|
self.report.update(queue_positions={'guest/g0': [100, 82]},
|
|
queue_hidden={'guest/g1': True, 'guest/g2': True}, landing_badges=[dict(at=[-4, 54],
|
|
text='待ち3人（他2人）', cell=[0, 0, 0],
|
|
members=['guest/g0', 'guest/g1', 'guest/g2'], visible_count=1, overflow=2)])
|
|
check.compare_queues(self.snapshot, self.report)
|
|
self.report['landing_badges'][0]['overflow'] = 0
|
|
with self.assertRaises(ValueError):
|
|
check.compare_queues(self.snapshot, self.report)
|
|
|
|
def test_fictional_goods_and_rooms_rejected(self):
|
|
check.compare_goods(self.snapshot, self.report)
|
|
check.compare_room_services(self.snapshot, self.report)
|
|
self.report['goods']['requests'].append(dict(stock=100))
|
|
with self.assertRaises(ValueError):
|
|
check.compare_goods(self.snapshot, self.report)
|
|
self.report['rooms']['invented'] = dict(count=0)
|
|
with self.assertRaises(ValueError):
|
|
check.compare_room_services(self.snapshot, self.report)
|
|
|
|
def test_room_reason_and_rescue_occupancy(self):
|
|
room = dict(instance_id='clinic', origin=[0, 0, 0], definition_id='infirmary',
|
|
state='NO_POWER', state_reasons=['NO_POWER'], suspension_reason='',
|
|
customer_capacity=2, active_staff_ids=[], active_services=[], occupied_cells=[])
|
|
self.runtime['facilities']['instances'] = [room]
|
|
self.runtime['staff']['tasks'] = [dict(task_id='rescue', task_type='rescue_transport',
|
|
state='WORKING', assigned_staff_id='carrier', target_entity_id='patient', rescue_facility_id='clinic')]
|
|
self.report['treated_patients'] = {'patient': 'clinic'}
|
|
self.report['rooms'] = {'clinic': dict(state=2, state_name='NO_POWER', reasons=['NO_POWER'],
|
|
operation=0, people=['patient'], count=1, limit=2, staff=[], suspension='')}
|
|
check.compare_room_services(self.snapshot, self.report)
|
|
for field, wrong in [('state', 6), ('operation', 1), ('people', []), ('count', 0)]:
|
|
changed = copy.deepcopy(self.report)
|
|
changed['rooms']['clinic'][field] = wrong
|
|
with self.assertRaises(ValueError):
|
|
check.compare_room_services(self.snapshot, changed)
|
|
|
|
def test_cabin_geometry_and_phase(self):
|
|
shaft = dict(config=dict(shaft_id='lift', stops=[[0, 0, 0]], width=2, capacity=4),
|
|
current_floor=0, phase='IDLE', operational=True, boarded=[])
|
|
self.runtime['elevators']['shafts'] = [shaft]
|
|
self.report.update(paused=True, animation_time=0, elevator_tracks={'lift': dict(floor=0, at=0)},
|
|
cabins=[dict(shaft_id='lift', rect=[80, 72, 24, 19], pixel=1, operational=True,
|
|
doors_open=True, badge_at=[80, 58], badge='0/4人', passengers=[])])
|
|
check.compare_cabin_geometry(self.snapshot, self.report)
|
|
for field, value in [('rect', [80, 70, 24, 19]), ('doors_open', False), ('badge', '4/4人')]:
|
|
changed = copy.deepcopy(self.report)
|
|
changed['cabins'][0][field] = value
|
|
with self.assertRaises(ValueError):
|
|
check.compare_cabin_geometry(self.snapshot, changed)
|
|
|
|
|
|
class AnchoredGeometryTests(unittest.TestCase):
|
|
"""Whole state classification fixtures, with independently chosen draw rows."""
|
|
def setUp(self):
|
|
SpatialTests.setUp(self)
|
|
self.runtime['world'].update(bounds=dict(min_x=0, max_x=3, min_y=0, max_y=1),
|
|
walkable_cells=[[x, y, 0] for y in range(2) for x in range(4)])
|
|
self.report.update(viewport=[408, 332], paused=True, animation_time=0,
|
|
actor_tracks={}, boarded={}, actor_aggregations=[], badges=[], traffic=[])
|
|
self.runtime.update(defense=dict(raids=[]), events=dict(actor_locks={}))
|
|
|
|
def person(self, key, cell, state='MOVING', **fields):
|
|
kind, identifier = key.split('/')
|
|
if kind == 'guest':
|
|
person = dict(guest_id=identifier, current_cell=cell, state=state,
|
|
current_service_id='', target_facility_id='')
|
|
self.runtime['guests']['guests'].append(person)
|
|
else:
|
|
person = dict(staff_id=identifier, current_cell=cell, state=state, health=100,
|
|
assigned_facility_id='', current_task_id='', carried_inventory_id='', role_id='guard')
|
|
self.runtime['staff']['staff'].append(person)
|
|
person.update(fields)
|
|
self.report['actor_tracks'][key] = dict(cell=cell[:], **{'from': cell[:]}, at=0, direction=1)
|
|
return person
|
|
|
|
def drawing(self, key, center, mode='idle', scale=3, receiving=False, riding=False, at=None):
|
|
kind, identifier = key.split('/')
|
|
source = self.runtime['staff']['staff'] if kind == 'staff' else self.runtime['guests']['guests']
|
|
person = next(r for r in source if r[kind+'_id'] == identifier)
|
|
hashed = 5381
|
|
for char in key:
|
|
hashed = (hashed*33+ord(char)) & 0xffffffff
|
|
row = dict(key=key, center=center, at=at or [center[0]-8*scale, center[1]+36-17*scale],
|
|
scale=scale, mode=mode, frame=hashed%15, variant=hashed%15, flip=False,
|
|
staff=kind == 'staff', state=check.enum_names(check.ROOT,
|
|
'scripts/'+kind+'/'+kind+'_runtime.gd', 'State').index(person['state']),
|
|
cell=person['current_cell'][:], receiving=receiving, carrying=False, riding=riding)
|
|
if not riding:
|
|
self.report['actor_drawings'].append(row)
|
|
return row
|
|
|
|
def room(self, fid, definition, cells, users=(), workers=()):
|
|
facility = dict(instance_id=fid, definition_id=definition, origin=cells[0], occupied_cells=cells,
|
|
state='OPEN', state_reasons=[], suspension_reason='', customer_capacity=10,
|
|
active_services=[dict(guest_ids=list(users))] if users else [], active_staff_ids=list(workers), queue_entries=[])
|
|
self.runtime['facilities']['instances'].append(facility)
|
|
self.runtime['world']['occupancies'].append(dict(entity_id=fid, cells=cells))
|
|
self.report['rooms'][fid] = dict(people=list(users), count=len(users), staff=list(workers),
|
|
state=check.enum_names(check.ROOT, 'scripts/facility/facility_instance.gd', 'State').index('OPEN'),
|
|
state_name='OPEN', reasons=[], operation=1, limit=10, suspension='', definition_id=definition,
|
|
rect=check.Transform(self.runtime, self.report).room(cells))
|
|
return facility
|
|
|
|
def queue_work(self):
|
|
facility = self.room('work', 'cafeteria', [[x,y,0] for y in (0,1) for x in (2,3)], workers=['w'])
|
|
for key in ('a', 'b'):
|
|
self.person('guest/'+key, [1,1,0], 'QUEUING', current_service_id='service', target_facility_id='work')
|
|
# Queue takes precedence even over a nonempty current_service_id.
|
|
facility['active_services'] = [dict(guest_ids=['a','b'])]
|
|
self.report['rooms']['work'].update(people=['a','b'], count=2)
|
|
facility['queue_entries'] = ['request']
|
|
self.runtime['facilities']['requests'] = [dict(request_id='request', guest_ids=['a','b'])]
|
|
self.person('staff/w', [2,1,0], 'WORKING', assigned_facility_id='work')
|
|
self.report['queue_positions'] = {'guest/a':[172,118], 'guest/b':[108,118]}
|
|
self.report['landing_badges'] = [dict(at=[140,54], text='待ち2人', cell=[1,1,0],
|
|
members=['guest/a','guest/b'], visible_count=2, overflow=0)]
|
|
self.drawing('guest/a', [172,118], 'eat', receiving=True)
|
|
self.drawing('guest/b', [108,118], 'eat', receiving=True)
|
|
self.drawing('staff/w', [236,126], 'work')
|
|
|
|
def test_queue_lane_and_work_staff(self):
|
|
self.queue_work()
|
|
check.compare_actor_geometry(self.snapshot, self.report)
|
|
|
|
def test_queue_precedence_order_and_category_forgeries(self):
|
|
self.queue_work()
|
|
for mutation in ('precedence', 'order', 'category', 'feet', 'pose'):
|
|
report = copy.deepcopy(self.report)
|
|
if mutation == 'precedence': report['actor_drawings'][0].update(center=[268,210], at=[244,195])
|
|
if mutation == 'order':
|
|
report['queue_positions']['guest/a'], report['queue_positions']['guest/b'] = report['queue_positions']['guest/b'], report['queue_positions']['guest/a']
|
|
if mutation == 'category': report['actor_drawings'][0].update(staff=True, region='work', in_transit=True)
|
|
if mutation == 'feet': report['actor_drawings'][0]['at'][1] -= 17
|
|
if mutation == 'pose': report['actor_drawings'][0]['mode'] = 'wait'
|
|
with self.subTest(mutation=mutation), self.assertRaises(ValueError):
|
|
check.compare_actor_geometry(self.snapshot, report)
|
|
|
|
def test_receiving_r13_guest_cannot_be_forced_into_access_cell(self):
|
|
fixture = json.loads((check.ROOT/'tests/fixtures/p23-03/r13_service_region_gap.json').read_text())
|
|
guest = fixture['guests'][0]
|
|
self.runtime['world']['bounds'].update(max_x=9)
|
|
self.report.update(viewport=[792,332], pan=[567,318])
|
|
fid = guest['target']
|
|
self.room(fid, 'cafeteria', [[x,y,0] for y in (0,1) for x in range(6,10)], users=[guest['id']])
|
|
self.person('guest/'+guest['id'], guest['cell'], 'RECEIVING_SERVICE', current_service_id='service', target_facility_id=fid)
|
|
# Exact historical room x1027..1283; access cell x963..1027.
|
|
self.assertEqual(self.report['rooms'][fid]['rect'], fixture['room_rects'][fid])
|
|
row = self.drawing('guest/'+guest['id'], [1069+2/3,528], 'eat', receiving=True, at=[1046,513])
|
|
check.compare_actor_geometry(self.snapshot, self.report)
|
|
row.update(center=[995,532], at=[971,517], region='floor')
|
|
with self.assertRaises(ValueError): check.compare_actor_geometry(self.snapshot, self.report)
|
|
|
|
def test_carried_patient_and_treated_staff(self):
|
|
self.person('staff/carrier', [0,0,0], 'AVAILABLE')
|
|
self.person('staff/patient', [0,0,0], 'INCAPACITATED')
|
|
self.person('staff/treated', [2,0,0], 'INCAPACITATED')
|
|
self.room('clinic', 'infirmary', [[2,0,0]])
|
|
self.runtime['staff']['tasks'] = [dict(task_type='rescue_transport', state=phase,
|
|
assigned_staff_id='carrier', target_entity_id=patient, rescue_facility_id='clinic')
|
|
for patient, phase in [('patient','MOVING'), ('treated','WORKING')]]
|
|
self.report.update(carried_patients={'patient':'carrier'}, treated_patients={'treated':'clinic'})
|
|
self.report['rooms']['clinic'].update(people=['treated'], count=1)
|
|
self.drawing('staff/carrier', [108,214])
|
|
self.drawing('staff/patient', [108,166], 'collapsed')
|
|
row = self.drawing('staff/treated', [236,190], 'collapsed')
|
|
check.compare_actor_geometry(self.snapshot, self.report)
|
|
self.report['actor_drawings'][1]['center'][1] += 4
|
|
self.report['actor_drawings'][1]['at'][1] += 4
|
|
with self.assertRaises(ValueError): check.compare_actor_geometry(self.snapshot, self.report)
|
|
|
|
def cabin(self, count):
|
|
shaft = json.loads((check.ROOT/'tests/fixtures/p23-03/queued_elevator.json').read_text())['saved']['shafts'][0]
|
|
shaft.update(current_floor=0, phase='IDLE', boarded=['guest:p%02d'%i for i in range(count)])
|
|
# Synthetic saved configuration, not a production definition or capacity edit.
|
|
shaft['config']['capacity'] = max(4, count)
|
|
self.runtime['elevators']['shafts'] = [shaft]
|
|
sid = shaft['config']['shaft_id']
|
|
self.report['elevator_tracks'] = {sid: dict(floor=0, **{'from':0}, at=0)}
|
|
cabin = dict(shaft_id=sid, rect=[92,174,96,76], pixel=4, operational=True, doors_open=True,
|
|
badge_at=[92,160], badge=('満員 ' if count >= 4 else '')+f'{count}/{max(4,count)}人', passengers=[])
|
|
self.report['cabins'] = [cabin]
|
|
for i in range(count):
|
|
key = 'guest/p%02d'%i
|
|
self.person(key, [0,0,0])
|
|
self.report['boarded'][key] = True
|
|
cabin['passengers'].append(self.drawing(key, [116+32*(i%2),206-36*((i//2)%2)], scale=2, riding=True))
|
|
return cabin
|
|
|
|
def occupied_rest(self):
|
|
state, _, audit = RestAuditTests().fixture()
|
|
old = state['systems']['runtime']
|
|
self.runtime['path'] = old['path']
|
|
self.person('staff/s', [0,0,0], 'OFF_DUTY')
|
|
self.room('a', 'staff_waiting_room', [[0,0,0]])
|
|
self.room('b', 'staff_waiting_room', [[0,0,0]])
|
|
for facility in self.runtime['facilities']['instances']:
|
|
facility['occupied_cells'] = [] # Public selector falls back to origin.
|
|
self.report['resting_staff'] = {'s':'a'}
|
|
self.report['rooms']['a'].update(people=['s'], count=1)
|
|
self.drawing('staff/s', [108,198])
|
|
return audit
|
|
|
|
def test_compare_rest_with_actual_occupied_facility(self):
|
|
audit = self.occupied_rest()
|
|
self.assertTrue(check.compare_rest(self.snapshot, self.report, audit)['s']['resting'])
|
|
self.report['actor_drawings'][0]['at'][1] += 1
|
|
with self.assertRaises(ValueError): check.compare_rest(self.snapshot, self.report, audit)
|
|
|
|
def test_compare_rest_independently_proves_no_fit(self):
|
|
audit = self.occupied_rest()
|
|
rest = self.report['actor_drawings'].pop()
|
|
rest.update(scale=1, at=[100,217])
|
|
order = [6,2,9,4,0,7,3,1,8,5]
|
|
identities = {}
|
|
for i in range(10):
|
|
key = 'guest/'+str(i)
|
|
self.person(key, [0,0,0])
|
|
center = [108+[0,16,-16,8,-8,24,-24][i%7],214-4*(i%2)]
|
|
row = self.drawing(key, center, scale=1, at=[76+6*order.index(i),center[1]+19])
|
|
identities[i] = dict(key=key, id=str(i), state=3, cell=[0,0,0], mode='idle')
|
|
self.report['actor_drawings'] = [r for r in self.report['actor_drawings'] if r['key'] != 'guest/5']+[rest]
|
|
self.report['actor_aggregations'] = [dict(members=[identities[i] for i in order],
|
|
drawn_keys=['guest/'+str(i) for i in order[:9]], at=[80,170], text='+1')]
|
|
check.compare_rest(self.snapshot, self.report, audit)
|
|
stale = copy.deepcopy(self.report)
|
|
stale['actor_aggregations'][0]['members'].append(dict(key='staff/s', id='s', state=0, cell=[0,0,0], mode='idle'))
|
|
with self.assertRaises(ValueError): check.compare_rest(self.snapshot, stale, audit)
|
|
self.report['actor_aggregations'][0]['members'].pop()
|
|
with self.assertRaises(ValueError): check.compare_rest(self.snapshot, self.report, audit)
|
|
|
|
def test_compare_rest_with_queued_staff(self):
|
|
audit = self.occupied_rest()
|
|
self.person('staff/q', [0,0,0], 'OFF_DUTY')
|
|
saved = json.loads((check.ROOT/'tests/fixtures/p23-03/queued_elevator.json').read_text())['saved']
|
|
saved['trips'][0]['actor_id'] = 'staff:q'
|
|
self.runtime['elevators'] = saved
|
|
self.runtime['world']['walkable_cells'].append([-1,0,0])
|
|
self.report['queue_positions'] = {'staff/q':[44,214]}
|
|
self.report['landing_badges'] = [dict(at=[-4,150], text='待ち1人', cell=[0,0,0],
|
|
members=['staff/q'], visible_count=1, overflow=0)]
|
|
queued = copy.deepcopy(audit['staff'][0])
|
|
queued.update(staff_id='q', in_transit=True)
|
|
queued['comparison_only'].update(resting=False, label='休憩先へ移動中')
|
|
audit['staff'].append(queued)
|
|
self.drawing('staff/q', [44,214], 'wait')
|
|
result = check.compare_rest(self.snapshot, self.report, audit)
|
|
self.assertTrue(result['s']['resting'])
|
|
self.assertFalse(result['q']['resting'])
|
|
self.report['actor_drawings'][-1].update(center=[108,198], at=[84,183])
|
|
with self.assertRaises(ValueError): check.compare_rest(self.snapshot, self.report, audit)
|
|
|
|
def test_passenger_and_floor_actor_front_and_back(self):
|
|
cabin = self.cabin(1)
|
|
self.person('guest/floor', [0,0,0])
|
|
self.drawing('guest/floor', [108,214])
|
|
cabin['passengers'][0]['at'][0] = 108
|
|
for layer in (0,1):
|
|
self.report['layer'] = layer
|
|
for guest in self.runtime['guests']['guests']: guest['current_cell'][2] = layer
|
|
for cell in self.runtime['elevators']['shafts'][0]['config']['stops']: cell[2] = layer
|
|
for row in self.report['actor_drawings']+cabin['passengers']: row['cell'][2] = layer
|
|
for track in self.report['actor_tracks'].values():
|
|
track['cell'][2] = track['from'][2] = layer
|
|
with self.subTest(layer=layer):
|
|
check.compare_cabin_geometry(self.snapshot, self.report)
|
|
check.compare_actor_geometry(self.snapshot, self.report)
|
|
|
|
def test_cabin_aggregation_retains_full_boarded_membership(self):
|
|
cabin = self.cabin(40)
|
|
originals = copy.deepcopy(cabin['passengers'])
|
|
retained = {}
|
|
for y in (170,206):
|
|
group = sorted([r for r in originals if r['center'][1] == y], key=lambda r:(r['center'][0],r['key']))
|
|
drawn = group[:14] # 88px holds fourteen 6px cores, never fifteen.
|
|
for i, row in enumerate(drawn):
|
|
row.update(scale=1, at=[94+6*i,y+19])
|
|
retained[row['key']] = row
|
|
self.report['actor_aggregations'].append(dict(
|
|
members=[dict(key=r['key'], id=r['key'][6:], state=r['state'], cell=r['cell'], mode=r['mode']) for r in group],
|
|
drawn_keys=[r['key'] for r in drawn], at=[96,178], text='+6'))
|
|
cabin['passengers'] = [retained[r['key']] for r in originals if r['key'] in retained]
|
|
check.compare_cabin_geometry(self.snapshot, self.report)
|
|
check.compare_actor_geometry(self.snapshot, self.report)
|
|
self.assertEqual(sum(len(a['members']) for a in self.report['actor_aggregations']), 40)
|
|
for mutation in ('lost_identity','badge'):
|
|
report = copy.deepcopy(self.report)
|
|
if mutation == 'lost_identity': report['actor_aggregations'][0]['members'].pop()
|
|
else: report['actor_aggregations'][0]['at'][0] += 1
|
|
with self.subTest(mutation=mutation), self.assertRaises(ValueError):
|
|
check.compare_actor_geometry(self.snapshot, report)
|
|
|
|
|
|
class ComparisonTests(unittest.TestCase):
|
|
def pair(self):
|
|
declaration = dict(metric='sum_destination_arrival_minutes_entire_cohort',
|
|
arrivals=[dict(id='g', offset=0)], window=2, to=[2, 1, 0],
|
|
improvement=dict(definition_id='guest_elevator', origin=[4, 0, 0]))
|
|
a = dict(declaration=declaration, initial=dict(seed=1))
|
|
a['arrival-g'] = dict(offset=0, payload=dict(guest_id='g'), result=dict(code='OK'))
|
|
for minute in range(3):
|
|
a['tick-%03d' % minute] = dict(minute=minute, guests=dict(guests=[dict(guest_id='g',
|
|
current_cell=[2, 1, 0] if minute == 2 else [2, 0, 0])]))
|
|
b = copy.deepcopy(a)
|
|
b['build-0'] = dict(payload=declaration['improvement'], result=dict(code='OK'))
|
|
b['tick-001']['guests']['guests'][0]['current_cell'] = [2, 1, 0]
|
|
return a, b
|
|
|
|
def test_strict_metric_from_typed_cells(self):
|
|
a, b = self.pair()
|
|
self.assertEqual(check.compare_layouts(a, b)['B'], 1)
|
|
for mutation in ('demand', 'incomplete', 'no_improvement', 'extra_build'):
|
|
with self.subTest(mutation=mutation):
|
|
a, b = self.pair()
|
|
if mutation == 'demand':
|
|
b['arrival-g']['payload']['guest_id'] = 'other'
|
|
elif mutation == 'incomplete':
|
|
for minute in (1, 2):
|
|
b['tick-%03d' % minute]['guests']['guests'][0]['current_cell'] = [2, 0, 0]
|
|
elif mutation == 'no_improvement':
|
|
b['tick-001'] = copy.deepcopy(a['tick-001'])
|
|
else:
|
|
b['build-1'] = copy.deepcopy(b['build-0'])
|
|
with self.assertRaises(ValueError):
|
|
check.compare_layouts(a, b)
|
|
|
|
|
|
class OfflineGeometryTests(unittest.TestCase):
|
|
"""Synthetic calibration examples; no OS or acceptance credit."""
|
|
|
|
def setUp(self):
|
|
self.calibration = dict(window_size=[100, 80], window_position=[240, 150],
|
|
screenshot_pixels=[200, 190], backing_scale=2,
|
|
canvas_stretch=1, titlebar_pixels=30)
|
|
self.request = dict(window_size=[100, 80], window_position=[240, 150],
|
|
screen_scale=2, content_scale_factor=1, target_pid=123)
|
|
self.clicks = [dict(target_pid=123, events=[dict(action='click',
|
|
coordinates=[20, 70], calibration=copy.deepcopy(self.calibration),
|
|
point=dict(screen=[10, 20], viewport=[10, 20],
|
|
viewport_to_screen=[1, 0, 0, 1, 0, 0]))])]
|
|
|
|
def transform(self):
|
|
return check.screenshot_transform(self.calibration, self.request, [200, 190], self.clicks)
|
|
|
|
def test_uses_measured_scale_origin_not_desktop_position(self):
|
|
self.assertEqual(check.affine_rect(self.transform(), [10, 20, 5, 7]), [20, 70, 10, 14])
|
|
|
|
def test_shifted_titlebar_fails_independent_click(self):
|
|
self.calibration['titlebar_pixels'] += 1
|
|
self.calibration['screenshot_pixels'][1] += 1
|
|
self.clicks[0]['events'][0]['calibration'] = copy.deepcopy(self.calibration)
|
|
with self.assertRaises(ValueError):
|
|
check.screenshot_transform(self.calibration, self.request, [200, 191], self.clicks)
|
|
|
|
def test_wrong_click_offset_fails(self):
|
|
self.clicks[0]['events'][0]['coordinates'][1] -= 30
|
|
with self.assertRaisesRegex(ValueError, 'click screenshot offset'):
|
|
self.transform()
|
|
|
|
def test_wrong_pid_cannot_calibrate(self):
|
|
self.clicks[0]['target_pid'] += 1
|
|
with self.assertRaisesRegex(ValueError, 'missing independent click'):
|
|
self.transform()
|
|
|
|
def test_nonfinite_and_false_scale_rejected(self):
|
|
for scale in (False, float('nan'), 0, -1):
|
|
self.calibration['backing_scale'] = scale
|
|
with self.assertRaisesRegex(ValueError, 'invalid screenshot calibration'):
|
|
self.transform()
|
|
|
|
def test_all_visibility_cases_are_accounted_for(self):
|
|
self.assertEqual(check.rectangle_eligibility([2, 3, 4, 5], [10, 10]),
|
|
('visible', [2, 3, 4, 5]))
|
|
self.assertEqual(check.rectangle_eligibility([-2, 3, 4, 5], [10, 10]),
|
|
('clipped', [0, 3, 2, 5]))
|
|
self.assertEqual(check.rectangle_eligibility([10, 3, 4, 5], [10, 10])[0], 'outside')
|
|
self.assertEqual(check.rectangle_eligibility([2, 3, 0, 5], [10, 10])[0], 'outside')
|
|
|
|
def test_invalid_rectangles_rejected(self):
|
|
for rect in ([0, 0, -1, 4], [float('nan'), 0, 1, 4]):
|
|
with self.assertRaisesRegex(ValueError, 'invalid clipping rectangle'):
|
|
check.rectangle_eligibility(rect, [10, 10])
|
|
|
|
def test_noise_is_measured_and_does_not_assert_class(self):
|
|
before = (2, 2, [[(10, 20, 30)]*2 for _ in range(2)])
|
|
after = copy.deepcopy(before)
|
|
after[2][1][1] = (10, 27, 30)
|
|
noise = check.paired_crop_noise(before, after, [0, 0, 2, 2])
|
|
self.assertEqual(noise, dict(samples=4, max_channel_delta=7,
|
|
mean_channel_delta=1.75, unchanged=3))
|
|
self.assertNotIn('pass', noise)
|
|
self.assertNotIn('state', noise)
|
|
|
|
def test_wrong_dimensions_and_empty_crop_rejected(self):
|
|
image = (1, 1, [[(0, 0, 0)]])
|
|
with self.assertRaisesRegex(ValueError, 'paired image dimensions'):
|
|
check.paired_crop_noise(image, (2, 1, []), [0, 0, 1, 1])
|
|
with self.assertRaisesRegex(ValueError, 'empty noise crop'):
|
|
check.paired_crop_noise(image, image, [1, 0, 1, 1])
|
|
|
|
def test_report_positions_and_source_provenance_are_recomputed(self):
|
|
source = 'scripts/ui/pixel_people.gd'
|
|
report = dict(rooms={}, landing_badges=[], cabins=[], resting_staff={}, badges=[],
|
|
actor_drawings=[dict(key='staff/s', staff=True, at=[1, 1], scale=1,
|
|
mode='idle', frame=0, flip=False)])
|
|
colors = list(dict.fromkeys(check.re.findall(r'Color\("([0-9a-fA-F]{6})"',
|
|
(check.ROOT/source).read_text())))
|
|
audit = dict(rows=[dict(category='staff', id='staff/s', rect=[1, 1, 16, 18],
|
|
style_source=source, style_colors=colors)])
|
|
check.validate_drawing_rows(report, audit)
|
|
for change in ('shift', 'identity', 'missing', 'palette'):
|
|
tampered = copy.deepcopy(audit)
|
|
if change == 'shift': tampered['rows'][0]['rect'][0] += 1
|
|
if change == 'identity': tampered['rows'][0]['category'] = 'guest'
|
|
if change == 'missing': tampered['rows'] = []
|
|
if change == 'palette': tampered['rows'][0]['style_colors'] = ['000000']
|
|
with self.subTest(change=change), self.assertRaises(ValueError):
|
|
check.validate_drawing_rows(report, tampered)
|
|
|
|
|
|
class FrameReviewTests(unittest.TestCase):
|
|
def test_frame_raw_context_valid_malformed_and_missing_rest(self):
|
|
fixture = SpatialTests()
|
|
fixture.setUp()
|
|
fixture.moving_four()
|
|
fixture.report.update(viewport=[216,236], traffic=[])
|
|
audit = fixture.drawing_audit()
|
|
for row in audit['rows']:
|
|
row.update(viewport_rect=row['rect'][:], screen_rect=row['rect'][:])
|
|
for context, rest, passes in ((fixture.snapshot,None,True), ({},None,False)):
|
|
result = check.frame_rows(fixture.report, audit, self.request, dict(calibration=self.calibration),
|
|
[self.image], self.clicks, state=context, rest_audit=rest)
|
|
self.assertEqual(not result['errors'], passes)
|
|
fixture.report['resting_staff'] = {'s':'a'}
|
|
for rest in (None, [], {}):
|
|
result = check.frame_rows(fixture.report, audit, self.request, dict(calibration=self.calibration),
|
|
[self.image], self.clicks, state=fixture.snapshot, rest_audit=rest)
|
|
self.assertFalse(result['mechanical_pass'])
|
|
fixture.report['resting_staff'] = {}
|
|
result = check.frame_rows(fixture.report, audit, self.request, dict(calibration=self.calibration),
|
|
[self.image], self.clicks)
|
|
self.assertFalse(result['mechanical_pass'])
|
|
self.assertIn('core overlap', str(result['errors']))
|
|
|
|
def test_frame_with_valid_occupied_rest_authority(self):
|
|
fixture = AnchoredGeometryTests()
|
|
fixture.setUp()
|
|
rest = fixture.occupied_rest()
|
|
rows = []
|
|
for category, identifier, rect, source, text in check.drawing_rectangle_rows(fixture.report):
|
|
self.assertIsNone(text)
|
|
colors = list(dict.fromkeys(check.re.findall(r'Color\("([0-9a-fA-F]{6})"', (check.ROOT/source).read_text())))
|
|
rows.append(dict(category=category, id=identifier, rect=rect, viewport_rect=rect[:],
|
|
screen_rect=rect[:], style_source=source, style_colors=colors))
|
|
result = check.frame_rows(fixture.report, dict(rows=rows), self.request, dict(calibration=self.calibration),
|
|
[self.image], self.clicks, state=fixture.snapshot, rest_audit=rest)
|
|
self.assertEqual(result['errors'], [])
|
|
rest['path_service_sha256'] = '0'*64
|
|
result = check.frame_rows(fixture.report, dict(rows=rows), self.request, dict(calibration=self.calibration),
|
|
[self.image], self.clicks, state=fixture.snapshot, rest_audit=rest)
|
|
self.assertFalse(result['mechanical_pass'])
|
|
self.assertIn('rest PathService source pin', result['errors'])
|
|
|
|
def test_os_offline_raw_context_binding_and_hashes(self):
|
|
def typed(value):
|
|
if isinstance(value, dict): return dict(type='Dictionary', entries=[[typed(k),typed(v)] for k,v in value.items()])
|
|
if isinstance(value, list): return dict(type='Array', items=[typed(v) for v in value])
|
|
return dict(type={str:'String',int:'int'}[type(value)], text=json.dumps(value))
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
stream = Path(temporary)
|
|
(stream/'objects').mkdir()
|
|
def put(value):
|
|
raw = json.dumps(typed(value)).encode()
|
|
digest = hashlib.sha256(raw).hexdigest()
|
|
(stream/'objects'/(digest+'.json')).write_bytes(raw)
|
|
return digest
|
|
report = {'actor_drawings':[]}
|
|
state = {'systems':{'runtime':{}}}
|
|
rest = {'staff':[], 'path_service_sha256':check.sha(check.ROOT/'scripts/world/path_service.gd')}
|
|
capture = dict(report=put(report), state=put(state), rest_audit=put(rest))
|
|
self.assertEqual(check.drawing_context(stream, capture, report), (state,rest))
|
|
for field in ('state','rest_audit','report'):
|
|
missing = dict(capture)
|
|
del missing[field]
|
|
with self.assertRaises(ValueError): check.drawing_context(stream, missing, report)
|
|
for field in ('state','rest_audit'):
|
|
bad = dict(capture, **{field:put({})})
|
|
with self.assertRaises(ValueError): check.drawing_context(stream, bad, report)
|
|
with self.assertRaises(ValueError): check.drawing_context(stream, capture, {'wrong':1})
|
|
with self.assertRaises(FileNotFoundError):
|
|
check.drawing_context(stream, dict(capture, state='0'*64), report)
|
|
(stream/'objects'/(capture['state']+'.json')).write_text('{}')
|
|
with self.assertRaisesRegex(ValueError, 'object hash'): check.drawing_context(stream, capture, report)
|
|
|
|
def setUp(self):
|
|
OfflineGeometryTests.setUp(self)
|
|
self.request.update(local_to_viewport=[1, 0, 0, 1, 0, 0],
|
|
viewport_to_screen=[1, 0, 0, 1, 0, 0],
|
|
local_to_screen=[1, 0, 0, 1, 0, 0])
|
|
self.report = dict(layer=0, viewport=[100, 80], rooms={}, cabins=[],
|
|
landing_badges=[], badges=[], resting_staff={}, traffic=[],
|
|
actor_drawings=[dict(key='staff/s', staff=True, at=[1, 1], scale=1, state=0,
|
|
mode='idle', frame=0, flip=False)])
|
|
source = 'scripts/ui/pixel_people.gd'
|
|
colors = list(dict.fromkeys(check.re.findall(r'Color\("([0-9a-fA-F]{6})"',
|
|
(check.ROOT/source).read_text())))
|
|
self.audit = dict(rows=[dict(category='staff', id='staff/s', rect=[1, 1, 16, 18],
|
|
viewport_rect=[1, 1, 16, 18], screen_rect=[1, 1, 16, 18],
|
|
style_source=source, style_colors=colors)])
|
|
self.image = (200, 190, [[(x % 256, y % 256, 0) for x in range(200)] for y in range(190)])
|
|
|
|
def frame(self):
|
|
return check.frame_rows(self.report, self.audit, self.request,
|
|
dict(calibration=self.calibration), [self.image], self.clicks)
|
|
|
|
def test_frame_without_raw_state_explicitly_denies_transit_exemptions(self):
|
|
self.report['actor_drawings'][0]['in_transit'] = True
|
|
self.assertEqual(self.frame()['transit_authority'],
|
|
'absent: fail closed, no transit exemptions')
|
|
|
|
def test_blank_crop_fails_without_threshold(self):
|
|
self.image = (200, 190, [[(0, 0, 0)] * 200 for _ in range(190)])
|
|
result = self.frame()
|
|
self.assertFalse(result['mechanical_pass'])
|
|
self.assertIn('blank crop', result['rows'][0]['failures'])
|
|
|
|
def test_missing_extra_and_unaudited_traffic_fail(self):
|
|
for change in ('missing', 'extra', 'traffic'):
|
|
with self.subTest(change=change):
|
|
original = copy.deepcopy(self.audit)
|
|
if change == 'missing': self.audit['rows'].clear()
|
|
if change == 'extra': self.audit['rows'] *= 2
|
|
if change == 'traffic':
|
|
self.report['traffic'] = [dict(cell=[-6, 2, 0], rect=[1, 1, 5, 5],
|
|
traffic=dict(current_density=1, temporary_block=False))]
|
|
result = self.frame()
|
|
self.assertFalse(result['mechanical_pass'])
|
|
self.assertEqual(len(result['rows']), len(self.audit['rows']))
|
|
self.assertIn('offline rectangle coverage', result['errors'])
|
|
self.audit = original
|
|
|
|
def test_dense_or_blocked_traffic_row_closes_audit_gap(self):
|
|
source = 'scenes/main/world_preview.gd'
|
|
colors = list(dict.fromkeys(check.re.findall(r'Color\("([0-9a-fA-F]{6})"',
|
|
(check.ROOT/source).read_text())))
|
|
for density, blocked in ((1, False), (0, True), (0, False)):
|
|
with self.subTest(density=density, blocked=blocked):
|
|
self.report['traffic'] = [dict(cell=[-6, 2, 0], rect=[1, 1, 5, 5],
|
|
traffic=dict(current_density=density, temporary_block=blocked))]
|
|
active = density > 0 or blocked
|
|
missing = self.frame()
|
|
self.assertEqual(missing['mechanical_pass'], not active)
|
|
if active:
|
|
self.assertIn('offline rectangle coverage', missing['errors'])
|
|
self.audit['rows'].append(dict(category='traffic', id='-6,2,0',
|
|
rect=[1, 1, 5, 5], viewport_rect=[1, 1, 5, 5],
|
|
screen_rect=[1, 1, 5, 5], style_source=source, style_colors=colors))
|
|
complete = self.frame()
|
|
self.assertTrue(complete['mechanical_pass'], complete['errors'])
|
|
self.audit['rows'].pop()
|
|
|
|
def test_wrong_state_enum_and_name_fail(self):
|
|
source = 'scripts/ui/pixel_tower_art.gd'
|
|
self.report['actor_drawings'] = []
|
|
room = dict(rect=[1, 1, 16, 18], state=3, state_name='NO_STAFF', operation=0)
|
|
self.report['rooms'] = {'r': room}
|
|
self.audit['rows'][0].update(category='room', id='r', style_source=source,
|
|
style_colors=list(dict.fromkeys(check.re.findall(r'Color\("([0-9a-fA-F]{6})"',
|
|
(check.ROOT/source).read_text()))))
|
|
self.assertEqual(self.frame()['rows'][0]['operation'], 'INACTIVE')
|
|
for state, name in ((99, 'NO_STAFF'), (3, 'OPEN'), (True, 'NO_ACCESS')):
|
|
room.update(state=state, state_name=name)
|
|
with self.assertRaises(ValueError): self.frame()
|
|
|
|
def test_frame_rejects_affine_and_calibration_shift(self):
|
|
self.request['local_to_screen'][4] += 1
|
|
with self.assertRaisesRegex(ValueError, 'affine composition'): self.frame()
|
|
self.request['local_to_screen'][4] -= 1
|
|
self.clicks[0]['events'][0]['coordinates'][1] += 1
|
|
with self.assertRaisesRegex(ValueError, 'click screenshot offset'): self.frame()
|
|
|
|
def test_eligible_crop_fully_outside_image_fails(self):
|
|
for key in ('viewport_to_screen', 'local_to_screen'):
|
|
self.request[key][4] = 300
|
|
self.audit['rows'][0]['screen_rect'][0] += 300
|
|
result = self.frame()
|
|
self.assertFalse(result['mechanical_pass'])
|
|
self.assertIn('eligible crop outside image', result['rows'][0]['failures'])
|
|
|
|
def test_manifest_crop_and_independent_review_binding(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
directory = Path(temporary)
|
|
manifest, digest = check.write_frame_review(directory, self.frame(),
|
|
[dict(whole_image_sha256='a'*64, raw_sha256='b'*64, conversion_sha256='c'*64)])
|
|
self.assertEqual(check.sha(directory/'manifest.json'), digest)
|
|
crop = manifest['rows'][0]['crops'][0]
|
|
self.assertEqual(check.sha(directory/crop['path']), crop['sha256'])
|
|
self.assertEqual(check.png_pixels((directory/crop['path']).read_bytes())[:2], (32, 36))
|
|
verdict_path = directory/'independent-verdict.json'
|
|
self.assertEqual(check.reviewed_pixel_categories(manifest, digest, verdict_path), set())
|
|
verdict = dict(manifest_sha256=digest, reviewer='independent-opus-direct-image',
|
|
rows=[dict(row_index=0, category=True, state=True, position=True)])
|
|
for change in ('wrong_sha', 'missing_row', 'extra_row', 'rejected_state'):
|
|
bad = copy.deepcopy(verdict)
|
|
if change == 'wrong_sha': bad['manifest_sha256'] = '0'*64
|
|
if change == 'missing_row': bad['rows'] = []
|
|
if change == 'extra_row': bad['rows'] *= 2
|
|
if change == 'rejected_state': bad['rows'][0]['state'] = False
|
|
verdict_path.write_text(json.dumps(bad))
|
|
with self.subTest(change=change), self.assertRaises(ValueError):
|
|
check.reviewed_pixel_categories(manifest, digest, verdict_path)
|
|
verdict_path.write_text(json.dumps(verdict))
|
|
self.assertEqual(check.reviewed_pixel_categories(manifest, digest, verdict_path), {'staff'})
|
|
manifest['mechanical_pass'] = False
|
|
self.assertEqual(check.reviewed_pixel_categories(manifest, digest, verdict_path), set())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|