add test

Author: heyitsaamir

packages/apps/src/microsoft/teams/apps/token_manager.py

@@ -90,7 +90,7 @@ async def _get_token(
                 return JsonWebToken(access_token)
             else:
                 self._logger.debug(f"TokenRes: {token_res}")
-                error = token_res.get("error", ValueError("Error retrieving token"))
+                error = token_res.get("error", "Error retrieving token")
                 if not isinstance(error, BaseException):
                     error = ValueError(error)
                 error_description = token_res.get("error_description", "Error retrieving token from MSAL")

packages/apps/tests/test_app.py

@@ -13,6 +13,7 @@
     Account,
     ConversationAccount,
     InvokeActivity,
+    ManagedIdentityCredentials,
     MessageActivity,
     TokenCredentials,
     TokenProtocol,
@@ -575,3 +576,102 @@ def test_user_agent_format(self, app_with_options: App):
         # Verify the http_client has the correct User-Agent header
         assert "User-Agent" in app_with_options.http_client._options.headers
         assert app_with_options.http_client._options.headers["User-Agent"] == expected_user_agent
+
+    @pytest.mark.parametrize(
+        "options_dict,env_vars,expected_client_id,expected_tenant_id,description",
+        [
+            # Inferred from client_id only
+            (
+                {"client_id": "test-managed-identity-client-id"},
+                {"CLIENT_SECRET": "", "TENANT_ID": "test-tenant-id"},
+                "test-managed-identity-client-id",
+                "test-tenant-id",
+                "inferred from client_id only",
+            ),
+            # managed_identity_client_id equals client_id (valid)
+            (
+                {"client_id": "test-client-id", "managed_identity_client_id": "test-client-id"},
+                {"CLIENT_SECRET": "", "TENANT_ID": "test-tenant-id"},
+                "test-client-id",
+                "test-tenant-id",
+                "managed_identity_client_id equals client_id",
+            ),
+            # From environment variables
+            (
+                {},
+                {"CLIENT_ID": "env-managed-identity-client-id", "CLIENT_SECRET": "", "TENANT_ID": "env-tenant-id"},
+                "env-managed-identity-client-id",
+                "env-tenant-id",
+                "from environment variables",
+            ),
+            # Explicit managed_identity_client_id
+            (
+                {
+                    "client_id": "test-app-id",
+                    "managed_identity_client_id": "test-app-id",
+                    "tenant_id": "test-tenant-id",
+                },
+                {"CLIENT_SECRET": ""},
+                "test-app-id",
+                "test-tenant-id",
+                "explicit managed_identity_client_id",
+            ),
+        ],
+    )
+    def test_app_init_with_managed_identity(
+        self,
+        mock_logger,
+        mock_storage,
+        options_dict: dict,
+        env_vars: dict,
+        expected_client_id: str,
+        expected_tenant_id: str,
+        description: str,
+    ):
+        """Test app initialization with managed identity credentials."""
+        options = AppOptions(logger=mock_logger, storage=mock_storage, **options_dict)
+
+        with patch.dict("os.environ", env_vars, clear=False):
+            app = App(**options)
+
+            assert app.credentials is not None, f"Failed for: {description}"
+            assert isinstance(app.credentials, ManagedIdentityCredentials), f"Failed for: {description}"
+            assert app.credentials.client_id == expected_client_id, f"Failed for: {description}"
+            assert app.credentials.tenant_id == expected_tenant_id, f"Failed for: {description}"
+
+    def test_app_init_with_managed_identity_client_id_mismatch(self, mock_logger, mock_storage):
+        """Test app init raises error when managed_identity_client_id != client_id (federated identity)."""
+        # When managed_identity_client_id differs from client_id, should raise error
+        # (Federated Identity Credentials not yet supported)
+        options = AppOptions(
+            logger=mock_logger,
+            storage=mock_storage,
+            client_id="app-client-id",
+            managed_identity_client_id="different-managed-identity-id",  # Different!
+        )
+
+        with patch.dict("os.environ", {"CLIENT_SECRET": "", "TENANT_ID": "test-tenant-id"}, clear=False):
+            with pytest.raises(ValueError) as exc_info:
+                App(**options)
+
+            assert "Federated Identity Credentials is not yet supported" in str(exc_info.value)
+            assert "managed_identity_client_id must equal client_id" in str(exc_info.value)
+
+    def test_app_init_with_client_secret_takes_precedence(self, mock_logger, mock_storage):
+        """Test that ClientCredentials is used when both client_secret and managed_identity_client_id are provided."""
+        # When client_secret is provided, it should take precedence over managed identity
+        options = AppOptions(
+            logger=mock_logger,
+            storage=mock_storage,
+            client_id="test-client-id",
+            client_secret="test-client-secret",
+            managed_identity_client_id="test-managed-id",  # This should be ignored
+            tenant_id="test-tenant-id",
+        )
+
+        app = App(**options)
+
+        assert app.credentials is not None
+        # Should use ClientCredentials, not ManagedIdentityCredentials
+        assert type(app.credentials).__name__ == "ClientCredentials"
+        assert app.credentials.client_id == "test-client-id"

packages/apps/tests/test_token_manager.py

@@ -3,11 +3,12 @@
 Licensed under the MIT License.
 """
 
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock, create_autospec, patch
 
 import pytest
-from microsoft.teams.api import ClientCredentials, JsonWebToken
+from microsoft.teams.api import ClientCredentials, JsonWebToken, ManagedIdentityCredentials
 from microsoft.teams.apps.token_manager import TokenManager
+from msal import ManagedIdentityClient  # pyright: ignore[reportMissingTypeStubs]
 
 # Valid JWT-like token for testing (format: header.payload.signature)
 VALID_TEST_TOKEN = (
@@ -105,3 +106,94 @@ async def test_get_graph_token_with_tenant(self):
             calls = mock_msal_class.call_args_list
             # Should have been called with different-tenant-id
             assert any("different-tenant-id" in str(call) for call in calls)
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "get_token_method,expected_resource",
+        [
+            ("get_bot_token", "https://api.botframework.com"),
+            ("get_graph_token", "https://graph.microsoft.com"),
+        ],
+    )
+    async def test_get_token_with_managed_identity(self, get_token_method: str, expected_resource: str):
+        """Test token retrieval using ManagedIdentityCredentials."""
+        mock_credentials = ManagedIdentityCredentials(
+            client_id="test-managed-identity-client-id",
+            tenant_id="test-tenant-id",
+        )
+
+        # Create a mock that will pass isinstance checks
+        mock_msal_client = create_autospec(ManagedIdentityClient, instance=True)
+        mock_msal_client.acquire_token_for_client.return_value = {"access_token": VALID_TEST_TOKEN}
+
+        manager = TokenManager(credentials=mock_credentials)
+
+        # Patch _get_msal_client to return our mock
+        with patch.object(manager, "_get_msal_client", return_value=mock_msal_client):
+            # Call the method dynamically
+            token = await getattr(manager, get_token_method)()
+
+            assert token is not None
+            assert isinstance(token, JsonWebToken)
+            assert str(token) == VALID_TEST_TOKEN
+
+            # Verify MSAL was called with resource parameter (not scopes list)
+            # and without /.default suffix
+            mock_msal_client.acquire_token_for_client.assert_called_once_with(resource=expected_resource)
+
+    @pytest.mark.asyncio
+    async def test_get_graph_token_with_managed_identity_and_tenant(self):
+        """Test getting tenant-specific graph token with ManagedIdentityCredentials."""
+        mock_credentials = ManagedIdentityCredentials(
+            client_id="test-managed-identity-client-id",
+            tenant_id="original-tenant-id",
+        )
+
+        # Create a mock that will pass isinstance checks
+        mock_msal_client = create_autospec(ManagedIdentityClient, instance=True)
+        mock_msal_client.acquire_token_for_client.return_value = {"access_token": VALID_TEST_TOKEN}
+
+        manager = TokenManager(credentials=mock_credentials)
+
+        # Track calls to _get_msal_client
+        get_msal_client_calls: list[str] = []
+
+        def track_get_msal_client(tenant_id: str):
+            get_msal_client_calls.append(tenant_id)
+            return mock_msal_client
+
+        # Patch _get_msal_client to track calls
+        with patch.object(manager, "_get_msal_client", side_effect=track_get_msal_client):
+            # Request token for different tenant
+            token = await manager.get_graph_token("different-tenant-id")
+
+            assert token is not None
+            assert isinstance(token, JsonWebToken)
+
+            # Verify _get_msal_client was called with different-tenant-id
+            assert "different-tenant-id" in get_msal_client_calls
+
+    @pytest.mark.asyncio
+    async def test_get_token_error_handling_with_managed_identity(self):
+        """Test error handling when token acquisition fails with ManagedIdentityCredentials."""
+        mock_credentials = ManagedIdentityCredentials(
+            client_id="test-managed-identity-client-id",
+            tenant_id="test-tenant-id",
+        )
+
+        # Create a mock that returns an error
+        mock_msal_client = create_autospec(ManagedIdentityClient, instance=True)
+        mock_msal_client.acquire_token_for_client.return_value = {
+            "error": "invalid_client",
+            "error_description": "Invalid managed identity configuration",
+        }
+
+        manager = TokenManager(credentials=mock_credentials)
+
+        # Patch _get_msal_client to return our mock
+        with patch.object(manager, "_get_msal_client", return_value=mock_msal_client):
+            # Should raise an error when token acquisition fails
+            with pytest.raises(ValueError) as exc_info:
+                await manager.get_bot_token()
+
+            assert "invalid_client" in str(exc_info.value)