tegwick
3e8996fe94
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a119cf-56d0-7fa3-840b-eb8732e1db0b
57 lines
3.2 KiB
Python
57 lines
3.2 KiB
Python
"""Exact release failure fences, mocked Kubernetes; no mail or cluster calls."""
|
|
import copy
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import pytest
|
|
|
|
@pytest.mark.parametrize('failure', ['capture','migration','patch_before','patch_unknown','rollout','verify',None])
|
|
def test_release_preserves_old_receiver_or_holds_new_receiver_stopped(tmp_path, monkeypatch, failure):
|
|
script=Path(__file__).resolve().parents[1]/'deploy/k8s/railiance/deploy-callers-2026-10-08.py'
|
|
spec=importlib.util.spec_from_file_location('caller_release',script)
|
|
module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
|
|
before={'spec':{'replicas':1,'template':{'spec':{'containers':[{'image':module.OLD}]} }},'status':{'readyReplicas':1}}
|
|
state=copy.deepcopy(before)
|
|
def document(*args):return copy.deepcopy(state)
|
|
def k(*args, **kwargs):
|
|
if args[0]=='patch':
|
|
operations=json.loads(args[args.index('-p')+1])
|
|
if operations[0]['path']=='/spec/template':
|
|
assert state['spec']['template']==operations[0]['value']
|
|
assert state['spec']['replicas']==1
|
|
state['spec']['replicas']=0
|
|
else:
|
|
assert module.image(state)==module.OLD
|
|
if failure=='patch_before':raise module.Refused('synthetic_failure')
|
|
state['spec']['template']['spec']['containers'][0]['image']=module.NEW
|
|
if failure=='patch_unknown':raise module.Refused('synthetic_lost_response')
|
|
if args[0]=='scale':state['spec']['replicas']=int(args[-1].split('=')[1])
|
|
if args[0]=='rollout' and module.image(state)==module.NEW and failure=='rollout':
|
|
raise module.Refused('synthetic_failure')
|
|
return b''
|
|
def capture(*args):
|
|
if failure=='capture':raise module.Refused('synthetic_failure')
|
|
return {'database_sha256':'synthetic','database_bytes':1}
|
|
def rehearse(*args):
|
|
if failure=='migration':raise module.Refused('synthetic_failure')
|
|
return {'all_five_table_rows_preserved':True,'mail_sent':False}
|
|
def verify():
|
|
if failure=='verify':raise module.Refused('synthetic_failure')
|
|
return {'mail_sent':False,'scoped_policy_active':False}
|
|
for name,value in [('k',k),('document',document),('capture',capture),('rehearse',rehearse),('verify',verify)]:
|
|
monkeypatch.setattr(module,name,value)
|
|
destination=tmp_path/'release'
|
|
body=json.dumps({'metadata':{'name':'synthetic-helper'}}).encode()
|
|
if failure:
|
|
with pytest.raises(module.Refused):module.deploy(before,body,destination)
|
|
receipt=json.loads((destination/'release-receipt.json').read_text())
|
|
assert receipt['status']=='failed'
|
|
changed=failure in {'patch_unknown','rollout','verify'}
|
|
assert module.image(state)==(module.NEW if changed else module.OLD)
|
|
assert state['spec']['replicas']==(0 if changed else 1)
|
|
else:
|
|
receipt=module.deploy(before,body,destination)
|
|
assert receipt['status']=='released_verified'
|
|
assert state['spec']['replicas']==1 and module.image(state)==module.NEW
|
|
assert destination.stat().st_mode & 0o777==0o700
|
|
assert (destination/'release-receipt.json').stat().st_mode & 0o777==0o600
|