Switch to pathlib

pykube/config.py

@@ -4,6 +4,7 @@
 import base64
 import copy
 import os
+import pathlib
 import tempfile
 
 import yaml
@@ -32,12 +33,14 @@ def from_service_account(
         """
         Construct KubeConfig from in-cluster service account.
         """
-        with open(os.path.join(path, "namespace")) as fp:
-            namespace = fp.read()
+        service_account_dir = pathlib.Path(path)
 
-        with open(os.path.join(path, "token")) as fp:
+        with service_account_dir.joinpath("token").open() as fp:
             token = fp.read()
 
+        with service_account_dir.joinpath("namespace").open() as fp:
+            namespace = fp.read()
+
         host = os.environ.get("PYKUBE_KUBERNETES_SERVICE_HOST")
         if host is None:
             host = os.environ["KUBERNETES_SERVICE_HOST"]
@@ -50,7 +53,9 @@ def from_service_account(
                     "name": "self",
                     "cluster": {
                         "server": "https://" + _join_host_port(host, port),
-                        "certificate-authority": os.path.join(path, "ca.crt"),
+                        "certificate-authority": str(
+                            service_account_dir.joinpath("ca.crt")
+                        ),
                     },
                 }
             ],
@@ -79,15 +84,15 @@ def from_file(cls, filename=None, **kwargs):
         """
         if not filename:
             filename = os.getenv("KUBECONFIG", "~/.kube/config")
-        filename = os.path.expanduser(filename)
-        if not os.path.isfile(filename):
+        filepath = pathlib.Path(filename).expanduser()
+        if not filepath.is_file():
             raise exceptions.PyKubeError(
                 "Configuration file {} not found".format(filename)
             )
-        with open(filename) as f:
+        with filepath.open() as f:
             doc = yaml.safe_load(f.read())
         self = cls(doc, **kwargs)
-        self.filename = filename
+        self.filepath = filepath
         return self
 
     @classmethod
@@ -140,13 +145,13 @@ def set_current_context(self, value):
         self._current_context = value
 
     @property
-    def kubeconfig_file(self):
+    def kubeconfig_path(self):
         """
         Returns the path to kubeconfig file, if it exists
         """
-        if not hasattr(self, "filename"):
+        if not hasattr(self, "filepath"):
             return None
-        return self.filename
+        return self.filepath
 
     @property
     def current_context(self):
@@ -167,7 +172,7 @@ def clusters(self):
                 cs[cr["name"]] = c = copy.deepcopy(cr["cluster"])
                 if "server" not in c:
                     c["server"] = "http://localhost"
-                BytesOrFile.maybe_set(c, "certificate-authority", self.kubeconfig_file)
+                BytesOrFile.maybe_set(c, "certificate-authority", self.kubeconfig_path)
             self._clusters = cs
         return self._clusters
 
@@ -181,8 +186,8 @@ def users(self):
             if "users" in self.doc:
                 for ur in self.doc["users"]:
                     us[ur["name"]] = u = copy.deepcopy(ur["user"])
-                    BytesOrFile.maybe_set(u, "client-certificate", self.kubeconfig_file)
-                    BytesOrFile.maybe_set(u, "client-key", self.kubeconfig_file)
+                    BytesOrFile.maybe_set(u, "client-certificate", self.kubeconfig_path)
+                    BytesOrFile.maybe_set(u, "client-key", self.kubeconfig_path)
             self._users = us
         return self._users
 
@@ -221,10 +226,11 @@ def namespace(self) -> str:
         return self.contexts[self.current_context].get("namespace", "default")
 
     def persist_doc(self):
-        if not self.kubeconfig_file:
+
+        if not self.kubeconfig_path:
             # Config was provided as string, not way to persit it
             return
-        with open(self.kubeconfig_file, "w") as f:
+        with self.kubeconfig_path.open("w") as f:
             yaml.safe_dump(
                 self.doc,
                 f,
@@ -248,46 +254,47 @@ class BytesOrFile:
     """
 
     @classmethod
-    def maybe_set(cls, d, key, kubeconfig_file):
+    def maybe_set(cls, d, key, kubeconfig_path):
         file_key = key
         data_key = "{}-data".format(key)
         if data_key in d:
-            d[file_key] = cls(data=d[data_key], kubeconfig_file=kubeconfig_file)
+            d[file_key] = cls(data=d[data_key], kubeconfig_path=kubeconfig_path)
             del d[data_key]
         elif file_key in d:
-            d[file_key] = cls(filename=d[file_key], kubeconfig_file=kubeconfig_file)
+            d[file_key] = cls(filename=d[file_key], kubeconfig_path=kubeconfig_path)
 
-    def __init__(self, filename=None, data=None, kubeconfig_file=None):
+    def __init__(self, filename=None, data=None, kubeconfig_path=None):
         """
         Creates a new instance of BytesOrFile.
 
         :Parameters:
            - `filename`: A full path to a file
            - `data`: base64 encoded bytes
         """
-        self._filename = None
+        self._path = None
         self._bytes = None
         if filename is not None and data is not None:
             raise TypeError("filename or data kwarg must be specified, not both")
         elif filename is not None:
 
+            path = pathlib.Path(filename)
             # If relative path is given, should be made absolute with respect to the directory of the kube config
             # https://kubernetes.io/docs/concepts/configuration/organize-cluster-access-kubeconfig/#file-references
-            if not os.path.isabs(filename):
-                if kubeconfig_file:
-                    filename = os.path.join(os.path.dirname(kubeconfig_file), filename)
+            if path.is_absolute():
+                if kubeconfig_path:
+                    path = kubeconfig_path.parent.join(path)
                 else:
                     raise exceptions.PyKubeError(
                         "{} passed as relative path, but cannot determine location of kube config".format(
                             filename
                         )
                     )
 
-            if not os.path.isfile(filename):
+            if path.is_file():
                 raise exceptions.PyKubeError(
                     "'{}' file does not exist".format(filename)
                 )
-            self._filename = filename
+            self._path = path
         elif data is not None:
             self._bytes = base64.b64decode(data)
         else:
@@ -297,8 +304,8 @@ def bytes(self):
         """
         Returns the provided data as bytes.
         """
-        if self._filename:
-            with open(self._filename, "rb") as f:
+        if self._path:
+            with self._path.open("rb") as f:
                 return f.read()
         else:
             return self._bytes
@@ -307,8 +314,8 @@ def filename(self):
         """
         Returns the provided data as a file location.
         """
-        if self._filename:
-            return self._filename
+        if self._path:
+            return str(self._path)
         else:
             with tempfile.NamedTemporaryFile(delete=False) as f:
                 f.write(self._bytes)

pykube/console.py

@@ -43,7 +43,7 @@ def main(argv=None):
         if k[0] != "_" and k[0] == k[0].upper():
             context[k] = v
 
-    banner = f"""Pykube v{pykube.__version__}, loaded "{config.filename}" with context "{config.current_context}".
+    banner = f"""Pykube v{pykube.__version__}, loaded "{config.filepath}" with context "{config.current_context}".
 
     Example commands:
       [d.name for d in Deployment.objects(api)]              # get names of deployments in default namespace

tests/test_config.py

@@ -2,6 +2,7 @@
 pykube.config unittests
 """
 import os
+import pathlib
 from pathlib import Path
 from unittest.mock import MagicMock
 
@@ -12,10 +13,9 @@
 from pykube import exceptions
 
 
-GOOD_CONFIG_FILE_PATH = os.path.sep.join(["tests", "test_config.yaml"])
-DEFAULTUSER_CONFIG_FILE_PATH = os.path.sep.join(
-    ["tests", "test_config_default_user.yaml"]
-)
+BASEDIR = Path("tests")
+GOOD_CONFIG_FILE_PATH = BASEDIR / "test_config.yaml"
+DEFAULTUSER_CONFIG_FILE_PATH = BASEDIR / "test_config_default_user.yaml"
 
 
 def test_from_service_account_no_file(tmpdir):
@@ -87,8 +87,8 @@ def test_from_default_kubeconfig(
     kubeconfig_env, expected_path, monkeypatch, kubeconfig
 ):
     mock = MagicMock()
-    mock.return_value = str(kubeconfig)
-    monkeypatch.setattr("os.path.expanduser", mock)
+    mock.return_value.expanduser.return_value = Path(kubeconfig)
+    monkeypatch.setattr(pathlib, "Path", mock)
 
     if kubeconfig_env is None:
         monkeypatch.delenv("KUBECONFIG", raising=False)
@@ -112,7 +112,7 @@ def test_init(self):
         Test Config instance creation.
         """
         # Ensure that a valid creation works
-        self.assertEqual(GOOD_CONFIG_FILE_PATH, self.cfg.filename)
+        self.assertEqual(GOOD_CONFIG_FILE_PATH, self.cfg.filepath)
 
         # Ensure that if a file does not exist the creation fails
         self.assertRaises(

tests/test_http.py

@@ -1,7 +1,7 @@
 """
 pykube.http unittests
 """
-import os
+from pathlib import Path
 from unittest.mock import MagicMock
 
 import pytest
@@ -11,11 +11,13 @@
 from pykube.http import DEFAULT_HTTP_TIMEOUT
 from pykube.http import HTTPClient
 
-GOOD_CONFIG_FILE_PATH = os.path.sep.join(["tests", "test_config_with_context.yaml"])
-CONFIG_WITH_INSECURE_SKIP_TLS_VERIFY = os.path.sep.join(
-    ["tests", "test_config_with_insecure_skip_tls_verify.yaml"]
+
+BASEDIR = Path("tests")
+GOOD_CONFIG_FILE_PATH = BASEDIR / "test_config_with_context.yaml"
+CONFIG_WITH_INSECURE_SKIP_TLS_VERIFY = (
+    BASEDIR / "test_config_with_insecure_skip_tls_verify.yaml"
 )
-CONFIG_WITH_OIDC_AUTH = os.path.sep.join(["tests", "test_config_with_oidc_auth.yaml"])
+CONFIG_WITH_OIDC_AUTH = BASEDIR / "test_config_with_oidc_auth.yaml"
 
 
 def test_http(monkeypatch):

tests/test_session.py

@@ -3,8 +3,8 @@
 """
 import copy
 import logging
-import os
 import tempfile
+from pathlib import Path
 
 from . import TestCase
 
@@ -64,8 +64,9 @@ def test_build_session_auth_provider(self):
 
         _log.info("Built config: %s", self.config)
         try:
-            tmp = tempfile.mktemp()
-            with open(tmp, "w") as f:
+            tmp = Path(tempfile.mktemp())
+
+            with tmp.open("w") as f:
                 f.write(gcloud_content)
 
             # TODO: this no longer works due to refactoring, GCP session handling is now done in KubernetesHTTPAdapter
@@ -75,5 +76,5 @@ def test_build_session_auth_provider(self):
             # self.assertEquals(session.credentials.get('client_id'), 'myclientid')
             # self.assertEquals(session.credentials.get('client_secret'), 'myclientsecret')
         finally:
-            if os.path.exists(tmp):
-                os.remove(tmp)
+            if tmp.exists():
+                tmp.unlink()