feat: exercise contained KeyCape approval client rollout

Assistant: codex
Assistant-Model: gpt-5.6-luna
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc

sso-mfa/k8s/keycape/test_approval_clients_rollout.py

@@ -0,0 +1,104 @@
+import base64
+import copy
+import importlib.util
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+import yaml
+
+PATH = Path(__file__).with_name('approval-clients-rollout.py')
+spec = importlib.util.spec_from_file_location('rollout', PATH)
+m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m)
+
+
+def fixture(raw=None):
+    raw = raw or 'authelia:\n  issuer: https://auth.coulomb.social\n  clientSecret: synthetic-fixture\nclients:\n  - clientId: existing-human\n    allowedScopes: [openid]\n# trailing owner comment\nother: preserved\n'
+    return {'kind': 'Secret', 'metadata': {'name': 'keycape-config', 'namespace': 'sso',
+            'uid': '2e94519d-1550-41c7-9701-2efe47fe1fd3', 'resourceVersion': '100'},
+            'data': {'config.yaml': base64.b64encode(raw.encode()).decode(), 'key.pem': 'c3ludGhldGlj'}}
+
+
+def deployment():
+    source = next(yaml.safe_load_all((PATH.parent / 'deployment.yaml').read_text()))
+    source['metadata'].update(uid='99ddd83c-cb3f-4847-bcf8-35f1aa87627f', resourceVersion='101')
+    return source
+
+
+class RolloutTests(unittest.TestCase):
+    def test_append_preserves_original_bytes_and_signing_key(self):
+        secret = fixture(); before = copy.deepcopy(secret)
+        actual = base64.b64decode(m.replacement(secret, m.registrations())).decode()
+        original = base64.b64decode(secret['data']['config.yaml']).decode()
+        insertion_start = original.index('other:')
+        self.assertTrue(actual.startswith(original[:insertion_start]))
+        self.assertTrue(actual.endswith(original[insertion_start:]))
+        self.assertEqual(secret, before)
+        self.assertEqual(len(yaml.safe_load(actual)['clients']), 3)
+
+    def test_indentless_and_final_client_sequence(self):
+        for text in ['authelia:\n  issuer: https://auth.coulomb.social\nclients:\n- clientId: human\nother: keep\n',
+                     'authelia:\n  issuer: https://auth.coulomb.social\nclients:\n  - clientId: human']:
+            parsed = yaml.safe_load(base64.b64decode(m.replacement(fixture(text), m.registrations())))
+            self.assertEqual(len(parsed['clients']), 3)
+
+    def test_reject_duplicate_existing_registration(self):
+        text = 'authelia:\n  issuer: https://auth.coulomb.social\nclients:\n  - clientId: secrets-engine-approval\n'
+        with self.assertRaisesRegex(m.LaneError, 'existing_client_registration'):
+            m.replacement(fixture(text), m.registrations())
+
+    def test_reject_ambiguous_yaml_and_wrong_issuer(self):
+        for raw in ['authelia: {}\nauthelia: {}\nclients: []',
+                    'authelia:\n  issuer: https://wrong.invalid\nclients:\n  - clientId: human\n']:
+            with self.assertRaises((m.LaneError, m.pin.IssuerPinError)):
+                m.replacement(fixture(raw), m.registrations())
+
+    def test_candidate_preserves_existing_settings(self):
+        old = deployment(); result = m.candidate_spec(old)
+        before = old['spec']['template']['spec']['containers'][0]
+        after = result['template']['spec']['containers'][0]
+        self.assertEqual(result['strategy'], {'type': 'Recreate'})
+        self.assertEqual(after['env'][:len(before['env'])], before['env'])
+        for field in ['livenessProbe', 'resources', 'volumeMounts', 'startupProbe']:
+            self.assertEqual(after[field], before[field])
+        self.assertEqual(after['image'], m.IMAGE)
+
+    def test_cas_patch_contains_both_preconditions_and_stdin_only(self):
+        with patch.object(m, 'command') as command:
+            command.return_value.stdout = b'{}'
+            m.patch_object(['kubectl'], 'secret', fixture(), '/data/config.yaml', 'synthetic-payload')
+            args, kwargs = command.call_args
+            self.assertNotIn('synthetic-payload', args[0])
+            self.assertEqual([p['path'] for p in kwargs['payload'][:2]], ['/metadata/uid', '/metadata/resourceVersion'])
+
+    def test_failed_acceptance_restores_compatible_pair(self):
+        state = {'secret': fixture(), 'deployment': deployment()}
+        original = copy.deepcopy(state)
+        def fake_patch(kube, kind, obj, path, value, dry=False):
+            changed = copy.deepcopy(obj)
+            if path == '/spec': changed['spec'] = value
+            else: changed['data']['config.yaml'] = value
+            if not dry:
+                changed['metadata']['resourceVersion'] = str(int(changed['metadata']['resourceVersion']) + 1)
+                state[kind] = changed
+            return copy.deepcopy(changed)
+        with tempfile.TemporaryDirectory() as directory, patch.object(m, 'assert_cluster'), \
+                patch.object(m, 'get', side_effect=lambda kube, kind, name: copy.deepcopy(state[kind])), \
+                patch.object(m, 'patch_object', side_effect=fake_patch), patch.object(m, 'ready', return_value={}), \
+                patch.object(m, 'acceptance', side_effect=m.LaneError('synthetic_acceptance_failure')):
+            receipt = {}
+            with self.assertRaisesRegex(m.LaneError, 'synthetic_acceptance_failure'):
+                m.rollout([], receipt, Path(directory) / 'recovery.json')
+            self.assertTrue(receipt['compatible_pair_restored'])
+            self.assertEqual(state['secret']['data'], original['secret']['data'])
+            self.assertEqual(state['deployment']['spec'], original['deployment']['spec'])
+            self.assertEqual((Path(directory) / 'recovery.json').stat().st_mode & 0o777, 0o600)
+
+    def test_unrelated_refusal_never_passes(self):
+        for status, body in [(500, {}), (400, {'error': 'invalid_profile_usage', 'feature': 'client_id'}),
+                              (401, {'error': 'invalid_profile_usage', 'feature': 'scope'})]:
+            with self.assertRaises(m.LaneError):
+                m.denied((status, body), 400, 'scope')
+
+if __name__ == '__main__': unittest.main()