fix: validate, clean base url

Author: jentyk

mpt_api_client/http/async_client.py

@@ -11,6 +11,7 @@
 from mpt_api_client.constants import APPLICATION_JSON
 from mpt_api_client.exceptions import MPTError, transform_http_status_exception
 from mpt_api_client.http.client import json_to_file_payload
+from mpt_api_client.http.client_utils import validate_base_url
 from mpt_api_client.http.types import (
     HeaderTypes,
     QueryParam,
@@ -38,13 +39,7 @@ def __init__(
                 "argument to MPTClient."
             )
 
-        base_url = base_url or os.getenv("MPT_URL")
-        if not base_url:
-            raise ValueError(
-                "Base URL is required. "
-                "Set it up as env variable MPT_URL or pass it as `base_url` "
-                "argument to MPTClient."
-            )
+        base_url = validate_base_url(base_url or os.getenv("MPT_URL"))
         base_headers = {
             "User-Agent": "swo-marketplace-client/1.0",
             "Authorization": f"Bearer {api_token}",

mpt_api_client/http/client.py

@@ -14,6 +14,7 @@
     MPTError,
     transform_http_status_exception,
 )
+from mpt_api_client.http.client_utils import validate_base_url
 from mpt_api_client.http.types import (
     HeaderTypes,
     QueryParam,
@@ -51,13 +52,7 @@ def __init__(
                 "argument to MPTClient."
             )
 
-        base_url = base_url or os.getenv("MPT_URL")
-        if not base_url:
-            raise ValueError(
-                "Base URL is required. "
-                "Set it up as env variable MPT_URL or pass it as `base_url` "
-                "argument to MPTClient."
-            )
+        base_url = validate_base_url(base_url or os.getenv("MPT_URL"))
         base_headers = {
             "User-Agent": "swo-marketplace-client/1.0",
             "Authorization": f"Bearer {api_token}",

mpt_api_client/http/client_utils.py

@@ -0,0 +1,25 @@
+import re
+from typing import Any
+from urllib.parse import urlsplit, urlunparse
+
+INVALID_ENV_URL_MESSAGE = "Invalid base URL. Expected scheme://host[:port]"
+PATH_TO_REMOVE_RE = re.compile(r"^/$|^/public/?$|^/public/v1/?$")
+
+
+def validate_base_url(base_url: Any) -> str:
+    """Validate base url."""
+    if not base_url or not isinstance(base_url, str):
+        raise ValueError(
+            "Base URL is required. "
+            "Set it up as env variable MPT_URL or pass it as `base_url` "
+            "argument to MPTClient."
+        )
+    split_result = urlsplit(base_url, scheme="https")
+    if split_result.scheme and split_result.hostname:
+        host = (
+            f"[{split_result.hostname}]" if ":" in split_result.hostname else split_result.hostname
+        )
+        port = f":{split_result.port}" if split_result.port else ""
+        path = PATH_TO_REMOVE_RE.sub("", split_result.path)
+        return urlunparse((split_result.scheme, f"{host}{port}", path, "", "", ""))
+    raise ValueError(INVALID_ENV_URL_MESSAGE)

tests/unit/http/test_client_utils.py

@@ -0,0 +1,42 @@
+import pytest
+
+from mpt_api_client.http.client_utils import validate_base_url
+
+
+@pytest.mark.parametrize(
+    ("input_url", "expected"),
+    [
+        ("//[2001:db8:85a3::8a2e:370:7334]:80/a", "https://[2001:db8:85a3::8a2e:370:7334]:80/a"),
+        ("//example.com", "https://example.com"),
+        ("http://example.com", "http://example.com"),
+        ("http://example.com:88/something/else", "http://example.com:88/something/else"),
+        ("http://user@example.com:88/", "http://example.com:88"),
+        ("http://user:pass@example.com:88/", "http://example.com:88"),
+        ("http://example.com/public", "http://example.com"),
+        ("http://example.com/public/", "http://example.com"),
+        ("http://example.com/public/else", "http://example.com/public/else"),
+        ("http://example.com/public/v1", "http://example.com"),
+        ("http://example.com/public/v1/", "http://example.com"),
+        ("http://example.com/else/public", "http://example.com/else/public"),
+        ("http://example.com/elsepublic", "http://example.com/elsepublic"),
+    ],
+)
+def test_protocol_and_host(input_url, expected):
+    result = validate_base_url(input_url)
+
+    assert result == expected
+
+
+@pytest.mark.parametrize(
+    "input_url",
+    [
+        "",
+        "http//example.com",
+        "://example.com",
+        "http:example.com",
+        "http:/example.com",
+    ],
+)
+def test_protocol_and_host_error(input_url):
+    with pytest.raises(ValueError):
+        validate_base_url(input_url)