Updating PR

Author: anniegracehu

provider-kubeconfig.py

@@ -203,7 +203,7 @@ def _apply_provider_rbac(self, sa, namespace, kubeconfig):
                     {"apiGroups": [""], "resources": ["resourcequotas"], "verbs": ["create", "delete", "deletecollection", "patch", "update"]},
                     {"apiGroups": [""], "resources": ["persistentvolumes", "persistentvolumeclaims"], "verbs": ["get", "watch", "list", "create", "delete", "update", "patch"]},
                 ]
-                # Match original: skip "*" resources (ruleGroup1, ruleGroup5, ruleGroup14)
+                # Skip "*" wildcard resources — not meaningful in the perms inventory
                 all_resources = [
                     res for r in rule_list
                     for res in r.get("resources", [])
@@ -366,16 +366,27 @@ def _extract_kubeconfig(self, sa, namespace, filename, serverip="", kubecfg="",
                     else:
                         time.sleep(2)
 
-                out, _ = run_command(" kubectl get secret " + sa + " -n " + namespace + " -o json " + kubecfg)
-                json_out = json.loads(out or "{}")
-                ca_cert = json_out.get("data", {}).get("ca.crt", "").strip()
+                out, err = run_command(" kubectl get secret " + sa + " -n " + namespace + " -o json " + kubecfg)
+                if not out:
+                    raise RuntimeError(
+                        f"Failed to fetch secret {sa!r} in ns {namespace!r}: {err}"
+                    )
+                json_out = json.loads(out)
+                ca_cert = json_out["data"]["ca.crt"].strip()
 
                 if serverip:
                     server = serverip if "https" in serverip else "https://" + serverip
                 else:
-                    out2, _ = run_command("kubectl -n default get endpoints kubernetes " + kubecfg + " | awk '{print $2}' | grep -v ENDPOINTS")
+                    out2, _ = run_command(
+                        "kubectl -n default get endpoints kubernetes " + kubecfg
+                        + " | awk '{print $2}' | grep -v ENDPOINTS"
+                    )
                     server = out2.strip() if out2 else ""
                     server = "https://" + server if server else ""
+                if not server or server.rstrip("/") == "https:":
+                    raise RuntimeError(
+                        "Could not determine API server endpoint; pass -s/--apiserverurl"
+                    )
 
                 self._create_kubecfg_file(sa, namespace, filename, token, ca_cert, server, kubecfg, cluster_name)
 

tests/test_provider_kubeconfig.py

@@ -7,6 +7,7 @@
 import subprocess
 import sys
 import unittest
+import uuid
 
 
 def _run_command(cmd):
@@ -20,7 +21,7 @@ def _cluster_available(kubeconfig=""):
     path = os.path.expanduser(kubeconfig).strip() if kubeconfig else ""
     k_opt = " --kubeconfig=" + path if path else ""
     out, err = _run_command("kubectl get ns default" + k_opt)
-    return out and "default" in out and "NotFound" not in (err or "")
+    return out and "default" in out and "NotFound" not in err
 
 
 SCRIPT = "provider-kubeconfig.py"
@@ -50,11 +51,8 @@ def test_help_shows_all_actions_flags_and_namespace(self):
         )
         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")
         for elem in HELP_ELEMENTS:
             self.assertIn(elem, out, f"Help should mention {elem}")
-        self.assertIn("namespace", out.lower())
 
     def test_update_without_permissionfile_exits_with_error(self):
         """update action without -p exits with code 1."""
@@ -85,39 +83,45 @@ def setUp(self):
         if not self.has_cluster:
             self.skipTest("No cluster reachable (set KUBECONFIG)")
 
-    def _create_and_get_kubeconfig(self, root, ns, sa="kubeplus-saas-provider", extra_args=None, output_filename=None):
+    def _create_and_get_kubeconfig(self, 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":
             create_args += ["-c", sa]
         if extra_args:
             create_args += extra_args
         proc = subprocess.run(
-            [sys.executable, os.path.join(root, SCRIPT)] + create_args + self.kubeconfig_arg,
-            cwd=root, capture_output=True, text=True, timeout=120,
+            [sys.executable, os.path.join(ROOT, SCRIPT)] + create_args + self.kubeconfig_arg,
+            cwd=ROOT, capture_output=True, text=True, timeout=120,
         )
         if proc.returncode != 0:
             return None, proc
         filename = output_filename or (sa + ".json")
         if not filename.endswith(".json"):
             filename += ".json"
-        kubeconfig_path = os.path.join(root, filename)
+        kubeconfig_path = os.path.join(ROOT, filename)
         if not os.path.exists(kubeconfig_path):
-            return None, proc
+            raise AssertionError(
+                f"Script exited 0 but kubeconfig not written: {kubeconfig_path}"
+            )
         with open(kubeconfig_path, "r", encoding="utf-8") as fp:
             cfg = json.load(fp)
         return cfg, proc
 
-    def _delete_for_cleanup(self, root, ns, sa="kubeplus-saas-provider", filename=None):
-        """Delete k8s resources and local files. Pass filename when -f was used (e.g. custom-provider-kubeconfig)."""
+    def _delete_for_cleanup(self, ns, sa="kubeplus-saas-provider", filename=None):
+        """Delete k8s resources, local files, and test namespace."""
         delete_args = ["delete", ns]
         if sa != "kubeplus-saas-provider":
             delete_args += ["-c", sa]
         if filename:
             delete_args += ["-f", filename]
         subprocess.run(
-            [sys.executable, os.path.join(root, SCRIPT)] + delete_args + self.kubeconfig_arg,
-            cwd=root, capture_output=True, timeout=60,
+            [sys.executable, os.path.join(ROOT, SCRIPT)] + delete_args + self.kubeconfig_arg,
+            cwd=ROOT, capture_output=True, timeout=60,
+        )
+        _run_command(
+            "kubectl delete namespace " + ns
+            + " --ignore-not-found --wait=false" + self.kubeconfig_flag
         )
 
     def _assert_kubeconfig_valid(
@@ -174,13 +178,13 @@ def _assert_kubeconfig_valid(
 
     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 "")
+        return out and sa in out and "NotFound" not in err
 
     def test_provider_kubeconfig_all_fields_nonempty(self):
         """Provider kubeconfig: every field that should exist is non-empty."""
-        ns = "kubeplus-test-prov-" + str(os.getpid())
+        ns = "kubeplus-test-prov-" + uuid.uuid4().hex[:8]
         try:
-            cfg, proc = self._create_and_get_kubeconfig(ROOT, ns)
+            cfg, proc = self._create_and_get_kubeconfig(ns)
             self.assertEqual(proc.returncode, 0, proc.stderr)
             self._assert_kubeconfig_valid(
                 cfg,
@@ -189,14 +193,14 @@ def test_provider_kubeconfig_all_fields_nonempty(self):
             )
             self.assertTrue(self._sa_exists("kubeplus-saas-provider", ns))
         finally:
-            self._delete_for_cleanup(ROOT, ns)
+            self._delete_for_cleanup(ns)
 
     def test_consumer_kubeconfig_all_fields_nonempty(self):
         """Consumer kubeconfig: every field non-empty, user name matches SA, namespace in context."""
-        ns = "kubeplus-test-cons-" + str(os.getpid())
+        ns = "kubeplus-test-cons-" + uuid.uuid4().hex[:8]
         consumer_sa = "test-consumer-sa"
         try:
-            cfg, proc = self._create_and_get_kubeconfig(ROOT, ns, sa=consumer_sa)
+            cfg, proc = self._create_and_get_kubeconfig(ns, sa=consumer_sa)
             self.assertEqual(proc.returncode, 0, proc.stderr)
             self._assert_kubeconfig_valid(
                 cfg,
@@ -205,29 +209,29 @@ def test_consumer_kubeconfig_all_fields_nonempty(self):
             )
             self.assertTrue(self._sa_exists(consumer_sa, ns))
         finally:
-            self._delete_for_cleanup(ROOT, ns, sa=consumer_sa)
+            self._delete_for_cleanup(ns, sa=consumer_sa)
 
     def test_flag_s_apiserverurl_reflected_in_kubeconfig(self):
         """-s/--apiserverurl sets cluster server in kubeconfig."""
-        ns = "kubeplus-test-s-" + str(os.getpid())
+        ns = "kubeplus-test-s-" + uuid.uuid4().hex[:8]
         test_server = "https://api.example.com:6443"
         try:
             cfg, proc = self._create_and_get_kubeconfig(
-                ROOT, ns,
+                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)
+            self._delete_for_cleanup(ns)
 
     def test_flag_x_clustername_reflected_in_kubeconfig(self):
         """-x/--clustername sets context name and cluster name in kubeconfig."""
-        ns = "kubeplus-test-x-" + str(os.getpid())
+        ns = "kubeplus-test-x-" + uuid.uuid4().hex[:8]
         test_cluster = "my-test-cluster"
         try:
             cfg, proc = self._create_and_get_kubeconfig(
-                ROOT, ns,
+                ns,
                 extra_args=["-x", test_cluster],
             )
             self.assertEqual(proc.returncode, 0, proc.stderr)
@@ -237,32 +241,32 @@ def test_flag_x_clustername_reflected_in_kubeconfig(self):
                 expected_namespace=ns,
             )
         finally:
-            self._delete_for_cleanup(ROOT, ns)
+            self._delete_for_cleanup(ns)
 
     def test_flag_f_filename_uses_custom_output_file(self):
         """-f/--filename writes kubeconfig to specified file."""
-        ns = "kubeplus-test-f-" + str(os.getpid())
+        ns = "kubeplus-test-f-" + uuid.uuid4().hex[:8]
         custom_name = "custom-provider-kubeconfig"
         try:
             cfg, proc = self._create_and_get_kubeconfig(
-                ROOT, ns,
+                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, filename=custom_name)
+            self._delete_for_cleanup(ns, filename=custom_name)
 
     def test_flags_s_and_x_combined(self):
         """-s and -x together: both server and cluster name in kubeconfig."""
-        ns = "kubeplus-test-sx-" + str(os.getpid())
+        ns = "kubeplus-test-sx-" + uuid.uuid4().hex[:8]
         test_server = "https://api.example.com:6443"
         test_cluster = "my-test-cluster"
         try:
             cfg, proc = self._create_and_get_kubeconfig(
-                ROOT, ns,
+                ns,
                 extra_args=["-s", test_server, "-x", test_cluster],
             )
             self.assertEqual(proc.returncode, 0, proc.stderr)
@@ -273,19 +277,19 @@ def test_flags_s_and_x_combined(self):
                 expected_namespace=ns,
             )
         finally:
-            self._delete_for_cleanup(ROOT, ns)
+            self._delete_for_cleanup(ns)
 
     def test_consumer_cannot_create_pod_in_other_namespace(self):
         """
         Consumer kubeconfig: verify create/delete in other namespaces is forbidden.
         Consumer RBAC should restrict operations; creating a pod in another ns should fail.
         """
-        ns = "kubeplus-test-restrict-" + str(os.getpid())
-        other_ns = "kubeplus-test-other-" + str(os.getpid())
+        ns = "kubeplus-test-restrict-" + uuid.uuid4().hex[:8]
+        other_ns = "kubeplus-test-other-" + uuid.uuid4().hex[:8]
         consumer_sa = "test-consumer-restrict"
         kubeconfig_path = os.path.join(ROOT, consumer_sa + ".json")
         try:
-            cfg, proc = self._create_and_get_kubeconfig(ROOT, ns, sa=consumer_sa)
+            cfg, proc = self._create_and_get_kubeconfig(ns, sa=consumer_sa)
             self.assertEqual(proc.returncode, 0, proc.stderr)
             self._assert_kubeconfig_valid(cfg, expected_namespace=ns, expected_user_name=consumer_sa)
 
@@ -305,7 +309,7 @@ def test_consumer_cannot_create_pod_in_other_namespace(self):
             )
         finally:
             _run_command("kubectl delete namespace " + other_ns + self.kubeconfig_flag + " 2>/dev/null")
-            self._delete_for_cleanup(ROOT, ns, sa=consumer_sa)
+            self._delete_for_cleanup(ns, sa=consumer_sa)
 
 
 if __name__ == "__main__":