test: add tests for invitation code support in OAuth login flow

Author: Koman Rudden

testv2/testv2_auth/test_invitation_code.py

@@ -0,0 +1,204 @@
+import unittest
+import asyncio
+from unittest.mock import patch, MagicMock
+from urllib.parse import urlparse, parse_qs
+
+from kinde_sdk.auth.oauth import OAuth
+from kinde_sdk.auth.enums import IssuerRouteTypes
+from kinde_sdk.auth.login_options import LoginOptions
+
+
+def run_async(coro):
+    """Helper function to run async tests"""
+    return asyncio.run(coro)
+
+
+class TestInvitationCode(unittest.TestCase):
+    """Tests for invitation code support in OAuth login flow."""
+
+    @patch("requests.get")
+    def setUp(self, mock_get):
+        """Set up test fixtures."""
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {
+            "authorization_endpoint": "https://example.com/oauth2/auth",
+            "token_endpoint": "https://example.com/oauth2/token",
+            "end_session_endpoint": "https://example.com/logout",
+            "userinfo_endpoint": "https://example.com/oauth2/userinfo",
+        }
+        mock_get.return_value = mock_response
+
+        self.oauth = OAuth(
+            client_id="test_client_id",
+            client_secret="test_client_secret",
+            redirect_uri="http://localhost:8000/callback",
+            host="https://test.kinde.com",
+        )
+        self.mock_storage = MagicMock()
+        self.oauth._session_manager = MagicMock()
+        self.oauth._session_manager.storage_manager = self.mock_storage
+        self.oauth.auth_url = "https://example.com/oauth2/auth"
+
+    # -- LoginOptions constants --
+
+    def test_login_options_has_invitation_code_constant(self):
+        """INVITATION_CODE constant is defined on LoginOptions."""
+        self.assertEqual(LoginOptions.INVITATION_CODE, "invitation_code")
+
+    def test_login_options_has_is_invitation_constant(self):
+        """IS_INVITATION constant is defined on LoginOptions."""
+        self.assertEqual(LoginOptions.IS_INVITATION, "is_invitation")
+
+    # -- generate_auth_url: invitation_code --
+
+    def test_invitation_code_appears_in_auth_url(self):
+        """invitation_code is forwarded as a query parameter."""
+        result = run_async(
+            self.oauth.generate_auth_url(
+                login_options={LoginOptions.INVITATION_CODE: "abc123"}
+            )
+        )
+        params = parse_qs(urlparse(result["url"]).query)
+        self.assertEqual(params["invitation_code"], ["abc123"])
+
+    def test_invitation_code_auto_sets_is_invitation(self):
+        """is_invitation is automatically set to 'true' when invitation_code is present."""
+        result = run_async(
+            self.oauth.generate_auth_url(
+                login_options={LoginOptions.INVITATION_CODE: "abc123"}
+            )
+        )
+        params = parse_qs(urlparse(result["url"]).query)
+        self.assertEqual(params["is_invitation"], ["true"])
+
+    def test_invitation_code_with_explicit_is_invitation_true(self):
+        """Explicit is_invitation=True is honoured alongside invitation_code."""
+        result = run_async(
+            self.oauth.generate_auth_url(
+                login_options={
+                    LoginOptions.INVITATION_CODE: "abc123",
+                    LoginOptions.IS_INVITATION: True,
+                }
+            )
+        )
+        params = parse_qs(urlparse(result["url"]).query)
+        self.assertEqual(params["invitation_code"], ["abc123"])
+        self.assertEqual(params["is_invitation"], ["true"])
+
+    def test_invitation_code_with_explicit_is_invitation_false(self):
+        """When invitation_code is present but is_invitation is explicitly False,
+        the auto-set logic still adds is_invitation='true'."""
+        result = run_async(
+            self.oauth.generate_auth_url(
+                login_options={
+                    LoginOptions.INVITATION_CODE: "abc123",
+                    LoginOptions.IS_INVITATION: False,
+                }
+            )
+        )
+        params = parse_qs(urlparse(result["url"]).query)
+        self.assertEqual(params["is_invitation"], ["true"])
+
+    def test_is_invitation_alone_without_invitation_code(self):
+        """is_invitation=True can be set independently of invitation_code."""
+        result = run_async(
+            self.oauth.generate_auth_url(
+                login_options={LoginOptions.IS_INVITATION: True}
+            )
+        )
+        params = parse_qs(urlparse(result["url"]).query)
+        self.assertEqual(params["is_invitation"], ["true"])
+        self.assertNotIn("invitation_code", params)
+
+    def test_is_invitation_false_alone_not_in_url(self):
+        """is_invitation=False without invitation_code produces no is_invitation param."""
+        result = run_async(
+            self.oauth.generate_auth_url(
+                login_options={LoginOptions.IS_INVITATION: False}
+            )
+        )
+        params = parse_qs(urlparse(result["url"]).query)
+        self.assertNotIn("is_invitation", params)
+
+    def test_no_invitation_params_by_default(self):
+        """Neither invitation_code nor is_invitation appear when not requested."""
+        result = run_async(self.oauth.generate_auth_url(login_options={}))
+        params = parse_qs(urlparse(result["url"]).query)
+        self.assertNotIn("invitation_code", params)
+        self.assertNotIn("is_invitation", params)
+
+    def test_invitation_code_empty_string_not_set(self):
+        """An empty invitation_code does not add is_invitation."""
+        result = run_async(
+            self.oauth.generate_auth_url(
+                login_options={LoginOptions.INVITATION_CODE: ""}
+            )
+        )
+        params = parse_qs(urlparse(result["url"]).query)
+        self.assertNotIn("is_invitation", params)
+
+    def test_invitation_code_none_not_set(self):
+        """invitation_code=None is ignored."""
+        result = run_async(
+            self.oauth.generate_auth_url(
+                login_options={LoginOptions.INVITATION_CODE: None}
+            )
+        )
+        params = parse_qs(urlparse(result["url"]).query)
+        self.assertNotIn("invitation_code", params)
+        self.assertNotIn("is_invitation", params)
+
+    # -- login() wrapper --
+
+    def test_login_passes_invitation_code(self):
+        """login() forwards invitation_code to the generated URL."""
+        url = run_async(
+            self.oauth.login(
+                login_options={LoginOptions.INVITATION_CODE: "inv_xyz"}
+            )
+        )
+        params = parse_qs(urlparse(url).query)
+        self.assertEqual(params["invitation_code"], ["inv_xyz"])
+        self.assertEqual(params["is_invitation"], ["true"])
+
+    def test_login_without_invitation_code(self):
+        """login() without invitation options produces no invitation params."""
+        url = run_async(self.oauth.login())
+        params = parse_qs(urlparse(url).query)
+        self.assertNotIn("invitation_code", params)
+        self.assertNotIn("is_invitation", params)
+
+    # -- register() wrapper --
+
+    def test_register_passes_invitation_code(self):
+        """register() forwards invitation_code to the generated URL."""
+        url = run_async(
+            self.oauth.register(
+                login_options={LoginOptions.INVITATION_CODE: "inv_reg"}
+            )
+        )
+        params = parse_qs(urlparse(url).query)
+        self.assertEqual(params["invitation_code"], ["inv_reg"])
+        self.assertEqual(params["is_invitation"], ["true"])
+
+    # -- Coexistence with other params --
+
+    def test_invitation_code_with_org_code(self):
+        """invitation_code works alongside org_code."""
+        result = run_async(
+            self.oauth.generate_auth_url(
+                login_options={
+                    LoginOptions.INVITATION_CODE: "abc123",
+                    LoginOptions.ORG_CODE: "org_456",
+                }
+            )
+        )
+        params = parse_qs(urlparse(result["url"]).query)
+        self.assertEqual(params["invitation_code"], ["abc123"])
+        self.assertEqual(params["is_invitation"], ["true"])
+        self.assertEqual(params["org_code"], ["org_456"])
+
+
+if __name__ == "__main__":
+    unittest.main()

testv2/testv2_framework/__init__.py

@@ -0,0 +1,63 @@
+import unittest
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from fastapi import FastAPI
+from starlette.testclient import TestClient
+from starlette.middleware.sessions import SessionMiddleware
+
+from kinde_fastapi.framework.fastapi_framework import FastAPIFramework
+
+
+class TestFastAPILoginInvitationCode(unittest.TestCase):
+    """Tests for invitation_code extraction in the FastAPI /login route."""
+
+    def setUp(self):
+        self.app = FastAPI()
+        self.app.add_middleware(SessionMiddleware, secret_key="test-secret")
+
+        self.framework = FastAPIFramework(app=self.app)
+
+        # Mock OAuth so no real auth happens
+        self.mock_oauth = MagicMock()
+        self.mock_oauth.login = AsyncMock(return_value="https://kinde.example.com/authorize")
+        self.framework._oauth = self.mock_oauth
+
+        # Register routes (skip middleware added by start() to avoid duplicate)
+        self.framework._register_kinde_routes()
+
+        self.client = TestClient(self.app, follow_redirects=False)
+
+    def test_login_with_invitation_code(self):
+        """invitation_code query param is forwarded to oauth.login()."""
+        resp = self.client.get("/login?invitation_code=inv_abc123")
+
+        self.mock_oauth.login.assert_called_once()
+        login_options = self.mock_oauth.login.call_args[0][0]
+        self.assertEqual(login_options["invitation_code"], "inv_abc123")
+
+    def test_login_without_invitation_code(self):
+        """No invitation_code means oauth.login() gets an empty dict."""
+        resp = self.client.get("/login")
+
+        self.mock_oauth.login.assert_called_once()
+        login_options = self.mock_oauth.login.call_args[0][0]
+        self.assertNotIn("invitation_code", login_options)
+
+    def test_login_with_empty_invitation_code(self):
+        """An empty invitation_code query param is not forwarded."""
+        resp = self.client.get("/login?invitation_code=")
+
+        self.mock_oauth.login.assert_called_once()
+        login_options = self.mock_oauth.login.call_args[0][0]
+        self.assertNotIn("invitation_code", login_options)
+
+    def test_login_redirects_to_oauth_url(self):
+        """The route returns a redirect to the URL from oauth.login()."""
+        resp = self.client.get("/login?invitation_code=inv_xyz")
+
+        self.assertEqual(resp.status_code, 307)
+        self.assertEqual(resp.headers["location"], "https://kinde.example.com/authorize")
+
+
+if __name__ == "__main__":
+    unittest.main()

testv2/testv2_framework/test_fastapi_invitation_code.py

@@ -0,0 +1,66 @@
+import unittest
+from unittest.mock import AsyncMock, MagicMock, patch
+from flask import Flask
+
+from kinde_flask.framework.flask_framework import FlaskFramework
+
+
+class TestFlaskLoginInvitationCode(unittest.TestCase):
+    """Tests for invitation_code extraction in the Flask /login route."""
+
+    def setUp(self):
+        self.app = Flask(__name__)
+        self.app.config["SECRET_KEY"] = "test-secret"
+        self.app.config["TESTING"] = True
+
+        self.framework = FlaskFramework(app=self.app)
+
+        # Mock OAuth so no real auth happens
+        self.mock_oauth = MagicMock()
+        self.mock_oauth.login = AsyncMock(return_value="https://kinde.example.com/authorize")
+        self.framework._oauth = self.mock_oauth
+
+        # Register routes
+        self.framework._initialized = False
+        self.framework._register_kinde_routes()
+
+        self.client = self.app.test_client()
+
+    def test_login_with_invitation_code(self):
+        """invitation_code query param is forwarded to oauth.login()."""
+        with self.app.test_request_context():
+            resp = self.client.get("/login?invitation_code=inv_abc123")
+
+        self.mock_oauth.login.assert_called_once()
+        login_options = self.mock_oauth.login.call_args[0][0]
+        self.assertEqual(login_options["invitation_code"], "inv_abc123")
+
+    def test_login_without_invitation_code(self):
+        """No invitation_code means oauth.login() gets an empty dict."""
+        with self.app.test_request_context():
+            resp = self.client.get("/login")
+
+        self.mock_oauth.login.assert_called_once()
+        login_options = self.mock_oauth.login.call_args[0][0]
+        self.assertNotIn("invitation_code", login_options)
+
+    def test_login_with_empty_invitation_code(self):
+        """An empty invitation_code query param is not forwarded."""
+        with self.app.test_request_context():
+            resp = self.client.get("/login?invitation_code=")
+
+        self.mock_oauth.login.assert_called_once()
+        login_options = self.mock_oauth.login.call_args[0][0]
+        self.assertNotIn("invitation_code", login_options)
+
+    def test_login_redirects_to_oauth_url(self):
+        """The route returns a redirect to the URL from oauth.login()."""
+        with self.app.test_request_context():
+            resp = self.client.get("/login?invitation_code=inv_xyz")
+
+        self.assertEqual(resp.status_code, 302)
+        self.assertEqual(resp.headers["Location"], "https://kinde.example.com/authorize")
+
+
+if __name__ == "__main__":
+    unittest.main()