Record live receiver gate and isolate readiness from attestation jobs
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
This commit is contained in:
codex 2026-09-11 07:26:16 +02:00
parent 5cc325b744
commit 290e2b51fd
4 changed files with 211 additions and 3 deletions

View file

@ -77,6 +77,25 @@ class Contracts(unittest.TestCase):
lane.receiver_check(['kubectl'],dep['spec']['template']['spec']['containers'][0]['image'])
def test_completed_attestation_pods_do_not_block_the_receiver(self):
image='forgejo.coulomb.social/coulomb/audit-core@sha256:'+'a'*64
dep={'metadata':{'uid':'d','generation':1,'resourceVersion':'1'},'spec':{'replicas':1,'template':{'spec':{'containers':[{'name':'audit-core','image':image}]}}},'status':{'observedGeneration':1,'readyReplicas':1}}
receiver={'metadata':{'name':'receiver','uid':'p','labels':{'app.kubernetes.io/name':'audit-core','app.kubernetes.io/component':'receiver'}},'spec':dep['spec']['template']['spec'],'status':{'containerStatuses':[{'ready':True}]}}
attest={'metadata':{'name':'attestation-completed','labels':{'app.kubernetes.io/name':'audit-core','app.kubernetes.io/component':'attest'}},'status':{'phase':'Succeeded'}}
def kube(argv):
if 'deployment' in argv: result=dep
elif 'pods' in argv:
selector=argv[argv.index('-l')+1]
terms=[pair.split('=',1) for pair in selector.split(',')]
result={'items':[p for p in [receiver,attest] if all(p['metadata']['labels'].get(k)==v for k,v in terms)]}
else:
self.assertIn('receiver',argv)
result={k:True for k in ['load_bearing','redact','write_only','source_exact','tenant_exact']}
return subprocess.CompletedProcess(argv,0,json.dumps(result).encode(),b'')
with patch.object(lane,'assert_cluster'),patch.object(lane,'command',side_effect=kube),patch.object(lane,'bao',side_effect=AssertionError('no credentials')):
self.assertEqual(lane.receiver_check(['kubectl'],image)['pod_uid'],'p')
class OpenBaoExercise(unittest.TestCase):
@classmethod
def setUpClass(cls):