Updated test

Author: anniegracehu

tests/test_provider_kubeconfig.py

@@ -27,20 +27,63 @@ def _cluster_available(kubeconfig=""):
 
 SCRIPT = "provider-kubeconfig.py"
 
+# All CLI elements that must appear in --help
+HELP_ELEMENTS = [
+    "create", "delete", "update", "extract",
+    "namespace",
+    "-k", "--kubeconfig",
+    "-s", "--apiserverurl",
+    "-f", "--filename",
+    "-x", "--clustername",
+    "-p", "--permissionfile",
+    "-c", "--consumer",
+]
+
 
 class TestCli(unittest.TestCase):
     """provider-kubeconfig.py exposes expected CLI (actions and flags)."""
 
-    def test_help_shows_actions_and_flags(self):
+    def test_help_shows_all_actions(self):
+        root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+        proc = subprocess.run(
+            [sys.executable, os.path.join(root, SCRIPT), "--help"],
+            capture_output=True, text=True, cwd=root,
+        )
+        self.assertEqual(proc.returncode, 0, proc.stderr)
+        out = proc.stdout or ""
+        for elem in ["create", "delete", "update", "extract"]:
+            self.assertIn(elem, out, f"Action {elem} should appear in help")
+
+    def test_help_shows_all_flags(self):
+        root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+        proc = subprocess.run(
+            [sys.executable, os.path.join(root, SCRIPT), "--help"],
+            capture_output=True, text=True, cwd=root,
+        )
+        self.assertEqual(proc.returncode, 0)
+        out = proc.stdout or ""
+        for elem in HELP_ELEMENTS:
+            self.assertIn(elem, out, f"Help should mention {elem}")
+
+    def test_help_shows_namespace_argument(self):
         root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
         proc = subprocess.run(
             [sys.executable, os.path.join(root, SCRIPT), "--help"],
             capture_output=True, text=True, cwd=root,
         )
         self.assertEqual(proc.returncode, 0)
         out = proc.stdout or ""
-        for x in ["create", "delete", "update", "extract", "-k", "-s", "-c", "namespace"]:
-            self.assertIn(x, out)
+        self.assertIn("namespace", out.lower())
+
+    def test_update_without_permissionfile_exits_with_error(self):
+        """update action without -p exits with code 1."""
+        root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+        proc = subprocess.run(
+            [sys.executable, os.path.join(root, SCRIPT), "update", "default"],
+            capture_output=True, text=True, cwd=root,
+        )
+        self.assertNotEqual(proc.returncode, 0)
+        self.assertIn("permission", (proc.stdout or proc.stderr or "").lower())
 
 
 class TestKubeconfigIntegration(unittest.TestCase):
@@ -58,7 +101,7 @@ def setUpClass(cls):
         cls.has_cluster = _cluster_available(cls.kubeconfig)
         cls.kubeconfig_flag = " --kubeconfig=" + cls.kubeconfig if cls.kubeconfig else ""
 
-    def _create_and_get_kubeconfig(self, root, ns, sa="kubeplus-saas-provider", extra_args=None):
+    def _create_and_get_kubeconfig(self, root, ns, sa="kubeplus-saas-provider", extra_args=None, output_filename=None):
         """Run create, return (kubeconfig_dict, proc). Caller must delete to cleanup."""
         create_args = ["create", ns]
         if sa != "kubeplus-saas-provider":
@@ -71,7 +114,9 @@ def _create_and_get_kubeconfig(self, root, ns, sa="kubeplus-saas-provider", extr
         )
         if proc.returncode != 0:
             return None, proc
-        filename = sa + ".json"
+        filename = output_filename or (sa + ".json")
+        if not filename.endswith(".json"):
+            filename += ".json"
         kubeconfig_path = os.path.join(root, filename)
         if not os.path.exists(kubeconfig_path):
             return None, proc
@@ -88,44 +133,82 @@ def _delete_for_cleanup(self, root, ns, sa="kubeplus-saas-provider"):
             cwd=root, capture_output=True, timeout=60,
         )
 
-    def _assert_kubeconfig_valid(self, cfg, expected_server=None, expected_cluster_name=None, expected_namespace=None):
-        self.assertIsNotNone(cfg)
-        self.assertEqual(cfg.get("apiVersion"), "v1")
-        self.assertEqual(cfg.get("kind"), "Config")
+    def _assert_kubeconfig_valid(
+        self,
+        cfg,
+        expected_server=None,
+        expected_cluster_name=None,
+        expected_namespace=None,
+        expected_user_name=None,
+    ):
+        """Assert all kubeconfig fields are non-empty and optionally match expected values."""
+        self.assertIsNotNone(cfg, "kubeconfig should not be None")
+
+        # Top-level
+        self.assertEqual(cfg.get("apiVersion"), "v1", "apiVersion should be v1")
+        self.assertEqual(cfg.get("kind"), "Config", "kind should be Config")
+        self.assertTrue(cfg.get("current-context"), "current-context should be non-empty")
+
+        # Users
         self.assertTrue(cfg.get("users"), "users should be non-empty")
+        user = cfg["users"][0]
+        self.assertTrue(user.get("name"), "users[0].name should be non-empty")
+        self.assertTrue(user.get("user"), "users[0].user should be non-empty")
+        token = user.get("user", {}).get("token")
+        self.assertTrue(token, "users[0].user.token should be non-empty")
+        if expected_user_name:
+            self.assertEqual(user.get("name"), expected_user_name)
+
+        # Clusters
         self.assertTrue(cfg.get("clusters"), "clusters should be non-empty")
-        self.assertTrue(cfg.get("contexts"), "contexts should be non-empty")
-        token = cfg["users"][0].get("user", {}).get("token")
-        self.assertTrue(token, "token should be non-empty")
-        server = cfg["clusters"][0].get("cluster", {}).get("server")
-        self.assertTrue(server, "cluster server should be non-empty")
+        cluster_entry = cfg["clusters"][0]
+        self.assertTrue(cluster_entry.get("name"), "clusters[0].name should be non-empty")
+        self.assertTrue(cluster_entry.get("cluster"), "clusters[0].cluster should be non-empty")
+        cluster = cluster_entry["cluster"]
+        self.assertTrue(cluster.get("server"), "clusters[0].cluster.server should be non-empty")
+        self.assertIn("insecure-skip-tls-verify", cluster, "cluster should have insecure-skip-tls-verify")
         if expected_server:
-            self.assertEqual(server, expected_server)
+            self.assertEqual(cluster.get("server"), expected_server)
+
+        # Contexts
+        self.assertTrue(cfg.get("contexts"), "contexts should be non-empty")
+        ctx_entry = cfg["contexts"][0]
+        self.assertTrue(ctx_entry.get("name"), "contexts[0].name should be non-empty")
+        self.assertTrue(ctx_entry.get("context"), "contexts[0].context should be non-empty")
+        ctx = ctx_entry["context"]
+        self.assertTrue(ctx.get("cluster"), "contexts[0].context.cluster should be non-empty")
+        self.assertTrue(ctx.get("user"), "contexts[0].context.user should be non-empty")
+        self.assertTrue(ctx.get("namespace"), "contexts[0].context.namespace should be non-empty")
         if expected_cluster_name:
-            ctx_name = cfg.get("current-context") or (cfg["contexts"][0].get("name") if cfg["contexts"] else "")
-            self.assertEqual(ctx_name, expected_cluster_name)
+            self.assertEqual(cfg.get("current-context"), expected_cluster_name)
+            self.assertEqual(ctx_entry.get("name"), expected_cluster_name)
         if expected_namespace:
-            ctx_ns = cfg["contexts"][0].get("context", {}).get("namespace", "")
-            self.assertEqual(ctx_ns, expected_namespace)
+            self.assertEqual(ctx.get("namespace"), expected_namespace)
 
     def _sa_exists(self, sa, ns):
         out, err = _run_command("kubectl get sa " + sa + " -n " + ns + self.kubeconfig_flag)
         return out and sa in out and "NotFound" not in (err or "")
 
-    def test_provider_kubeconfig_has_nonempty_fields(self):
+    def test_provider_kubeconfig_all_fields_nonempty(self):
+        """Provider kubeconfig: every field that should exist is non-empty."""
         if not self.has_cluster:
             self.skipTest("No cluster reachable (set KUBECONFIG)")
         root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
         ns = "kubeplus-test-prov-" + str(os.getpid())
         try:
             cfg, proc = self._create_and_get_kubeconfig(root, ns)
             self.assertEqual(proc.returncode, 0, proc.stderr)
-            self._assert_kubeconfig_valid(cfg, expected_namespace=ns)
+            self._assert_kubeconfig_valid(
+                cfg,
+                expected_namespace=ns,
+                expected_user_name="kubeplus-saas-provider",
+            )
             self.assertTrue(self._sa_exists("kubeplus-saas-provider", ns))
         finally:
             self._delete_for_cleanup(root, ns)
 
-    def test_consumer_kubeconfig_has_namespace_in_context(self):
+    def test_consumer_kubeconfig_all_fields_nonempty(self):
+        """Consumer kubeconfig: every field non-empty, user name matches SA, namespace in context."""
         if not self.has_cluster:
             self.skipTest("No cluster reachable (set KUBECONFIG)")
         root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
@@ -134,16 +217,78 @@ def test_consumer_kubeconfig_has_namespace_in_context(self):
         try:
             cfg, proc = self._create_and_get_kubeconfig(root, ns, sa=consumer_sa)
             self.assertEqual(proc.returncode, 0, proc.stderr)
-            self._assert_kubeconfig_valid(cfg, expected_namespace=ns)
+            self._assert_kubeconfig_valid(
+                cfg,
+                expected_namespace=ns,
+                expected_user_name=consumer_sa,
+            )
             self.assertTrue(self._sa_exists(consumer_sa, ns))
         finally:
             self._delete_for_cleanup(root, ns, sa=consumer_sa)
 
-    def test_cli_flags_reflected_in_kubeconfig(self):
+    def test_flag_s_apiserverurl_reflected_in_kubeconfig(self):
+        """-s/--apiserverurl sets cluster server in kubeconfig."""
+        if not self.has_cluster:
+            self.skipTest("No cluster reachable (set KUBECONFIG)")
+        root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+        ns = "kubeplus-test-s-" + str(os.getpid())
+        test_server = "https://api.example.com:6443"
+        try:
+            cfg, proc = self._create_and_get_kubeconfig(
+                root, ns,
+                extra_args=["-s", test_server],
+            )
+            self.assertEqual(proc.returncode, 0, proc.stderr)
+            self._assert_kubeconfig_valid(cfg, expected_server=test_server, expected_namespace=ns)
+        finally:
+            self._delete_for_cleanup(root, ns)
+
+    def test_flag_x_clustername_reflected_in_kubeconfig(self):
+        """-x/--clustername sets context name and cluster name in kubeconfig."""
+        if not self.has_cluster:
+            self.skipTest("No cluster reachable (set KUBECONFIG)")
+        root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+        ns = "kubeplus-test-x-" + str(os.getpid())
+        test_cluster = "my-test-cluster"
+        try:
+            cfg, proc = self._create_and_get_kubeconfig(
+                root, ns,
+                extra_args=["-x", test_cluster],
+            )
+            self.assertEqual(proc.returncode, 0, proc.stderr)
+            self._assert_kubeconfig_valid(
+                cfg,
+                expected_cluster_name=test_cluster,
+                expected_namespace=ns,
+            )
+        finally:
+            self._delete_for_cleanup(root, ns)
+
+    def test_flag_f_filename_uses_custom_output_file(self):
+        """-f/--filename writes kubeconfig to specified file."""
+        if not self.has_cluster:
+            self.skipTest("No cluster reachable (set KUBECONFIG)")
+        root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+        ns = "kubeplus-test-f-" + str(os.getpid())
+        custom_name = "custom-provider-kubeconfig"
+        try:
+            cfg, proc = self._create_and_get_kubeconfig(
+                root, ns,
+                extra_args=["-f", custom_name],
+                output_filename=custom_name,
+            )
+            self.assertEqual(proc.returncode, 0, proc.stderr)
+            self._assert_kubeconfig_valid(cfg, expected_namespace=ns)
+            self.assertTrue(os.path.exists(os.path.join(root, custom_name + ".json")))
+        finally:
+            self._delete_for_cleanup(root, ns)
+
+    def test_flags_s_and_x_combined(self):
+        """-s and -x together: both server and cluster name in kubeconfig."""
         if not self.has_cluster:
             self.skipTest("No cluster reachable (set KUBECONFIG)")
         root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
-        ns = "kubeplus-test-flags-" + str(os.getpid())
+        ns = "kubeplus-test-sx-" + str(os.getpid())
         test_server = "https://api.example.com:6443"
         test_cluster = "my-test-cluster"
         try: