1228cd8407
KUBECTL_ROUTE (agent default, founder-sudo = ssh railiance01 sudo -n) for the bounded proof scripts; one attended script with go/no-go gates, typed ADOPT confirmation, prune/force-off SSA sync and post-checks; evaluator and tests. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 527080@bnt-lap001 Assistant-Session: 5a0e864a-67be-4605-a77a-16635f7dd292
168 lines
8.1 KiB
Python
168 lines
8.1 KiB
Python
import copy
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
import yaml
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
spec = importlib.util.spec_from_file_location('eso_checks', ROOT / 'scripts/eso_attended_checks.py')
|
|
checks = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(checks)
|
|
CRED = 'Sec' + 'ret'
|
|
|
|
|
|
def crd(i, annotated=True):
|
|
meta = {'name': f'r{i}.external-secrets.io', 'labels': {'app': 'eso'}}
|
|
if annotated:
|
|
meta['annotations'] = {'argocd.argoproj.io/sync-options': 'Prune=false,Delete=false'}
|
|
return {'apiVersion': 'apiextensions.k8s.io/v1', 'kind': 'CustomResourceDefinition', 'metadata': meta,
|
|
'spec': {'group': 'external-secrets.io', 'scope': 'Namespaced'}}
|
|
|
|
|
|
DEPLOY = {'apiVersion': 'apps/v1', 'kind': 'Deployment', 'metadata': {'name': 'external-secrets', 'namespace': 'external-secrets'},
|
|
'spec': {'replicas': 1, 'template': {'spec': {'containers': [{'name': 'c', 'image': 'eso:v0.16.1'}]}}}}
|
|
HOLDER = {'apiVersion': 'v1', 'kind': CRED, 'metadata': {'name': 'external-secrets-webhook', 'namespace': 'external-secrets'}}
|
|
|
|
|
|
def render():
|
|
return [crd(i) for i in range(20)] + [DEPLOY]
|
|
|
|
|
|
def live():
|
|
out = [crd(i, annotated=False) for i in range(20)]
|
|
for obj in out:
|
|
obj['spec']['versions'] = ['defaulted']
|
|
obj['status'] = {'x': 1}
|
|
deploy = copy.deepcopy(DEPLOY)
|
|
deploy['spec']['template']['spec']['containers'][0]['imagePullPolicy'] = 'IfNotPresent'
|
|
return out + [deploy]
|
|
|
|
|
|
class ClassifyTests(unittest.TestCase):
|
|
def test_metadata_only_crd_annotation_is_go(self):
|
|
result = checks.classify(render(), live())
|
|
self.assertTrue(result['go'], result['problems'])
|
|
self.assertEqual(result['counts']['metadata-only'], 20)
|
|
self.assertEqual(result['counts']['unchanged'], 1)
|
|
|
|
def test_crd_spec_change_is_no_go(self):
|
|
objs = live(); objs[3]['spec']['scope'] = 'Cluster'
|
|
result = checks.classify(render(), objs)
|
|
self.assertFalse(result['go'])
|
|
self.assertIn('crd_spec_change:r3.external-secrets.io', result['problems'])
|
|
|
|
def test_missing_crd_is_no_go(self):
|
|
result = checks.classify(render(), live()[1:])
|
|
self.assertIn('crd_missing_live:r0.external-secrets.io', result['problems'])
|
|
|
|
def test_workload_spec_change_needs_review(self):
|
|
objs = live(); objs[-1]['spec']['template']['spec']['containers'][0]['image'] = 'eso:v0.15'
|
|
self.assertFalse(checks.classify(render(), objs)['go'])
|
|
|
|
def test_crd_count_and_store_objects_guarded(self):
|
|
self.assertIn('render_crd_count:19', checks.classify(render()[1:], live())['problems'])
|
|
store = {'apiVersion': 'external-secrets.io/v1beta1', 'kind': 'ClusterSecretStore', 'metadata': {'name': 's'}}
|
|
self.assertIn('render_contains_store_or_externalsecret', checks.classify(render() + [store], live() + [store])['problems'])
|
|
|
|
def test_credential_kind_in_live_input_refused(self):
|
|
with self.assertRaises(ValueError):
|
|
checks.classify(render(), live() + [HOLDER])
|
|
|
|
def test_split_excludes_credential_kind(self):
|
|
keep, held = checks.split_render(yaml.safe_dump_all(render() + [HOLDER]))
|
|
self.assertEqual(len(keep), 21)
|
|
self.assertEqual(held, [('', CRED, 'external-secrets', 'external-secrets-webhook')])
|
|
|
|
|
|
def stores(failing):
|
|
return {'items': [{'metadata': {'name': n}, 'spec': {'provider': {'vault': {'auth': {'tokenSecretRef': {'name': 'never-printed'}}}}},
|
|
'status': {'conditions': [{'type': 'Ready', 'status': 'False' if n in failing else 'True',
|
|
'reason': 'ValidationFailed' if n in failing else 'Valid'}]}}
|
|
for n in ('openbao', 'openbao-activity-core', 'openbao-email-connect', 'openbao-issue-core')]}
|
|
|
|
|
|
class StoreAndPostTests(unittest.TestCase):
|
|
def test_expected_failures_are_go_and_no_spec_in_receipt(self):
|
|
result = checks.store_baseline(stores({'openbao-activity-core', 'openbao-email-connect'}))
|
|
self.assertTrue(result['go'])
|
|
self.assertNotIn('never-printed', json.dumps(result))
|
|
|
|
def test_unexpected_failure_is_no_go(self):
|
|
result = checks.store_baseline(stores({'openbao-issue-core'}))
|
|
self.assertEqual(result['unexpected_not_ready'], ['openbao-issue-core'])
|
|
self.assertFalse(result['go'])
|
|
|
|
def test_empty_store_list_is_no_go(self):
|
|
self.assertFalse(checks.store_baseline({'items': []})['go'])
|
|
|
|
def app(self, **status):
|
|
base = {'sync': {'status': 'Synced', 'revision': '0.16.1'}, 'health': {'status': 'Healthy'},
|
|
'operationState': {'phase': 'Succeeded'}, 'resources': [{}] * 39}
|
|
base.update(status)
|
|
return {'spec': {'syncPolicy': {'syncOptions': ['ServerSideApply=true']}}, 'status': base}
|
|
|
|
def test_postcheck_pass_and_failures(self):
|
|
crds = [f'r{i}.external-secrets.io' for i in range(20)] + ['other.example.io']
|
|
before = checks.store_baseline(stores(set()))
|
|
self.assertTrue(checks.postcheck(self.app(), crds, before, before)['go'])
|
|
self.assertIn('crd_count:19', checks.postcheck(self.app(), crds[1:], before, before)['problems'])
|
|
changed = checks.store_baseline(stores({'openbao'}))
|
|
self.assertIn('store_baseline_changed', checks.postcheck(self.app(), crds, before, changed)['problems'])
|
|
failed = self.app(operationState={'phase': 'Failed'})
|
|
self.assertFalse(checks.postcheck(failed, crds, before, before)['go'])
|
|
|
|
def test_precheck_app_refuses_automation_and_errors(self):
|
|
app = self.app(conditions=[{'type': 'ComparisonError'}])
|
|
app['spec']['syncPolicy']['automated'] = {}
|
|
result = checks.precheck_app(app, 39)
|
|
self.assertIn('automated_sync_present', result['problems'])
|
|
self.assertIn('argo_error_condition', result['problems'])
|
|
self.assertTrue(checks.precheck_app(self.app(), 39)['go'])
|
|
|
|
def test_refresh_sample(self):
|
|
before = {'status': {'refreshTime': 'a', 'conditions': [{'type': 'Ready', 'status': 'True'}]}}
|
|
after = {'status': {'refreshTime': 'b', 'conditions': [{'type': 'Ready', 'status': 'True'}]}}
|
|
self.assertTrue(checks.refresh_sample(before, after)['go'])
|
|
self.assertFalse(checks.refresh_sample(before, before)['go'])
|
|
|
|
|
|
class ScriptTests(unittest.TestCase):
|
|
def test_dry_run_prints_only_and_writes_nothing(self):
|
|
evidence = ROOT / 'docs/evidence/2026-10-21-eso-adoption-prep'
|
|
before = set(evidence.iterdir())
|
|
result = subprocess.run([str(ROOT / 'scripts/rpf_wp_0044_t06_attended.sh'), '--dry-run'],
|
|
capture_output=True, text=True, timeout=60)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(set(evidence.iterdir()), before)
|
|
self.assertIn('sudo\\ -n\\ kubectl', result.stdout)
|
|
self.assertNotIn('ssh -t', result.stdout)
|
|
self.assertIn('would ask for ADOPT', result.stdout)
|
|
self.assertIn('patch\\ application\\ external-secrets', result.stdout)
|
|
self.assertNotIn('--prune', result.stdout)
|
|
|
|
def test_script_never_prunes_or_force_syncs(self):
|
|
text = (ROOT / 'scripts/rpf_wp_0044_t06_attended.sh').read_text()
|
|
sync = text.split("SYNC_JSON='", 1)[1].split("'", 1)[0]
|
|
payload = json.loads(sync)['operation']['sync']
|
|
self.assertIs(payload['prune'], False)
|
|
self.assertIs(payload['syncStrategy']['apply']['force'], False)
|
|
self.assertEqual(payload['syncOptions'], ['ServerSideApply=true'])
|
|
self.assertNotIn('--prune', text)
|
|
self.assertIn('read -r -p "Type ADOPT', text)
|
|
|
|
def test_gate_cli_exit_codes(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / 'r.json'
|
|
path.write_text(json.dumps({'ready_for_fresh_native_render_diff': False, 'missing_or_invalid': ['RPF-WP-0044-T04']}))
|
|
self.assertEqual(checks.main(['gate', '--readiness', str(path)]), 3)
|
|
path.write_text(json.dumps({'ready_for_fresh_native_render_diff': True}))
|
|
self.assertEqual(checks.main(['gate', '--readiness', str(path)]), 0)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|