523 lines
31 KiB
Python
523 lines
31 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 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_rectangle_pixel_and_transform_negatives(self):
|
|
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())))
|
|
report = dict(rooms={}, landing_badges=[], cabins=[], resting_staff={}, badges=[],
|
|
actor_drawings=[dict(key='staff/s', staff=True, at=[1, 1], scale=1)])
|
|
row = 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)
|
|
audit = dict(rows=[row])
|
|
matrix = [1, 0, 0, 1, 0, 0]
|
|
image = (24, 24, [[(113, 157, 148)]*24 for _ in range(24)])
|
|
self.assertEqual(check.compare_rectangles(report, audit, matrix, matrix, matrix, image), ['staff/staff/s'])
|
|
for change in ('rectangle', 'pixels', 'transform', 'coverage', 'palette'):
|
|
with self.subTest(change=change):
|
|
a, m, img = copy.deepcopy(audit), matrix[:], image
|
|
if change == 'rectangle': a['rows'][0]['rect'][0] += 1
|
|
if change == 'pixels': img = (24, 24, [[(0, 0, 0)]*24 for _ in range(24)])
|
|
if change == 'transform': m[4] = 10
|
|
if change == 'coverage': a['rows'].clear()
|
|
if change == 'palette': a['rows'][0]['style_colors'] = ['000000']
|
|
with self.assertRaises(ValueError): check.compare_rectangles(report, a, matrix, m, matrix, img)
|
|
|
|
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 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],
|
|
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=[]))
|
|
|
|
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):
|
|
self.runtime['elevators']['trips'] = [dict(actor_id='guest:g'+str(i), enqueued_at=i,
|
|
state='QUEUED', **{'from': [0, 0, 0]}) for i in range(3)]
|
|
self.report.update(queue_positions={'guest/g0': [84, 82], 'guest/g1': [100, 82]},
|
|
queue_hidden={'guest/g2': True}, landing_badges=[dict(at=[76, 54],
|
|
text='待ち3人（他1人）', cell=[0, 0, 0],
|
|
members=['guest/g0', 'guest/g1', 'guest/g2'], visible_count=2, overflow=1)])
|
|
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(shaft_id='lift', config=dict(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 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)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|