tegwick
6fbce73b87
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 903097@bnt-lap001 Assistant-Session: aa320c8f-c3fa-46e7-8541-0b58e1e45d23
103 lines
4.7 KiB
Python
103 lines
4.7 KiB
Python
"""Synthetic-only tests for the T27 bind-env candidate; no cluster, no values."""
|
|
import copy
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).parent
|
|
spec = importlib.util.spec_from_file_location('bind_env_adapter', ROOT / 'bind-env-adapter.py')
|
|
a = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(a)
|
|
spec = importlib.util.spec_from_file_location('bind_env_fixture', ROOT / 'test_keycape_client.py')
|
|
f = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(f)
|
|
|
|
CASE = {'spec': {'type': 'oidc-browser-relying-party@1.1', 'parameters': f.FORGEJO_PARAMS}}
|
|
|
|
|
|
def deployment(env=None):
|
|
return {'kind': 'Deployment', 'metadata': {'name': 'keycape', 'namespace': 'sso', 'resourceVersion': '1'},
|
|
'spec': {'template': {'spec': {'containers': [
|
|
{'name': 'keycape', 'env': list(env or [{'name': 'KEYCAPE_CONFIG', 'value': '/etc/keycape/config.yaml'}])}]}}}}
|
|
|
|
|
|
class Store:
|
|
def __init__(self, d):
|
|
self.d, self.writes = d, []
|
|
|
|
def read(self):
|
|
return copy.deepcopy(self.d)
|
|
|
|
def write(self, new, rv):
|
|
if rv != self.d['metadata']['resourceVersion']:
|
|
raise RuntimeError('conflict')
|
|
self.d = copy.deepcopy(new)
|
|
self.d['metadata']['resourceVersion'] = str(int(rv) + 1)
|
|
self.writes.append(self.d)
|
|
return self.d['metadata']['resourceVersion']
|
|
|
|
|
|
class BindEnvTests(unittest.TestCase):
|
|
def test_binding_is_reference_only(self):
|
|
b = a.binding_for_case(CASE)
|
|
self.assertEqual(b['name'], 'KEYCAPE_FORGEJO_CLIENT_SECRET')
|
|
self.assertEqual(b['valueFrom']['secretKeyRef'], {'name': 'keycape-forgejo-client', 'key': 'client-secret'})
|
|
self.assertNotIn('value', b)
|
|
|
|
def test_run_then_created_only_rollback(self):
|
|
s = Store(deployment())
|
|
ref, run, rollback = a.build(case=CASE, expected_resource_version='1', read=s.read, write=s.write)
|
|
self.assertEqual(ref, a.REFERENCE)
|
|
r = run()
|
|
self.assertEqual((r['outcome'], r['created'], r['values_emitted']), ('success', ('KEYCAPE_FORGEJO_CLIENT_SECRET',), False))
|
|
self.assertEqual(len(s.d['spec']['template']['spec']['containers'][0]['env']), 2)
|
|
self.assertEqual(rollback()['outcome'], 'rolled-back')
|
|
self.assertEqual(s.d['spec']['template']['spec']['containers'][0]['env'], deployment()['spec']['template']['spec']['containers'][0]['env'])
|
|
with self.assertRaises(a.Refused):
|
|
rollback()
|
|
|
|
def test_existing_identical_binding_is_no_change_and_not_rollbackable(self):
|
|
s = Store(deployment([a.binding_for_case(CASE)]))
|
|
_, run, rollback = a.build(case=CASE, expected_resource_version='1', read=s.read, write=s.write)
|
|
self.assertEqual(run()['outcome'], 'no-change')
|
|
self.assertEqual(s.writes, [])
|
|
with self.assertRaises(a.Refused):
|
|
rollback()
|
|
|
|
def test_refusals(self):
|
|
other = dict(a.binding_for_case(CASE), valueFrom={'secretKeyRef': {'name': 'x', 'key': 'client-secret'}})
|
|
literal = {'name': 'KEYCAPE_FORGEJO_CLIENT_SECRET', 'value': 'synthetic'}
|
|
for env in ([other], [literal]):
|
|
s = Store(deployment(env))
|
|
_, run, _ = a.build(case=CASE, expected_resource_version='1', read=s.read, write=s.write)
|
|
with self.assertRaises(a.Refused):
|
|
run()
|
|
self.assertEqual(s.writes, [])
|
|
s = Store(deployment())
|
|
_, run, _ = a.build(case=CASE, expected_resource_version='0', read=s.read, write=s.write)
|
|
with self.assertRaises(a.Refused):
|
|
run()
|
|
with self.assertRaises(a.Refused):
|
|
a.build(case=CASE, expected_resource_version='1', read=s.read, write=s.write, allow_unimplemented=True)
|
|
public = {'spec': {'type': 'oidc-browser-relying-party@1.1',
|
|
'parameters': dict(f.FORGEJO_PARAMS, client_type='public', mfa_required='false')}}
|
|
with self.assertRaises(a.Refused):
|
|
a.binding_for_case(public)
|
|
with self.assertRaises(Exception):
|
|
a.binding_for_case({'spec': {'type': 'other@1', 'parameters': f.FORGEJO_PARAMS}})
|
|
with self.assertRaises(a.Refused):
|
|
a.plan(dict(deployment(), metadata={'name': 'keycape', 'namespace': 'other'}), a.binding_for_case(CASE))
|
|
with self.assertRaises(a.Refused):
|
|
a.plan(deployment(), dict(a.binding_for_case(CASE), value='synthetic'))
|
|
|
|
def test_rollback_refused_after_drift(self):
|
|
s = Store(deployment())
|
|
_, run, rollback = a.build(case=CASE, expected_resource_version='1', read=s.read, write=s.write)
|
|
run()
|
|
s.d['metadata']['resourceVersion'] = '9'
|
|
with self.assertRaises(a.Refused):
|
|
rollback()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|