feat/(token): token endpoint is now configurable

Author: Divan009

openfga_sdk/credentials.py

@@ -9,8 +9,8 @@
 
    NOTE: This file was auto generated by OpenAPI Generator (https://openapi-generator.tech). DO NOT EDIT.
 """
-
-from urllib.parse import urlparse
+import re
+from urllib.parse import urlparse, urlunparse
 
 from openfga_sdk.exceptions import ApiValueError
 
@@ -160,6 +160,31 @@ def configuration(self, value):
         """
         self._configuration = value
 
+    def _parse_issuer(self, issuer: str):
+        default_endpoint_path = '/oauth/token'
+
+        parsed_url = urlparse(issuer.strip())
+
+        try:
+            parsed_url.port
+        except ValueError as e:
+            raise ApiValueError(e)
+
+        if parsed_url.netloc is None and parsed_url.path is None:
+            raise ApiValueError(f"Invalid issuer")
+
+        if parsed_url.scheme == "":
+            parsed_url = urlparse(f"https://{issuer}")
+        elif parsed_url.scheme not in ("http", "https"):
+            raise ApiValueError(f"Invalid issuer scheme {parsed_url.scheme} must be HTTP or HTTPS")
+
+        if parsed_url.path in ("", "/"):
+            parsed_url = parsed_url._replace(path=default_endpoint_path)
+
+        valid_url = urlunparse(parsed_url)
+
+        return valid_url
+
     def validate_credentials_config(self):
         """
         Check whether credentials configuration is valid
@@ -190,15 +215,6 @@ def validate_credentials_config(self):
                     "configuration `{}` requires client_id, client_secret, api_audience and api_issuer defined for client_credentials method."
                 )
             # validate token issuer
-            combined_url = "https://" + self.configuration.api_issuer
-            parsed_url = None
-            try:
-                parsed_url = urlparse(combined_url)
-            except ValueError:
-                raise ApiValueError(
-                    f"api_issuer `{self.configuration.api_issuer}` is invalid"
-                )
-            if parsed_url.netloc == "":
-                raise ApiValueError(
-                    f"api_issuer `{self.configuration.api_issuer}` is invalid"
-                )
+
+            parsed_issuer_url = self._parse_issuer(self.configuration.api_issuer)
+

test/credentials_test.py

@@ -13,7 +13,9 @@
 from unittest import IsolatedAsyncioTestCase
 
 import openfga_sdk
+from openfga_sdk import ApiValueError
 from openfga_sdk.credentials import CredentialConfiguration, Credentials
+import unittest
 
 
 class TestCredentials(IsolatedAsyncioTestCase):
@@ -172,3 +174,63 @@ def test_configuration_client_credentials_missing_api_audience(self):
         )
         with self.assertRaises(openfga_sdk.ApiValueError):
             credential.validate_credentials_config()
+
+
+
+class TestCredentialsIssuer(unittest.TestCase):
+    def setUp(self):
+        # Setup a basic configuration that can be modified per test case
+        self.configuration = CredentialConfiguration(api_issuer="https://example.com")
+        self.credentials = Credentials(method="client_credentials", configuration=self.configuration)
+
+    def test_valid_issuer_https(self):
+        # Test a valid HTTPS URL
+        self.configuration.api_issuer = "issuer.fga.example	"
+        result = self.credentials._parse_issuer(self.configuration.api_issuer)
+        self.assertEqual(result, "https://issuer.fga.example/oauth/token")
+
+    def test_valid_issuer_with_oauth_endpoint_https(self):
+        # Test a valid HTTPS URL
+        self.configuration.api_issuer = "https://example.com/oauth/token"
+        result = self.credentials._parse_issuer(self.configuration.api_issuer)
+        self.assertEqual(result, "https://example.com/oauth/token")
+
+    def test_valid_issuer_with_some_endpoint_https(self):
+        # Test a valid HTTPS URL
+        self.configuration.api_issuer = "https://example.com/oauth/some/endpoint"
+        result = self.credentials._parse_issuer(self.configuration.api_issuer)
+        self.assertEqual(result, "https://example.com/oauth/some/endpoint")
+
+    def test_valid_issuer_http(self):
+        # Test a valid HTTP URL
+        self.configuration.api_issuer = "fga.example/some_endpoint"
+        result = self.credentials._parse_issuer(self.configuration.api_issuer)
+        self.assertEqual(result, "https://fga.example/some_endpoint")
+
+    def test_invalid_issuer_no_scheme(self):
+        # Test an issuer URL without a scheme
+        self.configuration.api_issuer = "https://issuer.fga.example:8080/some_endpoint	"
+        result = self.credentials._parse_issuer(self.configuration.api_issuer)
+        self.assertEqual(result, "https://issuer.fga.example:8080/some_endpoint")
+
+    def test_invalid_issuer_bad_scheme(self):
+        # Test an issuer with an unsupported scheme
+        self.configuration.api_issuer = "ftp://example.com"
+        with self.assertRaises(ApiValueError):
+            self.credentials._parse_issuer(self.configuration.api_issuer)
+
+    def test_invalid_issuer_with_port(self):
+        # Test an issuer with an unsupported scheme
+        self.configuration.api_issuer = "https://issuer.fga.example:8080 "
+        result = self.credentials._parse_issuer(self.configuration.api_issuer)
+        self.assertEqual(result, "https://issuer.fga.example:8080/oauth/token")
+
+    # this should raise error
+    def test_invalid_issuer_bad_hostname(self):
+        # Test an issuer with an invalid hostname
+        self.configuration.api_issuer = "https://example?.com"
+        with self.assertRaises(ApiValueError):
+            self.credentials._parse_issuer(self.configuration.api_issuer)
+
+if __name__ == "__main__":
+    unittest.main()