Join URL path components without os.path/posixpath

pykube/http.py

@@ -4,7 +4,6 @@
 import datetime
 import json
 import os
-import posixpath
 import shlex
 import subprocess
 from typing import Optional
@@ -23,7 +22,7 @@
 from urllib.parse import urlparse
 
 from .exceptions import HTTPError
-from .utils import jsonpath_installed, jsonpath_parse
+from .utils import jsonpath_installed, jsonpath_parse, join_url_path
 from .config import KubeConfig
 
 from . import __version__
@@ -139,7 +138,9 @@ def _setup_request_auth(self, config, request, kwargs):
                 parsed_out = json.loads(output)
                 token = parsed_out["status"]["token"]
             else:
-                raise NotImplementedError(f'auth exec api version {api_version} not implemented')
+                raise NotImplementedError(
+                    f"auth exec api version {api_version} not implemented"
+                )
 
             request.headers["Authorization"] = "Bearer {}".format(token)
             return None
@@ -286,10 +287,8 @@ def get_kwargs(self, **kwargs) -> dict:
                 if namespace:
                     bits.extend(["namespaces", namespace])
         url = kwargs.get("url", "")
-        if url.startswith("/"):
-            url = url[1:]
         bits.append(url)
-        kwargs["url"] = self.url + posixpath.join(*bits)
+        kwargs["url"] = self.url + join_url_path(*bits, join_empty=True)
         if "timeout" not in kwargs:
             # apply default HTTP timeout
             kwargs["timeout"] = self.timeout

pykube/objects.py

@@ -1,6 +1,5 @@
 import copy
 import json
-import os.path as op
 from inspect import getmro
 from typing import Optional
 from typing import Type
@@ -11,6 +10,7 @@
 from .mixins import ReplicatedMixin
 from .mixins import ScalableMixin
 from .query import Query
+from .utils import join_url_path
 from .utils import obj_merge
 
 
@@ -96,9 +96,8 @@ def api_kwargs(self, **kwargs):
         else:
             subresource = kwargs.pop("subresource", None) or ""
             operation = kwargs.pop("operation", "")
-            kw["url"] = op.normpath(
-                op.join(self.endpoint, self.name, subresource, operation)
-            )
+            kw["url"] = join_url_path(self.endpoint, self.name, subresource, operation)
+
         params = kwargs.pop("params", None)
         if params is not None:
             query_string = urlencode(params)

pykube/utils.py

@@ -1,4 +1,5 @@
 import re
+from typing import List
 
 try:
     from jsonpath_ng import parse as jsonpath
@@ -62,3 +63,27 @@ def repl(m):
         return jsonpath(path).find(obj)[0].value
 
     return re.sub(r"(\{([^\}]*)\})", repl, template)
+
+
+def join_url_path(*components, join_empty: bool = False) -> str:
+    """Join given URL path components and return absolute path starting with '/'."""
+    new_comps: List[str] = []
+    for comp in components:
+        comp = comp.strip("/")
+        if comp in ("", "."):
+            continue
+        else:
+            new_comps.append(comp)
+
+    if components and components[-1] == "" and join_empty:
+        trailing_slash = True
+    elif components and components[-1] == "/":
+        trailing_slash = True
+    else:
+        trailing_slash = False
+
+    if trailing_slash:
+        new_comps.append("")
+
+    path = "/".join(new_comps)
+    return "/" + path

tests/test_utils.py

@@ -1,3 +1,4 @@
+from pykube.utils import join_url_path
 from pykube.utils import obj_merge
 
 
@@ -28,3 +29,16 @@ def test_obj_merge():
     assert obj_merge(
         {"a": {"b": [1, 2]}}, {"a": {"b": [3, 4, 5], "c": [1, 2]}}, is_strategic=False
     ) == {"a": {"b": [1, 2]}}
+
+
+def test_join_url_path():
+    assert join_url_path() == "/"
+    assert join_url_path("") == "/"
+    assert join_url_path("", "/") == "/"
+    assert join_url_path("first", "") == "/first"
+    assert join_url_path("first", "/") == "/first/"
+    assert join_url_path("first", "second") == "/first/second"
+    assert join_url_path("/first", "second/") == "/first/second"
+    assert join_url_path("first", "second", "") == "/first/second"
+    assert join_url_path("first", "/", "second", "", "") == "/first/second"
+    assert join_url_path("/first", "second", "", join_empty=True) == "/first/second/"