P08-03: 配置の通常操作受入

tools/test_p08_normal_placement_evidence.py

@@ -135,6 +135,115 @@ class EvidenceTests(unittest.TestCase):
         self.edit('pair/reader-save-result-0001.json', lambda d: d.update(result=F.typed({'ok':True,'operation':'load'}), requested_path='/outside/formal'))
         self.reject('save outside private user profile')
 
+    def test_notification_payload_is_recomputed_from_returned_dict(self):
+        path = 'pair/reader-checkpoint.json'
+        original = (self.out/path).read_bytes()
+        fields = ('ok','code','message','operation','save_path','reason','request_id',
+            'checkpoint_absolute_minute','duration_us','runtime_duration_us',
+            'selected_candidate_id','selected_candidate_kind','selected_generation',
+            'save_id','saved_absolute_minute')
+        for field in fields:
+            with self.subTest(field=field):
+                data = json.loads(original)
+                event = C.typed_member(data,'systems','runtime','domain_events','recent_events')['items'][-1]
+                value = C.typed_member(event,'payload',field)
+                decoded = C.decode(value)
+                changed = not decoded if isinstance(decoded,bool) else decoded+1 if isinstance(decoded,int) else decoded+'-drift'
+                value.update(C.scalar(changed,value['type']))
+                F.put(self.out/path,data); F.repin(self.out)
+                self.reject('source-defined operation drift')
+        (self.out/path).write_bytes(original); F.repin(self.out)
+
+    def test_notification_envelope_cannot_drift(self):
+        path = 'pair/writer-checkpoint.json'
+        original = (self.out/path).read_bytes()
+        for field in ('event_id','event_type','source_service','absolute_minute','causation_command_id','entity_ids'):
+            with self.subTest(field=field):
+                data = json.loads(original)
+                event = C.typed_member(data,'systems','runtime','domain_events','recent_events')['items'][-1]
+                value = C.typed_member(event,field)
+                if field == 'entity_ids': value['items'].append(C.scalar('unexpected-entity'))
+                else:
+                    old = C.decode(value)
+                    value.update(C.scalar(old+1 if isinstance(old,int) else old+'-drift',value['type']))
+                F.put(self.out/path,data); F.repin(self.out)
+                self.reject('source-defined operation drift')
+        (self.out/path).write_bytes(original); F.repin(self.out)
+
+    def test_notification_counter_must_increment_exactly_once(self):
+        def corrupt(data):
+            C.typed_member(data,'systems','runtime','id_generator','counters','domain_event')['text'] = '9'
+        self.edit('pair/reader-checkpoint.json',corrupt)
+        self.reject('source-defined operation drift')
+
+    def test_notification_published_ids_cannot_be_omitted(self):
+        self.edit('pair/writer-checkpoint.json',lambda d: C.typed_member(d,'systems','runtime','domain_events','published_ids')['items'].clear())
+        self.reject('source-defined operation drift')
+
+    def test_extra_notification_is_rejected(self):
+        def corrupt(data):
+            events = C.typed_member(data,'systems','runtime','domain_events','recent_events')['items']
+            events.append(copy.deepcopy(events[-1]))
+        self.edit('pair/reader-checkpoint.json',corrupt)
+        self.reject('source-defined operation drift')
+
+    def test_existing_event_history_cannot_be_rewritten(self):
+        def corrupt(data):
+            event = C.typed_member(data,'systems','runtime','domain_events','recent_events')['items'][0]
+            C.typed_member(event,'event_type').update(text='"REWRITTEN_HISTORY"')
+        self.edit('pair/reader-checkpoint.json',corrupt)
+        self.reject('source-defined operation drift')
+
+    def test_postsave_unrelated_resource_drift_is_rejected(self):
+        self.edit('pair/writer-checkpoint.json',lambda d: C.typed_member(d,'game_state','resources','gold').update(text='9999'))
+        self.reject('source-defined operation drift')
+
+    def test_postload_unknown_field_drift_is_rejected(self):
+        self.edit('pair/reader-checkpoint.json',lambda d: C.typed_member(d,'systems','runtime','inventory')['entries'].append([C.scalar('unknown'),C.scalar(1)]))
+        self.reject('source-defined operation drift')
+
+    def test_presave_must_match_read_only_formal_model(self):
+        self.edit('pair/writer-presave.json',lambda d: C.typed_member(d,'clock','absolute_minute').update(text='361'))
+        self.reject('wrong presave formal model')
+
+    def test_read_only_inspection_must_match_complete_formal_model(self):
+        self.edit('pair/reader-save-result-0002.json',lambda d: C.typed_member(d['result'],'snapshot','clock','absolute_minute').update(text='361'))
+        self.reject('wrong read-only formal model')
+
+    def test_inspection_cannot_lose_read_only_attribution(self):
+        self.edit('pair/writer-save-result-0002.json',lambda d: C.typed_member(d['result'],'read_only').update(text='false'))
+        self.reject('missing read-only formal model')
+
+    def test_returned_result_payload_tampering_is_rejected(self):
+        self.edit('pair/reader-save-result-0001.json',lambda d: C.typed_member(d['result'],'duration_us').update(text='999'))
+        self.reject('source-defined operation drift')
+
+    def test_notification_payload_type_drift_is_rejected(self):
+        def corrupt(data):
+            event = C.typed_member(data,'systems','runtime','domain_events','recent_events')['items'][-1]
+            C.typed_member(event,'payload','operation')['type'] = 'String'
+            C.typed_member(event,'payload','operation')['text'] = '"recovery"'
+        self.edit('pair/reader-checkpoint.json',corrupt)
+        self.reject('source-defined operation drift')
+
+    def test_operation_count_is_source_derived_with_recent_limit(self):
+        model = C.decode(json.loads((self.out/'pair/writer-formal-model.json').read_text()))
+        runtime = model['systems']['runtime']
+        runtime['id_generator']['counters']['domain_event'] = 900
+        stream = runtime['domain_events']
+        stream['published_ids'] = ['domain_event.%012d' % n for n in range(1,901)]
+        stream['recent_events'] = [{'ordinal':n} for n in range(645,901)]
+        expected = C.decode(C.expected_operation(F.typed(model),{'ok':True},'SAVE_CHECKPOINT_SUCCEEDED'))
+        new = expected['systems']['runtime']
+        self.assertEqual(901,new['id_generator']['counters']['domain_event'])
+        self.assertEqual(901,len(new['domain_events']['published_ids']))
+        self.assertEqual(256,len(new['domain_events']['recent_events']))
+        self.assertEqual(stream['recent_events'][1:],new['domain_events']['recent_events'][:-1])
+        self.assertEqual('domain_event.000000000901',new['domain_events']['recent_events'][-1]['event_id'])
+        del runtime['id_generator']['counters']
+        with self.assertRaisesRegex(ValueError,'missing/duplicate typed model member'):
+            C.expected_operation(F.typed(model),{'ok':True},'SAVE_CHECKPOINT_SUCCEEDED')
+
     def test_missing_formal_member(self):
         for name in ('formal','formal.bak'):
             with self.subTest(package=name):