fix: ConnectionManager handles trailing slash in bash_url

Author: DanielYang59

src/keycloak/connection.py

@@ -24,11 +24,6 @@
 
 from __future__ import annotations
 
-try:
-    from urllib.parse import urljoin
-except ImportError:  # pragma: no cover
-    from urlparse import urljoin  # pyright: ignore[reportMissingImports]
-
 from typing import Any
 
 import httpx
@@ -167,6 +162,13 @@ def base_url(self) -> str | None:
     def base_url(self, value: str | None) -> None:
         self._base_url = value
 
+    def _build_url(self, path: str) -> str:
+        """Join the base_url and path, and handle trailing slashes."""
+        if not self.base_url or path.startswith(("http://", "https://")):
+            return path
+
+        return f"{self.base_url.rstrip('/')}/{path.lstrip('/')}"
+
     @property
     def timeout(self) -> int | None:
         """
@@ -334,7 +336,7 @@ def raw_get(self, path: str, **kwargs: Any) -> Response:  # noqa: ANN401
             raise AttributeError(msg)
         try:
             return self._s.get(
-                urljoin(self.base_url, path),
+                self._build_url(path),
                 params=kwargs,
                 headers=self.headers,
                 timeout=self.timeout,
@@ -364,7 +366,7 @@ def raw_post(self, path: str, data: dict | str | MultipartEncoder, **kwargs: Any
             raise AttributeError(msg)
         try:
             return self._s.post(
-                urljoin(self.base_url, path),
+                self._build_url(path),
                 params=kwargs,
                 data=data,
                 headers=self.headers,
@@ -396,7 +398,7 @@ def raw_put(self, path: str, data: dict | str | MultipartEncoder, **kwargs: Any)
 
         try:
             return self._s.put(
-                urljoin(self.base_url, path),
+                self._build_url(path),
                 params=kwargs,
                 data=data,
                 headers=self.headers,
@@ -428,7 +430,7 @@ def raw_delete(self, path: str, data: dict | None = None, **kwargs: Any) -> Resp
 
         try:
             return self._s.delete(
-                urljoin(self.base_url, path),
+                self._build_url(path),
                 params=kwargs,
                 data=data or {},
                 headers=self.headers,
@@ -458,7 +460,7 @@ async def a_raw_get(self, path: str, **kwargs: Any) -> AsyncResponse:  # noqa: A
 
         try:
             return await self.async_s.get(
-                urljoin(self.base_url, path),
+                self._build_url(path),
                 params=self._filter_query_params(kwargs),
                 headers=self.headers,
                 timeout=self.timeout,
@@ -493,7 +495,7 @@ async def a_raw_post(
         try:
             return await self.async_s.request(
                 method="POST",
-                url=urljoin(self.base_url, path),
+                url=self._build_url(path),
                 params=self._filter_query_params(kwargs),
                 **self._prepare_httpx_request_content(data),
                 headers=self.headers,
@@ -528,7 +530,7 @@ async def a_raw_put(
 
         try:
             return await self.async_s.put(
-                urljoin(self.base_url, path),
+                self._build_url(path),
                 params=self._filter_query_params(kwargs),
                 **self._prepare_httpx_request_content(data),
                 headers=self.headers,
@@ -564,7 +566,7 @@ async def a_raw_delete(
         try:
             return await self.async_s.request(
                 method="DELETE",
-                url=urljoin(self.base_url, path),
+                url=self._build_url(path),
                 **self._prepare_httpx_request_content(data or {}),
                 params=self._filter_query_params(kwargs),
                 headers=self.headers,

src/keycloak/keycloak_admin.py

@@ -2386,7 +2386,7 @@ def get_group_by_path(self, path: str) -> dict:
         :return: Keycloak server response (GroupRepresentation)
         :rtype: dict
         """
-        params_path = {"realm-name": self.connection.realm_name, "path": path}
+        params_path = {"realm-name": self.connection.realm_name, "path": path.lstrip("/")}
         data_raw = self.connection.raw_get(
             urls_patterns.URL_ADMIN_GROUP_BY_PATH.format(**params_path),
         )
@@ -9276,7 +9276,7 @@ async def a_get_group_by_path(self, path: str) -> dict:
         :return: Keycloak server response (GroupRepresentation)
         :rtype: dict
         """
-        params_path = {"realm-name": self.connection.realm_name, "path": path}
+        params_path = {"realm-name": self.connection.realm_name, "path": path.lstrip("/")}
         data_raw = await self.connection.a_raw_get(
             urls_patterns.URL_ADMIN_GROUP_BY_PATH.format(**params_path),
         )

tests/test_connection.py

@@ -95,3 +95,28 @@ def test_counter_part() -> None:
             continue
 
         assert async_method[2:] in sync_methods
+
+
+def test_build_url() -> None:
+    """Test URL building and sub-path preservation."""
+    # Scenario 1: Base URL WITHOUT a trailing slash
+    cm = ConnectionManager(base_url="http://test.test/auth")
+
+    assert cm._build_url("realms/master") == "http://test.test/auth/realms/master"
+    assert cm._build_url("/realms/master") == "http://test.test/auth/realms/master"
+
+    # Scenario 2: Base URL WITH a trailing slash
+    cm_slashed = ConnectionManager(base_url="http://test.test/auth/")
+
+    assert cm_slashed._build_url("realms/master") == "http://test.test/auth/realms/master"
+    assert cm_slashed._build_url("/realms/master") == "http://test.test/auth/realms/master"
+
+    # Scenario 3: Path is already an absolute URL
+    assert cm._build_url("http://absolute.test/realms") == "http://absolute.test/realms"
+    assert cm._build_url("https://absolute.test/realms") == "https://absolute.test/realms"
+
+    # Scenario 4: Empty base URL
+    cm_empty = ConnectionManager(base_url="")
+
+    assert cm_empty._build_url("realms/master") == "realms/master"
+    assert cm_empty._build_url("/realms/master") == "/realms/master"