Build tenant based domain url

Author: fewa123few

splunk_sdk/base_client.py

@@ -84,6 +84,55 @@ def _wrapper(self, *args, **kwargs):
     return _wrapper
 
 
+def build_url(context: Context, route="", add_scheme=False, omit_tenant=False) -> str:
+    """
+    Builds a partial (without scheme) or full URL from param "route" and based on tenant specification in the context.
+    It supports both scenarios of adopting tenant-scoped hostname or legacy hostname. Default mode is legacy hostname.
+    If context.tenant_scoped is True, tenant-scoped hostname is selected. Non-system tenant will be added to the domain.
+    Region will be added to the domain for system tenant.
+    If context.tenant_scoped is False, there will be no additional string added to domain name (legacy mode).
+
+    :param context: Context object which provides all necessary information
+    :param route: API path to call
+    :param add_scheme: set to True if needs to add scheme, default to False
+    :param omit_tenant: set to True if tenant name need not to be specified in URL
+    :return: The full URL.
+    """
+    if route is None:
+        route = ""
+    route = route.strip()
+    if route.startswith("/"):
+        route = route[1:]
+    if context.host is None or not context.host.strip():
+        raise ValueError("Invalid value for context.host")
+    host = context.host.strip()
+    if context.port is not None and context.port:
+        host = f"{host}:{context.port}"
+    if context.tenant is None or not context.tenant.strip():
+        raise ValueError("Invalid value for context.tenant")
+    tenant = context.tenant.strip()
+    if not omit_tenant or tenant == "system":
+        route = f"{tenant}/{route}"
+    if context.tenant_scoped is True:
+        if tenant == "system":
+            if context.region is None or not context.region.strip():
+                raise ValueError("Invalid value for context.region")
+            region = context.region.strip()
+            url = f"region-{region}.{host}/{route}"
+        else:
+            url = f"{tenant}.{host}/{route}"
+    else:
+        url = f"{host}/{route}"
+
+    if add_scheme is True:
+        if context.scheme is None or not context.scheme.strip():
+            raise ValueError("Invalid value for context.scheme")
+        scheme = context.scheme.strip()
+        return f"{scheme}://{url}"
+    else:
+        return url
+
+
 class BaseClient(object):
     """
     The BaseClient class encapsulates conventions, authorization, and URL handling

test/test_base_client.py

@@ -18,7 +18,7 @@
 from test.fixtures import get_client_auth_manager_scoped as client_auth_manager_scoped  # NOQA
 from test.fixtures import get_service_principal_auth_manager as service_principal_auth_manager  # NOQA
 from splunk_sdk.common.context import Context
-from splunk_sdk.base_client import BaseClient
+from splunk_sdk.base_client import BaseClient, build_url
 from splunk_sdk.identity import Identity as IdentityAndAccessControl
 
 @pytest.mark.usefixtures("pkce_auth_manager")  # NOQA
@@ -193,3 +193,33 @@ def test_base_client_instance_with_sp_auth(service_principal_auth_manager):
     assert (default_config is not None)
     assert (base_client is not None)
     _assert_sp_credentials_auth_context(base_client.auth_manager.context)
+
+
+def test_build_url():
+    # normal tenant
+    context = Context(host="test.splunk.com", tenant="tttt", tenant_scoped=True, scheme="http")
+    assert ("http://tttt.test.splunk.com/tttt/v2/test" == build_url(context, "/v2/test", True))
+    context = Context(host="test.splunk.com", tenant="tttt", tenant_scoped=False, port=9999)
+    assert ("test.splunk.com:9999/tttt/v2/test" == build_url(context, "v2/test", False))
+    context = Context(host="test.splunk.com", tenant="tttt", tenant_scoped=False, port=9999)
+    assert ("test.splunk.com:9999/v2/test" == build_url(context, "v2/test", False, True))
+    # system tenant
+    context = Context(host="test.splunk.com", tenant="system", tenant_scoped=True, region="iad10")
+    assert ("region-iad10.test.splunk.com/system/v2/test" == build_url(context, "/v2/test"))
+    context = Context(host="test.splunk.com", tenant="system", tenant_scoped=False)
+    assert ("test.splunk.com/system/v2/test" == build_url(context, "v2/test"))
+    context = Context(host="test.splunk.com", tenant="system", tenant_scoped=False)
+    assert ("https://test.splunk.com/system/v2/test" == build_url(context, "v2/test", True))
+    context = Context(host="test.splunk.com", tenant="system", tenant_scoped=False)
+    assert ("https://test.splunk.com/system/v2/test" == build_url(context, "v2/test", True, True))
+    # input sanity test
+    context = Context(host=" test.splunk.com ", tenant=" system ", tenant_scoped=True, region=" iad10 ")
+    assert ("region-iad10.test.splunk.com/system/v2/test" == build_url(context, "   v2/test "))
+    with pytest.raises(ValueError):
+        build_url(Context(host=None, tenant="system", tenant_scoped=False), "v2/test")
+    with pytest.raises(ValueError):
+        build_url(Context(tenant=" ", tenant_scoped=True), "v2/test")
+    with pytest.raises(ValueError):
+        build_url(Context(tenant_scoped=True), "v2/test")
+    with pytest.raises(ValueError):
+        build_url(Context(tenant_scoped=True, scheme=""), "v2/test", True)