Fix reloading service account credentials (#580)

Author: jacobtomlinson

kr8s/_auth.py

@@ -78,7 +78,7 @@ async def reauthenticate(self) -> None:
             else:
                 if self._kubeconfig_path_or_dict is not False:
                     await self._load_kubeconfig()
-                if self._serviceaccount and not self.server:
+                if self._serviceaccount:
                     await self._load_service_account()
             if not self.server:
                 raise ValueError("Unable to find valid credentials")
@@ -264,10 +264,11 @@ async def _load_service_account(self) -> None:
         self._serviceaccount = os.path.expanduser(self._serviceaccount)
         if not os.path.isdir(self._serviceaccount):
             return
-        self.server = self._format_server_address(
-            os.environ["KUBERNETES_SERVICE_HOST"],
-            os.environ["KUBERNETES_SERVICE_PORT"],
-        )
+        if not self.server:
+            self.server = self._format_server_address(
+                os.environ["KUBERNETES_SERVICE_HOST"],
+                os.environ["KUBERNETES_SERVICE_PORT"],
+            )
         async with await anyio.open_file(
             os.path.join(self._serviceaccount, "token")
         ) as f:

kr8s/tests/test_auth.py

@@ -243,7 +243,7 @@ def test_kubeconfig_isdir_fail(tmp_path):
         kr8s.api(kubeconfig=tmp_path)
 
 
-async def test_service_account(serviceaccount):
+async def test_service_account(serviceaccount, k8s_token):
     api = await kr8s.asyncio.api(
         serviceaccount=serviceaccount, kubeconfig="/no/file/here"
     )
@@ -252,10 +252,18 @@ async def test_service_account(serviceaccount):
     serviceaccount = Path(serviceaccount)
     assert api.auth.server
     assert api.auth.token == (serviceaccount / "token").read_text()
-    assert str(serviceaccount) in api.auth.server_ca_file
+    assert api.auth.token == k8s_token
+    assert api.auth.server_ca_file
+    assert str(serviceaccount) in str(api.auth.server_ca_file)
     assert "BEGIN CERTIFICATE" in Path(api.auth.server_ca_file).read_text()
     assert api.auth.namespace == (serviceaccount / "namespace").read_text()
 
+    (serviceaccount / "token").write_text("foo")
+    await api.reauthenticate()
+    assert api.auth.token == "foo"
+
+    (serviceaccount / "token").write_text(k8s_token)
+
 
 async def test_service_account_with_kubeconfig_namespace(serviceaccount):
     kubeconfig = await KubeConfig(