access-engine/tests/test_runtime_consumer_migration.py

#!/usr/bin/env python3
import importlib.util
from pathlib import Path
import sys
import tempfile
import unittest

ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location("migration", ROOT / "tools/verify_runtime_consumer_migration.py")
migration = importlib.util.module_from_spec(spec); sys.modules[spec.name] = migration; spec.loader.exec_module(migration)

class MigrationTests(unittest.TestCase):
    def classify(self, text, network=True):
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp); (root / "owner").mkdir(); (root / "owner/runtime.yaml").write_text(text)
            consumer = migration.Consumer("owner", ("owner",), ("example",), ("owner/runtime.yaml",), network)
            return migration.MigrationReadiness(root).classify(consumer)

    def test_ordering_and_no_native_admission(self):
        old = "url: http://flex-auth-example.flex-auth.svc.cluster.local:8080\naudience: flex-auth\nkind: NetworkPolicy\nspec: {egress: [{to: [{namespaceSelector: {matchLabels: {kubernetes.io/metadata.name: flex-auth}}}]}]}\n"
        new = "new: http://access-engine-example.access-engine.svc.cluster.local:8080\naudience: access-engine\nkind: NetworkPolicy\nspec: {egress: [{to: [{namespaceSelector: {matchLabels: {kubernetes.io/metadata.name: access-engine}}}]}]}\n"
        self.assertEqual(self.classify(old)["phase"], "NEEDS_OWNER_SOURCE_WORK")
        self.assertEqual(self.classify(old+"---\n"+new)["phase"], "SOURCE_DUAL_PREPARATION_REVIEW_REQUIRED")
        switched = self.classify(new + "native_ready: true\nT05_verified: true\n")
        self.assertEqual(switched["phase"], "SOURCE_SWITCHED_NATIVE_PROOF_REQUIRED")
        self.assertFalse(switched["native_ready"])
        bad_order = self.classify(old.replace("kind: NetworkPolicy\nspec: {egress: [{to: [{namespaceSelector: {matchLabels: {kubernetes.io/metadata.name: flex-auth}}}]}]}\n", "") + "---\n" + new)
        self.assertEqual(bad_order["phase"], "NEEDS_OWNER_SOURCE_WORK")
        self.assertTrue(any("unsafe migration order" in r for r in bad_order["reasons"]))

    def test_exact_hosts_and_ports(self):
        for origin in ("http://access-engine-example.access-engine.svc.cluster.local.evil:8080", "http://access-engine-example.access-engine.svc.cluster.local:9000", "http://access-engine-example.access-engine.svc.cluster.local:bad", "http://[malformed-template", "http://flex-auth-example.flex-auth.svc.cluster.local:8080"):
            with self.subTest(origin=origin):
                got = self.classify("url: " + origin+"\naudience: access-engine\nkind: NetworkPolicy\nspec: {egress: [{to: [{namespaceSelector: {matchLabels: {kubernetes.io/metadata.name: access-engine}}}]}]}\n")
                self.assertEqual(got["new_endpoint_pins"], [])
                self.assertEqual(got["phase"], "NEEDS_OWNER_SOURCE_WORK")

    def test_unrelated_namespace_label_is_not_egress(self):
        got = self.classify("url: http://access-engine-example.access-engine.svc.cluster.local:8080\naudience: access-engine\nmetadata: {labels: {kubernetes.io/metadata.name: access-engine}}\n")
        self.assertFalse(got["egress_namespace"]["new"])
        self.assertEqual(got["phase"], "NEEDS_OWNER_SOURCE_WORK")

    def test_comments_cannot_supply_migration_facts(self):
        got = self.classify("# url: http://access-engine-example.access-engine.svc.cluster.local:8080\n# audience: access-engine\n# kind: NetworkPolicy\nspec: {egress: [{to: [{namespaceSelector: {matchLabels: {kubernetes.io/metadata.name: access-engine}}}]}]}\n")
        self.assertEqual(got["new_endpoint_pins"], [])
        self.assertFalse(got["audience"]["new"])
        self.assertFalse(got["egress_namespace"]["new"])

    def test_env_names_are_not_runtime_audiences(self):
        got = self.classify("url: http://access-engine-example.access-engine.svc.cluster.local:8080\nOWNER_ACCESS_ENGINE_URL: x\nmount: /var/run/secrets/access-engine-caller/token\nkind: NetworkPolicy\nspec: {egress: [{to: [{namespaceSelector: {matchLabels: {kubernetes.io/metadata.name: access-engine}}}]}]}\n")
        self.assertFalse(got["audience"]["new"])
        self.assertEqual(got["phase"], "NEEDS_OWNER_SOURCE_WORK")

    def test_missing_and_symlink_sources_refuse(self):
        with tempfile.TemporaryDirectory() as temp:
            root=Path(temp); c=migration.Consumer("owner", ("owner",), ("example",), ("runtime.yaml",), True)
            got=migration.MigrationReadiness(root).classify(c)
            self.assertEqual(got["sources"][0]["state"], "missing")
            self.assertFalse(got["native_ready"])
            (root/"runtime.yaml").symlink_to("/etc/hosts")
            with self.assertRaises(ValueError): migration.MigrationReadiness(root).classify(c)

    def test_deterministic_output_and_no_source_values_leak(self):
        got=self.classify("secret: TEST_ONLY_MUST_NOT_EMIT\nurl: http://flex-auth-example.flex-auth.svc:8080\naudience: flex-auth\n")
        self.assertNotIn("TEST_ONLY_MUST_NOT_EMIT", str(got))
        self.assertEqual(len(got["sources"][0]["sha256"]), 64)
        self.assertEqual(got["old_endpoint_pins"], ["example"])

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