register, login, logout implemented with framework independence along with unit tests

Author: Farazzaidi22

kinde_sdk/auth/oauth.py

@@ -249,4 +249,69 @@ def get_tokens_for_core(self, user_id: str) -> Optional[Dict[str, str]]:
         if refresh_token:
             tokens["refresh_token"] = refresh_token
 
-        return tokens
+        return tokens
+    
+    def register(self, state: Optional[str] = None, scope: Optional[List[str]] = None, audience: Optional[str] = None) -> str:
+        """
+        Generate the registration URL for user sign-up.
+
+        Args:
+            state (Optional[str]): A state parameter for CSRF protection.
+            scope (Optional[List[str]]): A list of scopes to request.
+            audience (Optional[str]): The audience for the token.
+
+        Returns:
+            str: The registration URL.
+        """
+        params = {
+            "client_id": self.client_id,
+            "response_type": "code",
+            "redirect_uri": self.redirect_uri,
+            "scope": " ".join(scope) if scope else "openid profile email",  # Default scope
+            "state": state or "",
+        }
+        if audience:
+            params["audience"] = audience
+        return f"{self.auth_url}/register?{urlencode(params)}"
+    
+
+    def login(self, state: Optional[str] = None, scope: Optional[List[str]] = None, audience: Optional[str] = None) -> str:
+        """
+        Generate the login URL for user authentication.
+
+        Args:
+            state (Optional[str]): A state parameter for CSRF protection.
+            scope (Optional[List[str]]): A list of scopes to request.
+            audience (Optional[str]): The audience for the token.
+
+        Returns:
+            str: The login URL.
+        """
+        params = {
+            "client_id": self.client_id,
+            "response_type": "code",
+            "redirect_uri": self.redirect_uri,
+            "scope": " ".join(scope) if scope else "openid profile email",  # Default scope
+            "state": state or "",
+        }
+        if audience:
+            params["audience"] = audience
+        return f"{self.auth_url}/login?{urlencode(params)}"
+    
+    def logout(self, state: Optional[str] = None) -> str:
+        """
+        Generate the logout URL.
+
+        Args:
+            state (Optional[str]): A state parameter for CSRF protection.
+
+        Returns:
+            str: The logout URL.
+        """
+        params = {
+            "client_id": self.client_id,
+            "logout_uri": self.redirect_uri,
+        }
+        if state:
+            params["state"] = state
+        return f"{self.logout_url}?{urlencode(params)}"

testv2/testv2_auth/test_oauth.py

@@ -14,6 +14,34 @@ def setUp(self):
             userinfo_url="https://example.com/userinfo"  # Added userinfo_url
         )
 
+    def test_register(self):
+        # Test the register method
+        url = self.oauth.register(state="xyz", scope=["openid", "profile", "email"])
+        parsed_url = urlparse(url)
+        query_params = parse_qs(parsed_url.query)
+
+        self.assertEqual(parsed_url.scheme, "https")
+        self.assertEqual(parsed_url.netloc, "example.com")
+        self.assertEqual(parsed_url.path, "/auth/register")
+        self.assertEqual(query_params["client_id"][0], "test_client_id")
+        self.assertEqual(query_params["redirect_uri"][0], "http://localhost/callback")
+        self.assertEqual(query_params["scope"][0], "openid profile email")  # Expected scope
+        self.assertEqual(query_params["state"][0], "xyz")
+
+    def test_login(self):
+        # Test the login method
+        url = self.oauth.login(state="xyz", scope=["openid", "profile", "email"])
+        parsed_url = urlparse(url)
+        query_params = parse_qs(parsed_url.query)
+
+        self.assertEqual(parsed_url.scheme, "https")
+        self.assertEqual(parsed_url.netloc, "example.com")
+        self.assertEqual(parsed_url.path, "/auth/login")
+        self.assertEqual(query_params["client_id"][0], "test_client_id")
+        self.assertEqual(query_params["redirect_uri"][0], "http://localhost/callback")
+        self.assertEqual(query_params["scope"][0], "openid profile email")  # Expected scope
+        self.assertEqual(query_params["state"][0], "xyz")
+
     def test_get_login_url(self):
         # Explicitly specify the scope
         url = self.oauth.get_login_url(state="xyz", scope=["openid", "profile", "email"])
@@ -57,7 +85,8 @@ def test_get_tokens_for_core(self):
         self.assertEqual(tokens["refresh_token"], "test_refresh_token")
 
     def test_logout(self):
-        logout_url = self.oauth.logout("test_user_id")
+        # Test the logout method
+        logout_url = self.oauth.logout(state="xyz")
         parsed_url = urlparse(logout_url)
         query_params = parse_qs(parsed_url.query)
 
@@ -66,6 +95,7 @@ def test_logout(self):
         self.assertEqual(parsed_url.path, "/logout")
         self.assertEqual(query_params["client_id"][0], "test_client_id")
         self.assertEqual(query_params["logout_uri"][0], "http://localhost/callback")
+        self.assertEqual(query_params["state"][0], "xyz")
 
-if __name__ == "__main__":
-    unittest.main()
+# if __name__ == "_main_":
+#     unittest.main()