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()