added test cases to cover enroll, challenge and verify steps

Author: subhankarmaiti

src/auth0_server_python/auth_types/__init__.py

@@ -530,6 +530,8 @@ class MfaVerifyResponse(BaseModel):
     id_token: Optional[str] = None
     refresh_token: Optional[str] = None
     scope: Optional[str] = None
+    audience: Optional[str] = None
+    recovery_code: Optional[str] = None
 
 
 # MFA Requirements

src/auth0_server_python/tests/test_mfa_client.py

@@ -238,6 +238,28 @@ async def test_enroll_email_oob_success(self, mocker):
         assert isinstance(result, OobEnrollmentResponse)
         assert result.oob_channel == "email"
 
+    @pytest.mark.asyncio
+    async def test_enroll_push_auth0_channel_success(self, mocker):
+        client = _make_client()
+        response = AsyncMock()
+        response.status_code = 200
+        response.json = MagicMock(return_value={
+            "authenticator_type": "oob",
+            "oob_channel": "auth0",
+            "oob_code": "oob_push_123",
+            "binding_method": "prompt",
+            "recovery_codes": ["rc1", "rc2"]
+        })
+        mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response)
+
+        result = await client.enroll_authenticator({
+            "mfa_token": "tok",
+            "authenticator_types": ["oob"],
+            "oob_channels": ["auth0"]
+        })
+        assert isinstance(result, OobEnrollmentResponse)
+        assert result.oob_channel == "auth0"
+
     @pytest.mark.asyncio
     async def test_enroll_api_error(self, mocker):
         client = _make_client()
@@ -379,6 +401,68 @@ async def test_challenge_api_error(self, mocker):
             })
         assert "Token expired" in str(exc.value)
 
+    @pytest.mark.asyncio
+    async def test_challenge_expired_mfa_token(self, mocker):
+        """When Auth0 returns expired_token for an expired mfa_token."""
+        client = _make_client()
+        response = AsyncMock()
+        response.status_code = 401
+        response.json = MagicMock(return_value={
+            "error": "expired_token",
+            "error_description": "mfa_token is expired"
+        })
+        mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response)
+
+        with pytest.raises(MfaChallengeError) as exc:
+            await client.challenge_authenticator({
+                "mfa_token": "expired_tok",
+                "challenge_type": "otp"
+            })
+        assert "mfa_token is expired" in str(exc.value)
+
+    @pytest.mark.asyncio
+    async def test_challenge_email_with_authenticator_id(self, mocker):
+        """Challenge an email authenticator with a specific authenticator_id."""
+        client = _make_client()
+        response = AsyncMock()
+        response.status_code = 200
+        response.json = MagicMock(return_value={
+            "challenge_type": "oob",
+            "oob_code": "oob_email_challenge_123",
+            "binding_method": "prompt"
+        })
+        mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response)
+
+        result = await client.challenge_authenticator({
+            "mfa_token": "tok",
+            "challenge_type": "oob",
+            "authenticator_id": "email|dev_Fvx38nHufsGL5lWI"
+        })
+        assert result.challenge_type == "oob"
+        assert result.oob_code == "oob_email_challenge_123"
+        assert result.binding_method == "prompt"
+
+    @pytest.mark.asyncio
+    async def test_challenge_sms_with_authenticator_id(self, mocker):
+        """Challenge an SMS authenticator with a specific authenticator_id."""
+        client = _make_client()
+        response = AsyncMock()
+        response.status_code = 200
+        response.json = MagicMock(return_value={
+            "challenge_type": "oob",
+            "oob_code": "oob_sms_challenge_456",
+            "binding_method": "prompt"
+        })
+        mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response)
+
+        result = await client.challenge_authenticator({
+            "mfa_token": "tok",
+            "challenge_type": "oob",
+            "authenticator_id": "sms|dev_h1uXXoVjQ5BpU9iQ"
+        })
+        assert result.challenge_type == "oob"
+        assert result.oob_code == "oob_sms_challenge_456"
+
 
 # ── verify ───────────────────────────────────────────────────────────────────
 
@@ -446,6 +530,115 @@ async def test_verify_no_credential_raises(self):
             await client.verify({"mfa_token": "tok"})
         assert "No verification credential" in str(exc.value)
 
+    @pytest.mark.asyncio
+    async def test_verify_sends_mfa_token_as_form_data(self, mocker):
+        """Verify that mfa_token is sent as form_data, not as Authorization header."""
+        client = _make_client()
+        response = AsyncMock()
+        response.status_code = 200
+        response.json = MagicMock(return_value={
+            "access_token": "at",
+            "token_type": "Bearer",
+            "expires_in": 3600
+        })
+
+        captured_request = {}
+
+        async def mock_post(self_client, url, **kwargs):
+            captured_request["url"] = url
+            captured_request["kwargs"] = kwargs
+            return response
+
+        mocker.patch("httpx.AsyncClient.post", new=mock_post)
+
+        await client.verify({
+            "mfa_token": "my_mfa_token",
+            "otp": "123456"
+        })
+
+        # Verify: mfa_token in form data body, NOT in Authorization header
+        assert "data" in captured_request["kwargs"]
+        form_data = captured_request["kwargs"]["data"]
+        assert form_data["mfa_token"] == "my_mfa_token"
+        assert "Content-Type" in captured_request["kwargs"].get("headers", {})
+        assert captured_request["kwargs"]["headers"]["Content-Type"] == "application/x-www-form-urlencoded"
+        # Should NOT use auth= parameter (no BearerAuth)
+        assert "auth" not in captured_request["kwargs"]
+
+    @pytest.mark.asyncio
+    async def test_verify_expired_mfa_token(self, mocker):
+        """When Auth0 returns expired_token for an expired mfa_token."""
+        client = _make_client()
+        response = AsyncMock()
+        response.status_code = 401
+        response.json = MagicMock(return_value={
+            "error": "expired_token",
+            "error_description": "mfa_token is expired"
+        })
+        mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response)
+
+        with pytest.raises(MfaVerifyError) as exc:
+            await client.verify({"mfa_token": "expired_tok", "otp": "123456"})
+        assert "mfa_token is expired" in str(exc.value)
+
+    @pytest.mark.asyncio
+    async def test_verify_invalid_challenge_type(self, mocker):
+        """When Auth0 returns invalid_request for an unsupported challenge type."""
+        client = _make_client()
+        response = AsyncMock()
+        response.status_code = 400
+        response.json = MagicMock(return_value={
+            "error": "invalid_request",
+            "error_description": "Invalid challenge type"
+        })
+        mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response)
+
+        with pytest.raises(MfaVerifyError) as exc:
+            await client.verify({"mfa_token": "tok", "recovery_code": "ABCD-1234"})
+        assert "Invalid challenge type" in str(exc.value)
+
+    @pytest.mark.asyncio
+    async def test_verify_response_includes_recovery_code(self, mocker):
+        """When MFA verification returns a new recovery_code (e.g., after recovery code use)."""
+        client = _make_client()
+        response = AsyncMock()
+        response.status_code = 200
+        response.json = MagicMock(return_value={
+            "access_token": "new_at",
+            "token_type": "Bearer",
+            "expires_in": 3600,
+            "recovery_code": "NEW-RECOVERY-CODE-XYZ"
+        })
+        mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response)
+
+        result = await client.verify({
+            "mfa_token": "tok",
+            "recovery_code": "OLD-RECOVERY-CODE"
+        })
+        assert isinstance(result, MfaVerifyResponse)
+        assert result.recovery_code == "NEW-RECOVERY-CODE-XYZ"
+
+    @pytest.mark.asyncio
+    async def test_verify_push_oob_success(self, mocker):
+        """Verify with OOB code from push notification challenge."""
+        client = _make_client()
+        response = AsyncMock()
+        response.status_code = 200
+        response.json = MagicMock(return_value={
+            "access_token": "push_at",
+            "token_type": "Bearer",
+            "expires_in": 3600
+        })
+        mocker.patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=response)
+
+        result = await client.verify({
+            "mfa_token": "tok",
+            "oob_code": "oob_push_code",
+            "binding_code": ""
+        })
+        assert isinstance(result, MfaVerifyResponse)
+        assert result.access_token == "push_at"
+
     @pytest.mark.asyncio
     async def test_verify_wrong_code_raises(self, mocker):
         client = _make_client()

src/auth0_server_python/tests/test_server_client.py

@@ -19,6 +19,7 @@
     ListConnectedAccountsResponse,
     LoginWithCustomTokenExchangeOptions,
     LogoutOptions,
+    MfaRequirements,
     TransactionData,
 )
 from auth0_server_python.error import (
@@ -2936,3 +2937,77 @@ async def _fake_fetch(self, domain):
         assert exc.value.mfa_token != "raw_mfa_token_xyz"  # encrypted
     finally:
         sc_mod.ServerClient._fetch_oidc_metadata = original_fetch
+
+
+@pytest.mark.asyncio
+async def test_get_access_token_mfa_required_with_enroll_requirements(mocker):
+    """
+    When get_token_by_refresh_token returns mfa_required with enroll requirements,
+    get_access_token should raise MfaRequiredError with mfa_requirements containing enroll.
+    """
+    mock_secret = "a-test-secret-with-enough-length"
+    mock_store = MagicMock()
+    mock_store.get = AsyncMock(return_value=None)
+    mock_store.set = AsyncMock()
+    mock_store.delete = AsyncMock()
+
+    import auth0_server_python.auth_server.server_client as sc_mod
+
+    original_fetch = sc_mod.ServerClient._fetch_oidc_metadata
+
+    async def _fake_fetch(self, domain):
+        return {
+            "authorization_endpoint": f"https://{domain}/authorize",
+            "token_endpoint": f"https://{domain}/oauth/token",
+            "end_session_endpoint": f"https://{domain}/v2/logout",
+            "backchannel_logout_supported": True,
+        }
+
+    sc_mod.ServerClient._fetch_oidc_metadata = _fake_fetch
+    try:
+        client = ServerClient(
+            domain="auth0.local",
+            client_id="cid",
+            client_secret="csecret",
+            secret=mock_secret,
+            transaction_store=mock_store,
+            state_store=mock_store,
+        )
+
+        # Simulate state with a refresh_token and expired access token
+        mock_store.get = AsyncMock(return_value={
+            "refresh_token": "rt_123",
+            "token_sets": [
+                {
+                    "audience": "default",
+                    "access_token": "expired_at",
+                    "expires_at": 0,
+                }
+            ]
+        })
+
+        # Simulate mfa_required with enroll requirements
+        mfa_err = ApiError(
+            code="mfa_required",
+            message="Multifactor authentication required",
+        )
+        mfa_err.mfa_token = "raw_mfa_token_enroll"
+        mfa_err.mfa_requirements = MfaRequirements(
+            enroll=[
+                {"type": "otp"},
+                {"type": "phone"},
+                {"type": "push-notification"}
+            ]
+        )
+
+        mocker.patch.object(client, "get_token_by_refresh_token",
+                           new_callable=AsyncMock, side_effect=mfa_err)
+
+        with pytest.raises(MfaRequiredError) as exc:
+            await client.get_access_token()
+
+        assert exc.value.mfa_token is not None
+        assert exc.value.mfa_token != "raw_mfa_token_enroll"  # encrypted
+        assert exc.value.mfa_requirements is not None
+    finally:
+        sc_mod.ServerClient._fetch_oidc_metadata = original_fetch