railiance-platform/tests/test_keycape_issuer_eso.py

"""Public metadata fixtures only: no API, credentials, or native calls."""
import copy
import sys
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
import keycape_issuer_eso as contract


class FixedIssuerContract(unittest.TestCase):
    def observations(self):
        resources = contract.documents("forgejo")
        for resource in resources:
            resource["metadata"].update(uid="public-fixture-uid", resourceVersion="7")
            resource["status"] = {"conditions": [{"type": "Ready", "status": "True"}]}
        return resources

    def test_independent_forgejo_binding(self):
        # Independent owner-source example: CCR0041 role/SA/audience and the
        # existing issuer ESO draft's provider/target/relative key convention.
        store, external = contract.documents("forgejo")
        self.assertEqual(store["kind"], "ClusterSecretStore")
        self.assertEqual(store["metadata"], {"name": "openbao-keycape-forgejo-issuer"})
        self.assertEqual(store["spec"], {
            "provider": {"vault": {
                "server": "http://openbao.openbao.svc:8200", "path": "platform", "version": "v2",
                "auth": {"kubernetes": {"mountPath": "kubernetes",
                    "role": "external-secrets-keycape-forgejo",
                    "serviceAccountRef": {"name": "external-secrets", "namespace": "external-secrets",
                                          "audiences": ["openbao"]}}}}},
            "conditions": [{"namespaces": ["sso"]}]})
        self.assertEqual(external["metadata"], {"name": "keycape-forgejo-client", "namespace": "sso"})
        self.assertEqual(external["spec"], {
            "refreshInterval": "5m",
            "secretStoreRef": {"kind": "ClusterSecretStore", "name": "openbao-keycape-forgejo-issuer"},
            "target": {"name": "keycape-forgejo-client", "creationPolicy": "Owner", "deletionPolicy": "Retain"},
            "data": [{"secretKey": "client-secret", "remoteRef": {
                "key": "workloads/keycape/forgejo", "property": "CLIENT_SECRET"}}]})

    def test_valid_ready_metadata_allows_extra_controller_metadata(self):
        store, external = self.observations()
        contract.require_fixed_store(store, "forgejo")
        contract.require_ready_delivery(external, "forgejo")

    def test_store_drift_refuses(self):
        changes = [
            ("kind", "SecretStore"),
            ("metadata.namespace", "sso"),
            ("metadata.name", "caller-store"),
            ("spec.provider.vault.server", "https://wrong.invalid"),
            ("spec.provider.vault.path", "tenant"),
            ("spec.provider.vault.auth.kubernetes.role", "external-secrets-keycape-sibling"),
            ("spec.provider.vault.auth.kubernetes.serviceAccountRef.audiences", ["wrong"]),
            ("spec.provider.vault.auth.kubernetes.serviceAccountRef.namespace", "sso"),
            ("spec.conditions", [{"namespaces": ["sso", "other"]}]),
        ]
        for path, value in changes:
            with self.subTest(path=path):
                store = self.observations()[0]
                mutate(store, path, value)
                with self.assertRaises(contract.Refused):
                    contract.require_fixed_store(store, "forgejo")

    def test_external_drift_refuses(self):
        changes = [
            ("metadata.namespace", ""), ("metadata.namespace", None), ("metadata.namespace", "other"),
            ("metadata.name", "keycape-sibling-client"),
            ("spec.secretStoreRef.kind", "SecretStore"),
            ("spec.secretStoreRef.name", "openbao-keycape-forgejo-consumer"),
            ("spec.target.name", "other"), ("spec.target.creationPolicy", "Merge"),
            ("spec.target.deletionPolicy", "Delete"),
            ("spec.data", [{"secretKey": "client-secret", "remoteRef": {
                "key": "platform/workloads/keycape/forgejo", "property": "CLIENT_SECRET"}}]),
            ("spec.data", [{"secretKey": "wrong", "remoteRef": {
                "key": "workloads/keycape/forgejo", "property": "CLIENT_SECRET"}}]),
            ("spec.data", [{"secretKey": "client-secret", "remoteRef": {
                "key": "workloads/keycape/forgejo", "property": "wrong"}}]),
        ]
        for path, value in changes:
            with self.subTest(path=path, value=value):
                external = self.observations()[1]
                mutate(external, path, value)
                with self.assertRaises(contract.Refused):
                    contract.require_ready_delivery(external, "forgejo")

    def test_missing_or_unready_refuses(self):
        for check, index in [(contract.require_fixed_store, 0), (contract.require_ready_delivery, 1)]:
            for status in [None, {}, [], "bad", {"conditions": None},
                           {"conditions": [{"type": "Ready", "status": "False"}]}]:
                with self.subTest(check=check.__name__, status=status):
                    observed = self.observations()[index]
                    observed["status"] = status
                    with self.assertRaises(contract.Refused):
                        check(observed, "forgejo")
            for malformed in [None, [], {}, {"metadata": None}]:
                with self.assertRaises(contract.Refused):
                    check(malformed, "forgejo")

    def test_no_caller_store_or_path_override(self):
        for invalid in ["", "f", "../forgejo", "forgejo/sibling", "Forgejo", "forgejo\n", "-forgejo", "f" * 49, None]:
            with self.subTest(invalid=invalid), self.assertRaises(contract.Refused):
                contract.documents(invalid)
        with self.assertRaises(TypeError):
            contract.external_secret("forgejo", store_name="caller-store")

    def test_type_admitted_trailing_dash_is_preserved(self):
        self.assertEqual(contract.external_secret("forgejo-")["metadata"]["name"],
                         "keycape-forgejo--client")
        self.assertEqual(contract.documents("forgejo-")[0]["metadata"]["name"],
                         "openbao-keycape-forgejo--issuer")

    def test_delivery_document_is_only_external_and_detached(self):
        external = contract.external_secret("forgejo")
        self.assertEqual(external["kind"], "ExternalSecret")
        external["spec"]["secretStoreRef"]["name"] = "tampered"
        self.assertEqual(contract.external_secret("forgejo")["spec"]["secretStoreRef"]["name"],
                         "openbao-keycape-forgejo-issuer")


def mutate(resource, path, value):
    parts = path.split(".")
    for key in parts[:-1]:
        resource = resource[key]
    resource[parts[-1]] = copy.deepcopy(value)


if __name__ == "__main__":
    unittest.main(verbosity=2)