tegwick
6a0773bcfa
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
74 lines
4.1 KiB
Python
74 lines
4.1 KiB
Python
import base64
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
spec=importlib.util.spec_from_file_location('repair',Path(__file__).with_name('identity-provisioner-reconcile.py'))
|
|
repair=importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(repair)
|
|
META={'uid':'c6a9e6be-5bb5-47e6-9faa-06b8d72afec3','resource_version':'1234'}
|
|
SYNTHETIC='synthetic-only-current-password'
|
|
|
|
|
|
class ReconcileTests(unittest.TestCase):
|
|
def test_patch_has_exact_guards_and_changes_only_the_named_field(self):
|
|
changes=repair.patch(SYNTHETIC,META)
|
|
self.assertEqual(['test','test','replace'],[x['op'] for x in changes])
|
|
self.assertEqual('/metadata/uid',changes[0]['path'])
|
|
self.assertEqual('/metadata/resourceVersion',changes[1]['path'])
|
|
self.assertEqual('/data/LLDAP_LDAP_USER_PASS',changes[2]['path'])
|
|
self.assertEqual(SYNTHETIC,base64.b64decode(changes[2]['value']).decode())
|
|
|
|
def test_stale_metadata_stops_before_credential_use(self):
|
|
with patch.object(repair,'inspect',return_value=META), patch.object(repair,'run') as run:
|
|
with self.assertRaisesRegex(repair.Refused,'stale_metadata'):
|
|
repair.reconcile(SYNTHETIC,mode='apply',expected_uid=META['uid'],expected_resource_version='old')
|
|
run.assert_not_called()
|
|
|
|
def test_provider_refusal_stops_before_any_write(self):
|
|
with patch.object(repair,'inspect',return_value=META), patch.object(repair,'run',side_effect=repair.Refused('command_refused')) as run:
|
|
with self.assertRaises(repair.Refused):
|
|
repair.reconcile(SYNTHETIC,mode='apply',expected_uid=META['uid'],expected_resource_version=META['resource_version'])
|
|
self.assertEqual(1,run.call_count)
|
|
self.assertNotIn('patch',run.call_args.args[0])
|
|
|
|
def test_check_does_not_write_and_keeps_password_out_of_arguments(self):
|
|
with patch.object(repair,'inspect',return_value=META), patch.object(repair,'run',return_value='directory-check-passed') as run:
|
|
result=repair.reconcile(SYNTHETIC,mode='check')
|
|
self.assertFalse(result['secret_written'])
|
|
self.assertNotIn(SYNTHETIC,str(run.call_args.args))
|
|
self.assertNotIn(SYNTHETIC,json.dumps(result))
|
|
self.assertEqual(SYNTHETIC,json.loads(run.call_args.kwargs['data'])['password'])
|
|
|
|
def test_apply_dry_runs_then_writes_reloads_and_verifies(self):
|
|
results=['directory-check-passed','secret/lldap-secrets','secret/lldap-secrets','restarted','ready','reloaded-check-passed']
|
|
with patch.object(repair,'inspect',return_value=META), patch.object(repair,'run',side_effect=results) as run:
|
|
result=repair.reconcile(SYNTHETIC,mode='apply',expected_uid=META['uid'],expected_resource_version=META['resource_version'])
|
|
self.assertEqual('reconciled',result['result'])
|
|
self.assertFalse(result['provider_password_changed'])
|
|
calls=run.call_args_list
|
|
self.assertIn('--dry-run=server',calls[1].args[0])
|
|
self.assertNotIn('--dry-run=server',calls[2].args[0])
|
|
for call in calls:
|
|
self.assertNotIn(SYNTHETIC,str(call.args))
|
|
self.assertNotIn(base64.b64encode(SYNTHETIC.encode()).decode(),str(call.args))
|
|
|
|
def test_child_errors_never_reveal_output(self):
|
|
failed=subprocess.CompletedProcess([],1,stdout=SYNTHETIC,stderr=SYNTHETIC)
|
|
with patch.object(subprocess,'run',return_value=failed):
|
|
with self.assertRaises(repair.Refused) as result:
|
|
repair.run(['anything'])
|
|
self.assertEqual('command_refused',str(result.exception))
|
|
|
|
def test_wrong_cluster_and_controller_owner_refused(self):
|
|
with patch.object(repair,'run',return_value='foreign-cluster'):
|
|
with self.assertRaisesRegex(repair.Refused,'wrong_cluster'): repair.inspect()
|
|
with patch.object(repair,'run',side_effect=[repair.CLUSTER_UID,META['uid']+' 1234 1']):
|
|
with self.assertRaisesRegex(repair.Refused,'controller_owned_secret'): repair.inspect()
|
|
|
|
|
|
if __name__=='__main__': unittest.main()
|