P23-03: 生活の観察

tools/test_p23_life_observation_evidence.py

@@ -18,6 +18,54 @@ sys.dont_write_bytecode = True
 import check_p23_life_observation_evidence as check
 
 
+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()
@@ -431,12 +479,14 @@ class SpatialTests(unittest.TestCase):
             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)])
+        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):
@@ -470,7 +520,7 @@ class SpatialTests(unittest.TestCase):
                 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),
+        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)},