tegwick
e0359bcb6c
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
91 lines
4.5 KiB
Python
91 lines
4.5 KiB
Python
import copy
|
|
import json
|
|
import unittest
|
|
from pathlib import Path
|
|
from change_guard import CANDIDATE, check_current, check_admitted, transport_result, check_dependencies
|
|
|
|
ROOT = Path(__file__).parent
|
|
|
|
|
|
class ExactDeploymentBoundary(unittest.TestCase):
|
|
def setUp(self):
|
|
self.sealed = json.loads((ROOT / 'sealed-live-deployment.json').read_text())
|
|
self.admitted = copy.deepcopy(self.sealed)
|
|
self.admitted['spec']['template']['spec']['containers'][0]['image'] = CANDIDATE
|
|
|
|
def test_only_expected_image_is_accepted(self):
|
|
check_current(self.sealed, self.sealed)
|
|
check_admitted(self.admitted, self.sealed)
|
|
|
|
def test_bookkeeping_is_ignored(self):
|
|
value = copy.deepcopy(self.admitted)
|
|
value['status'] = {'readyReplicas': 0}
|
|
value['metadata']['generation'] = 100
|
|
value['metadata']['resourceVersion'] = 'changed'
|
|
value['metadata']['annotations']['kubectl.kubernetes.io/last-applied-configuration'] = 'server bookkeeping'
|
|
check_admitted(value, self.sealed)
|
|
|
|
def test_unrelated_drift_and_recreated_uid_are_refused(self):
|
|
mutators = [
|
|
lambda v: v['metadata'].update(uid='recreated'),
|
|
lambda v: v['metadata'].update(finalizers=['new']),
|
|
lambda v: v['metadata'].update(ownerReferences=[{'uid': 'new'}]),
|
|
lambda v: v['spec']['template']['spec'].update(hostNetwork=True),
|
|
lambda v: v['spec']['template']['spec'].update(serviceAccountName='other'),
|
|
lambda v: v['spec']['template']['spec'].update(automountServiceAccountToken=True),
|
|
lambda v: v['spec']['template']['spec']['containers'][0].update(env=[]),
|
|
lambda v: v['spec']['template']['spec']['containers'][0]['securityContext'].update(privileged=True),
|
|
lambda v: v['spec'].update(replicas=2),
|
|
lambda v: v['spec'].update(strategy={'type': 'RollingUpdate'}),
|
|
]
|
|
for mutate in mutators:
|
|
with self.subTest(mutate=mutate):
|
|
current = copy.deepcopy(self.sealed)
|
|
mutate(current)
|
|
with self.assertRaises(ValueError):
|
|
check_current(current, self.sealed)
|
|
admitted = copy.deepcopy(self.admitted)
|
|
mutate(admitted)
|
|
with self.assertRaises(ValueError):
|
|
check_admitted(admitted, self.sealed)
|
|
|
|
def test_unexpected_image_is_refused(self):
|
|
value = copy.deepcopy(self.admitted)
|
|
value['spec']['template']['spec']['containers'][0]['image'] = 'other'
|
|
with self.assertRaises(ValueError):
|
|
check_admitted(value, self.sealed)
|
|
|
|
def test_malformed_successful_transport_is_refused(self):
|
|
for output in ('banner', '{}', '[]', '{"all_checks_pass": "true"}', '{"all_checks_pass": false}'):
|
|
self.assertIs(transport_result(0, output)['all_checks_pass'], False)
|
|
self.assertIs(transport_result(1, '{"all_checks_pass":true}')['all_checks_pass'], False)
|
|
self.assertIs(transport_result(0, '{"all_checks_pass":true}')['all_checks_pass'], True)
|
|
|
|
def test_runtime_config_and_pvc_custody_are_sealed(self):
|
|
sealed = json.loads((ROOT / 'sealed-native-dependencies.json').read_text())
|
|
cm = copy.deepcopy(sealed['configmap'])
|
|
cm.pop('canonical_data_sha256')
|
|
service = json.loads((ROOT.parents[2] / 'deploy/lifecycle/service.candidate.json').read_text())
|
|
cm['data'] = next(x for x in service['items'] if x['kind'] == 'ConfigMap')['data']
|
|
pvc = copy.deepcopy(sealed['pvc'])
|
|
pvc['status'] = {'phase': 'Bound'}
|
|
check_dependencies(cm, pvc, sealed)
|
|
for mutate in (
|
|
lambda c, p: c['metadata'].update(uid='recreated'),
|
|
lambda c, p: c.update(immutable=False),
|
|
lambda c, p: c['data'].update({'runtime.json': '{}'}),
|
|
lambda c, p: c.update(binaryData={'extra': 'changed'}),
|
|
lambda c, p: p['metadata'].update(uid='recreated'),
|
|
lambda c, p: p['spec'].update(volumeName='replaced'),
|
|
lambda c, p: p['status'].update(phase='Pending'),
|
|
lambda c, p: p['metadata'].update(finalizers=['changed']),
|
|
):
|
|
with self.subTest(mutate=mutate):
|
|
changed_cm, changed_pvc = copy.deepcopy(cm), copy.deepcopy(pvc)
|
|
mutate(changed_cm, changed_pvc)
|
|
with self.assertRaises(ValueError):
|
|
check_dependencies(changed_cm, changed_pvc, sealed)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|