railiance-platform/tests/test_eso_attended_checks.py

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