"""Known denominators and origin declaration boundaries; no all-web inference.""" import copy import itertools import json from pathlib import Path import tempfile import sys import unittest from unittest import mock from types import SimpleNamespace from cryptography.hazmat.primitives.asymmetric import ec from odexa_ref import crypto from profiles import observation_coverage as o, storage_sessions as ss from profiles.gateway_reports import delivery_report from profiles.free_service import later as add_seconds from test_free_contracts import fixtures, uid, NOW, later import test_gateway_reports as gateway_fixture def declaration(pol_bytes,binding,observer,*,kind='agent',bid=None): pol=crypto.strict_json(pol_bytes); origin=pol['origin'] boundary=dict(id=bid or origin+'/observers/'+kind,kind=kind,observer_id=observer,binding=copy.deepcopy(binding), resource_ids=['article'],actions=['retrieve','store'],purposes=['public_retrieval'],starts_at=pol['issued_at'], ends_at=pol['expires_at'],state='instrumented',event_profiles=['core']) return dict(protocol_version=o.VERSION,profile=o.PROFILE,registry_id=origin+o.PATH,origin=origin,revision=1, policy_id=pol['policy_id'],policy_revision=pol['revision'],policy_digest=crypto.digest(pol_bytes), issued_at=pol['issued_at'],expires_at=pol['expires_at'],boundaries=[boundary]) def snapshot(registry,pol_bytes,checked=NOW): wire=crypto.json_bytes(registry); origin=registry['origin'] return o.RegistrySnapshot(wire,pol_bytes,dict(origin=origin,registry_url=origin+o.PATH,policy_url=origin+'/odexa.json', registry_digest=crypto.digest(wire),policy_digest=crypto.digest(pol_bytes),checked_at=checked, tls_verified=True,redirected=False,revalidated=True)) def expected(boundary,event,*,number=20,kind='action',at=NOW,due=None,profile='core',digest=None): subject=dict(kind=kind,id=event['operation']['id'] if kind.startswith(('action','storage_')) else event['http']['delivery_id'] if kind=='delivery' else event['event_id'],reporter_id=event['reporter_id'], ingress_id=event['http']['ingress_id'] if kind=='delivery' else None, checkpoint_index=event['operation']['storage']['checkpoint_index'] if kind=='storage_checkpoint' else None, payload_digest=digest if kind=='report' else None) return dict(id=uid(number),boundary_id=boundary,resource_url=event['resource_url'],action=event['actions'][0], purpose=event['purposes'][0],agreement_id=event['agreement_id'],expected_at=at,due_at=due or later(10),subject=subject,event_profile=profile, basis='gateway_admission' if kind=='delivery' else 'report_outbox' if kind=='report' else 'accepted_duty') class CoverageTests(unittest.TestCase): def setUp(self): self.values,context=fixtures(); self.pol=context[0]; self.event=self.values['report']; self.binding=self.values['binding'] self.r=declaration(self.pol,self.binding,self.event['reporter_id']) self.bid=self.r['boundaries'][0]['id']; self.n=100 self.query=dict(url=self.event['resource_url'],action='store',purpose='public_retrieval',starts_at=NOW,ends_at=later(10)) def intake(self,event=None,*,received=2,payload=None): event=copy.deepcopy(event or self.event); payload=payload or crypto.json_bytes(event); self.n+=1 signed=crypto.sign_jws(payload,ec.derive_private_key(3,ec.SECP256R1()),'https://agent.example/keys/one','odexa-event+jws') return dict(copy.deepcopy(self.values['intake']),record_id=uid(self.n),received_at=later(received), assurance='origin_key_verified' if event['source']=='origin_observed' else 'client_key_verified', authenticated_reporter_id=event['reporter_id'],payload_digest=crypto.digest(payload), payload_b64url=crypto.b64u(payload),reporter_jws=signed) def reduce(self,records=(),expectations=(),*,as_of=None): return o.reduce_coverage(snapshot(self.r,self.pol),query=self.query,verified_intakes=list(records), expectations=list(expectations),as_of=as_of or later(30)) def test_static_publication_has_unknown_traffic_and_null_not_zero_percent(self): self.r['boundaries']=[]; result=self.reduce() self.assertEqual(result['scope_segments'][0]['declared_state'],'unknown') self.assertEqual(result['boundaries'],[]); self.assertIsNone(result['global_capture_percent']) self.assertIsNone(result['global_traffic_denominator']); self.assertFalse(result['resource_permission_granted']) def test_unobserved_partial_overlap_and_interval_gaps_remain_visible(self): b=self.r['boundaries'][0]; b.update(starts_at=later(2),ends_at=later(8),state='partial') other=copy.deepcopy(b); other.update(id=self.r['origin']+'/observers/other',starts_at=later(4),ends_at=later(6)) self.r['boundaries'].append(other) result=self.reduce(); segments=result['scope_segments'] self.assertEqual([s['declared_state'] for s in segments],['unknown','partial','partial','partial','unknown']) self.assertTrue(segments[2]['overlap']); self.assertTrue(result['boundary_counts_are_not_additive']) b.update(kind='unobserved',state='unobserved',observer_id=None,binding=None,event_profiles=[]) self.r['boundaries']=[b]; self.assertEqual(self.reduce()['scope_segments'][1]['declared_state'],'unobserved') def test_known_cohort_separates_timely_late_missing_and_pending(self): events=[]; expectations=[] for i in range(4): e=copy.deepcopy(self.event); e['event_id']=uid(200+i); e['operation']['id']=uid(210+i); events.append(e) expectations.append(expected(self.bid,e,number=220+i,due=later(40) if i==3 else later(10))) result=self.reduce([self.intake(events[0]),self.intake(events[1],received=20)],expectations) row=result['boundaries'][0] self.assertEqual([row[k] for k in ('expected','timely','late','missing','pending')],[4,1,1,1,1]) self.assertEqual(row['timely_ratio'],dict(numerator=1,denominator=4,percent=25.0)) self.assertEqual(row['received_ratio']['percent'],50.0) self.assertEqual(row['outcomes'],{'client_action_completed':2}); self.assertFalse(result['downstream_use_verified']) def test_as_of_does_not_backfill_a_missing_report_with_future_receipt(self): e=expected(self.bid,self.event); record=self.intake(received=20) before=self.reduce([record],[e],as_of=later(15)); after=self.reduce([record],[e]) self.assertEqual(before['future_records'],1); self.assertEqual(before['boundaries'][0]['missing'],1) self.assertEqual(after['boundaries'][0]['late'],1); self.assertEqual(after['boundaries'][0]['timely'],0) def test_zero_expectations_is_unknown_ratio_and_unexpected_reports_are_not_a_denominator(self): row=self.reduce([self.intake()])['boundaries'][0] self.assertEqual(row['unexpected'],1); self.assertEqual(row['expected'],0); self.assertIsNone(row['received_ratio']['percent']) def test_query_cohort_accepts_later_completion_but_excludes_unrelated_outside_records(self): event=copy.deepcopy(self.event); event['occurred_at']=later(12);event['operation']['ended_at']=later(12) e=expected(self.bid,self.event,due=later(15)); record=self.intake(event,received=13) self.assertEqual(self.reduce([record],[e])['boundaries'][0]['timely'],1) row=self.reduce([record])['boundaries'][0] self.assertEqual((row['observed'],row['unexpected']),(0,0)) def test_exact_retries_and_duplicate_expectation_ids_do_not_inflate_rates(self): record=self.intake(); later_receipt=dict(record,record_id=uid(333),received_at=later(25)) e=expected(self.bid,self.event); equivalent=dict(e,id=uid(334)) results=[self.reduce(records,[e,e,equivalent]) for records in ([record,later_receipt],[later_receipt,record])] self.assertEqual(results[0],results[1]); result=results[0]; row=result['boundaries'][0] self.assertEqual(result['exact_retries'],1); self.assertEqual(result['expectation_retries'],2) self.assertEqual((row['expected'],row['timely']),(1,1)) def test_event_and_terminal_conflicts_are_order_independent(self): a=self.intake(); changed=copy.deepcopy(self.event); changed['operation']['state']='failed' same_identity=self.intake(changed); e=expected(self.bid,self.event) for records in itertools.permutations([a,same_identity]): result=self.reduce(records,[e]); self.assertEqual(result['event_conflicts'],1) self.assertEqual(result['boundaries'][0]['received_ratio']['numerator'],0) changed['event_id']=uid(350); terminal=self.intake(changed) all_results=[self.reduce(records,[e]) for records in itertools.permutations([a,terminal,a])] self.assertTrue(all(r==all_results[0] for r in all_results)); self.assertEqual(all_results[0]['boundaries'][0]['conflicted'],1) def test_expectation_conflicts_never_shrink_or_rewrite_the_denominator(self): e=expected(self.bid,self.event); changed=dict(e,due_at=later(40)) results=[self.reduce([self.intake()],[e,changed]),self.reduce([self.intake()],[changed,e])] self.assertEqual(results[0],results[1]); row=results[0]['boundaries'][0] self.assertEqual((row['expected'],row['conflicted'],row['received_ratio']['numerator']),(1,1,0)) outside=dict(e,resource_url=self.r['origin']+'/elsewhere'); other=dict(outside,due_at=later(40)) self.assertEqual(self.reduce([], [outside,other])['boundaries'][0]['expected'],0) def test_collector_receipt_uses_exact_payload_and_is_not_another_use(self): self.r['boundaries'][0].update(kind='collector',observer_id=self.binding['service_id']) payload=json.dumps(self.event,indent=1).encode(); record=self.intake(payload=payload) e=expected(self.bid,self.event,kind='report',digest=crypto.digest(payload)) row=self.reduce([record],[e])['boundaries'][0] self.assertEqual(row['outcomes'],{'collector_received':1}); self.assertEqual(row['timely'],1) e['subject']['payload_digest']=crypto.digest(crypto.json_bytes(self.event)) self.assertEqual(self.reduce([record],[e])['boundaries'][0]['missing'],1) def test_gateway_attempt_unresolved_admission_and_full_vs_failed_transport_are_separate(self): self.query['action']='retrieve' self.bid=self.r['origin']+'/odexa/gateway' self.r['boundaries'][0].update(id=self.bid,kind='gateway',observer_id=self.r['origin']+'/gateway') event=delivery_report(self.binding,self.values['introspection_response'],self.r['origin']+'/gateway',event_id=uid(360), hop_id=uid(361),started_at=NOW,occurred_at=later(1),bytes_written=3,representation_digest=crypto.digest(b'abc'),complete=True) e=expected(self.bid,event,kind='delivery'); missing=copy.deepcopy(e); missing.update(id=uid(362)); missing['subject']['id']=uid(363) row=self.reduce([self.intake(event)],[e,missing])['boundaries'][0] self.assertEqual(row['unresolved_admissions'],1); self.assertEqual(row['outcomes'],{'gateway_full_write_completed':1}) failed=delivery_report(self.binding,self.values['introspection_response'],event['reporter_id'],event_id=uid(364), hop_id=uid(365),started_at=NOW,occurred_at=later(1),bytes_written=1,representation_digest=None,complete=False) self.assertEqual(self.reduce([self.intake(failed)],[e])['boundaries'][0]['outcomes'],{'gateway_write_failed':1}) def storage_events(self): write=copy.deepcopy(self.event); write.update(event_profile=ss.PROFILE,quantity=None,unit=None) write['operation'].update(kind='action',copy_context=dict(copy_id=uid(370),acquired_at=NOW,retain_until=later(120),parent_write=None),storage=None) start=copy.deepcopy(write); start['event_id']=uid(371) start['operation'].update(id=uid(372),kind='storage_session',state='in_progress',started_at=later(1),ended_at=None, storage=dict(trigger='start',checkpoint_index=0,scheduled_at=later(1))) start['related_events']=[dict(reporter_id=write['reporter_id'],event_id=write['event_id'],payload_digest=crypto.digest(crypto.json_bytes(write)))] return write,start def test_selected_storage_write_and_continuing_session_are_separate(self): self.r['boundaries'][0]['event_profiles']=[ss.PROFILE] write,start=self.storage_events() entries=[expected(self.bid,write,number=380,profile=ss.PROFILE), expected(self.bid,start,number=381,kind='storage_start',profile=ss.PROFILE)] result=self.reduce([self.intake(write),self.intake(start)],entries); row=result['boundaries'][0] self.assertEqual(row['timely'],2) self.assertEqual(row['outcomes'],{'client_action_completed':1,'client_storage_start':1}) self.r['boundaries'][0]['event_profiles']=['core']; row=self.reduce([self.intake(start)],[entries[1]])['boundaries'][0] self.assertEqual(row['profile_mismatches'],1); self.assertEqual(row['profile_mismatch'],1); self.assertEqual(row['missing'],0) def test_unknown_scope_reporter_profile_and_unverified_intake_do_not_become_coverage(self): e=expected(self.bid,self.event); e['boundary_id']=self.r['origin']+'/not-declared' self.assertEqual(len(self.reduce([],[e])['unknown_expectations']),1) self.r['boundaries'][0]['observer_id']='https://other.example/reporter' result=self.reduce([self.intake()]); self.assertEqual(result['unmatched_records'],1) self.r['boundaries'][0]['observer_id']=self.event['reporter_id'] unsigned=dict(self.intake(),assurance='unverified',authenticated_reporter_id=None,reporter_jws=None) result=self.reduce([unsigned]); self.assertEqual(result['unverified_records'],1); self.assertEqual(result['boundaries'][0]['observed'],0) event=copy.deepcopy(self.event); event['event_profile']='guessed-future-profile' self.assertEqual(self.reduce([self.intake(event)])['invalid_records'],1) def test_another_agreement_or_selected_profile_cannot_fill_a_known_gap(self): e=expected(self.bid,self.event); event=copy.deepcopy(self.event); event['agreement_id']=uid(390) row=self.reduce([self.intake(event)],[e])['boundaries'][0] self.assertEqual(row['binding_mismatch'],1); self.assertEqual(row['received_ratio']['numerator'],0) write,_=self.storage_events(); self.r['boundaries'][0]['event_profiles']=['core',ss.PROFILE] row=self.reduce([self.intake(write)],[e])['boundaries'][0] self.assertEqual(row['profile_mismatch'],1); self.assertEqual(row['received_ratio']['numerator'],0) def test_closed_registry_exact_policy_scope_time_and_no_secret_fields(self): for mutate in [lambda r:r.update(token='secret'),lambda r:r.update(protocol_version='1.2.0-draft.2'), lambda r:r.update(policy_digest='sha256:'+'0'*64),lambda r:r.update(revision=True), lambda r:r['boundaries'][0].update(resource_ids=['unlisted']), lambda r:r['boundaries'][0].update(ends_at=r['issued_at']), lambda r:r['boundaries'][0].update(kind='unobserved'), lambda r:r['boundaries'].append(copy.deepcopy(r['boundaries'][0]))]: bad=copy.deepcopy(self.r); mutate(bad) with self.assertRaises((ValueError,TypeError)): o.validate_registry(bad,self.pol) with self.assertRaises(ValueError): o.validate_registry(self.r,self.pol+b' ') def test_structural_fixture_semantics_are_checked_separately(self): corpus=cases() for case in corpus['cases']: with self.subTest(name=case['name']): validator=(lambda value:o.validate_registry(value,self.pol)) if case['definition']=='registry' else ( lambda value:o.validate_expectation(value,self.r['origin'])) if case['semantic_valid']: validator(case['value']) else: with self.assertRaises((ValueError,TypeError)):validator(case['value']) def test_history_restart_rollback_same_revision_conflict_and_repeated_read_clock(self): with tempfile.TemporaryDirectory() as directory: path=Path(directory)/'history.sqlite3'; h=o.RegistryHistory(path) h.observe(snapshot(self.r,self.pol),clock=lambda:NOW) h.observe(snapshot(self.r,self.pol,later(3)),clock=lambda:later(3)) with self.assertRaises(ValueError): h.observe(snapshot(self.r,self.pol,later(2)),clock=lambda:later(2)) original=copy.deepcopy(self.r); self.r['revision']=2 h.observe(snapshot(self.r,self.pol,later(4)),clock=lambda:later(4)); h=o.RegistryHistory(path) with self.assertRaises(ValueError): h.observe(snapshot(original,self.pol,later(5)),clock=lambda:later(5)) self.r['boundaries'][0]['state']='partial' with self.assertRaises(ValueError): h.observe(snapshot(self.r,self.pol,later(5)),clock=lambda:later(5)) self.assertEqual(o.validate_snapshot(h.get(self.r['origin'],1)),original) def test_unverified_or_stale_publication_cannot_be_retained(self): with tempfile.TemporaryDirectory() as directory: h=o.RegistryHistory(Path(directory)/'history.sqlite3'); snap=snapshot(self.r,self.pol) with self.assertRaises(ValueError):h.observe(snap,clock=lambda:later(6)) for field,value in [('tls_verified',False),('redirected',True),('registry_digest','sha256:'+'0'*64)]: changed=o.RegistrySnapshot(snap.registry_bytes,snap.policy_bytes,dict(snap.observation,**{field:value})) with self.assertRaises(ValueError):h.observe(changed,clock=lambda:NOW) with self.assertRaises(ValueError):h.get(self.r['origin'],1) def test_registry_acquisition_age_covers_both_requests(self): for delay,accepted in ((2,True),(4,False)): with self.subTest(delay=delay), tempfile.TemporaryDirectory() as directory: h=o.RegistryHistory(Path(directory)/'history.sqlite3'); elapsed=[0] def request(origin,path,**kwargs): elapsed[0]+=delay return SimpleNamespace(body=self.pol if path=='/odexa.json' else crypto.json_bytes(self.r)) transport=SimpleNamespace(request=request) with mock.patch.object(o.time,'monotonic',side_effect=lambda:elapsed[0]): if accepted: snap=o.fetch_registry(transport,self.r['origin'],history=h,clock=lambda:later(elapsed[0])) self.assertEqual(snap.observation['checked_at'],NOW) self.assertEqual(h.get(self.r['origin'],1).observation['checked_at'],NOW) else: with self.assertRaises(ValueError): o.fetch_registry(transport,self.r['origin'],history=h,clock=lambda:later(elapsed[0])) with self.assertRaises(ValueError):h.get(self.r['origin'],1) class CoverageTLSTests(unittest.TestCase): @classmethod def setUpClass(cls): gateway_fixture.GatewayReportTLSTests.setUpClass() @classmethod def tearDownClass(cls): gateway_fixture.GatewayReportTLSTests.tearDownClass() def setUp(self): self.g=gateway_fixture.GatewayReportTLSTests(); self.g.setUp(); self.f=self.g.f self.r=declaration(crypto.json_bytes(self.f.policy),self.f.binding,self.f.clients['gateway']['reporter_id'], kind='gateway',bid=self.f.origin+'/odexa/gateway') self.code=200; original=self.f.origin_server.route def route(method,path,headers,body): if path==o.PATH:return self.code,{'Content-Type':'application/json'},crypto.json_bytes(self.r) return original(method,path,headers,body) self.f.origin_server.route=route self.history=o.RegistryHistory(self.f.path/'coverage.sqlite3') def tearDown(self):self.g.tearDown() def fetch(self):return o.fetch_registry(self.f.transport,self.f.origin,history=self.history,clock=self.f.now) def test_static_registry_fetch_needs_no_service_payment_or_credential(self): self.r['boundaries']=[]; snap=self.fetch() self.assertEqual(o.validate_snapshot(snap)['boundaries'],[]) self.assertEqual([p for _,p,_,_ in self.f.origin_server.requests],['/odexa.json',o.PATH]) self.assertEqual(self.f.provider_server.requests,[]) for _,_,headers,_ in self.f.origin_server.requests:self.assertNotIn('Authorization',headers) def test_actual_signed_collector_intake_correlates_to_trusted_admission(self): rid=self.g.queue(); self.assertEqual(self.g.gateway.publish_reports()[0]['state'],'acknowledged') retained=self.g.gateway.report_records()[0] body,_=crypto.verify_jws(bytes(retained['acknowledgement']),crypto.public_jwk(self.f.signing_key),'odexa-event-record+jws') intake=crypto.strict_json(body) payload,_=crypto.verify_jws(intake['reporter_jws'],crypto.public_jwk(self.f.gateway_key),'odexa-event+jws') self.assertEqual(crypto.digest(payload),intake['payload_digest']); event=crypto.strict_json(payload) admitted=crypto.strict_json(bytes(self.g.gateway.records()[0]['response']))['admission']['checked_at'] e=expected(self.r['boundaries'][0]['id'],event,kind='delivery',at=admitted,due=add_seconds(admitted,60)) query=dict(url=event['resource_url'],action='retrieve',purpose='public_retrieval',starts_at=admitted,ends_at=add_seconds(self.f.now(),1)) result=o.reduce_coverage(self.fetch(),query=query,verified_intakes=[intake],expectations=[e],as_of=add_seconds(self.f.now(),2)) row=result['boundaries'][0]; self.assertEqual(row['received_ratio'],dict(numerator=1,denominator=1,percent=100.0)) self.assertIsNone(result['global_capture_percent']); self.assertFalse(result['downstream_use_verified']) self.assertEqual(rid,event['http']['delivery_id']) def test_failed_acquisition_or_changed_policy_never_substitutes_cached_registry(self): self.fetch(); self.code=503 with self.assertRaises(ValueError):self.fetch() self.code=200; self.f.policy['revision']=2 with self.assertRaises(ValueError):self.fetch() retained=self.history.get(self.f.origin,1) self.assertEqual(o.validate_snapshot(retained)['policy_revision'],1) def test_report_only_participation_never_claims_gateway_delivery(self): f=self.f; f.request['actions']=['store']; f.agree(wire=True) event=copy.deepcopy(fixtures()[0]['report']); event.update(f.binding,resource_url=f.request['url'], reporter_id=f.clients['agent']['reporter_id'],policy_id=f.policy['policy_id'],policy_revision=f.policy['revision'], agreement_id=f.agreement_id,occurred_at=f.now()) event['operation'].update(started_at=event['occurred_at'],ended_at=event['occurred_at']) signed=crypto.sign_jws(crypto.json_bytes(event),f.agent_key,f.clients['agent']['key_id'],'odexa-event+jws').encode() _,wire=f.call('events',signed,wire=True,status=201) payload,_=crypto.verify_jws(wire,crypto.public_jwk(f.signing_key),'odexa-event-record+jws') intake=crypto.strict_json(payload) crypto.verify_jws(intake['reporter_jws'],crypto.public_jwk(f.agent_key),'odexa-event+jws') self.r=declaration(crypto.json_bytes(f.policy),f.binding,event['reporter_id']) e=expected(self.r['boundaries'][0]['id'],event,at=event['occurred_at'],due=add_seconds(event['occurred_at'],60)) query=dict(url=event['resource_url'],action='store',purpose='public_retrieval',starts_at=event['occurred_at'],ends_at=add_seconds(f.now(),1)) result=o.reduce_coverage(self.fetch(),query=query,verified_intakes=[intake],expectations=[e],as_of=add_seconds(f.now(),2)) self.assertEqual(result['boundaries'][0]['outcomes'],{'client_action_completed':1}) self.assertFalse(result['downstream_use_verified']); self.assertEqual(self.g.gateway.records(),[]) self.assertFalse(any(path.endswith(('/tokens','/introspect')) for _,path,_,_ in f.provider_server.requests)) def cases(): values,context=fixtures(); pol=context[0]; binding=values['binding']; event=values['report'] registry=declaration(pol,binding,event['reporter_id']); entries=[] def add(name,definition,value,structural=True,semantic=True): entries.append(dict(name=name,definition=definition,value=copy.deepcopy(value),structural_valid=structural,semantic_valid=semantic)) add('coverage_agent','registry',registry) for kind,observer in [('gateway',registry['origin']+'/gateway'),('collector',binding['service_id'])]: add('coverage_'+kind,'registry',declaration(pol,binding,observer,kind=kind)) static=dict(copy.deepcopy(registry),boundaries=[]); add('coverage_static','registry',static) unobserved=copy.deepcopy(registry); unobserved['boundaries'][0].update(kind='unobserved',state='unobserved',observer_id=None,binding=None,event_profiles=[]) add('coverage_unobserved','registry',unobserved) expected_action=expected(registry['boundaries'][0]['id'],event); add('coverage_expected_action','expectation',expected_action) expected_report=expected(registry['boundaries'][0]['id'],event,kind='report',digest=crypto.digest(crypto.json_bytes(event))) add('coverage_expected_receipt','expectation',expected_report) mutations=[('closed',lambda r:r.update(token='not-public'),False), ('legacy_version',lambda r:r.update(protocol_version='1.2.0-draft.2'),False), ('boolean_revision',lambda r:r.update(revision=True),False), ('wrong_registry_url',lambda r:r.update(registry_id='https://other.example'+o.PATH),True), ('wrong_policy_digest',lambda r:r.update(policy_digest='sha256:'+'0'*64),True), ('wrong_policy_revision',lambda r:r.update(policy_revision=2),True), ('unknown_resource',lambda r:r['boundaries'][0].update(resource_ids=['unknown']),True), ('reverse_interval',lambda r:r['boundaries'][0].update(ends_at=r['issued_at']),True), ('unobserved_with_actor',lambda r:r['boundaries'][0].update(kind='unobserved'),False), ('gateway_storage_profile',lambda r:r['boundaries'][0].update(kind='gateway',event_profiles=[ss.PROFILE]),False), ('duplicate_id',lambda r:r['boundaries'].append(dict(copy.deepcopy(r['boundaries'][0]),state='partial')),True)] for name,mutate,structural in mutations: value=copy.deepcopy(registry); mutate(value); add('coverage_'+name,'registry',value,structural,False) for name,mutate,structural in [('deadline_reversed',lambda e:e.update(due_at=later(-1)),True), ('wrong_basis',lambda e:e.update(basis='gateway_admission'),True), ('storage_not_selected',lambda e:e['subject'].update(kind='storage_start'),True), ('expectation_closed',lambda e:e.update(secret='not-public'),False), ('missing_agreement',lambda e:e.update(agreement_id=None),True)]: value=copy.deepcopy(expected_action);mutate(value);add('coverage_'+name,'expectation',value,structural,False) return dict(schema='observation',cases=entries) if __name__=='__main__': if '--cases' in sys.argv: print(json.dumps(cases(),indent=2)) else: unittest.main()