From 92de7a6c6e600a34cc9686c7a5c9e6082c75a909 Mon Sep 17 00:00:00 2001 From: Dmytro Struk <13853051+dmytrostruk@users.noreply.github.com> Date: Thu, 11 Sep 2025 17:31:57 -0700 Subject: [PATCH] Add unit tests --- .../packages/copilotstudio/tests/conftest.py | 91 +++++++ .../copilotstudio/tests/test_acquire_token.py | 243 ++++++++++++++++++ .../copilotstudio/tests/test_agent.py | 167 ++++++++++++ 3 files changed, 501 insertions(+) create mode 100644 python/packages/copilotstudio/tests/conftest.py create mode 100644 python/packages/copilotstudio/tests/test_acquire_token.py create mode 100644 python/packages/copilotstudio/tests/test_agent.py diff --git a/python/packages/copilotstudio/tests/conftest.py b/python/packages/copilotstudio/tests/conftest.py new file mode 100644 index 0000000000..b836a5e71a --- /dev/null +++ b/python/packages/copilotstudio/tests/conftest.py @@ -0,0 +1,91 @@ +# Copyright (c) Microsoft. All rights reserved. + +from typing import Any +from unittest.mock import MagicMock + +import pytest +from microsoft_agents.copilotstudio.client import CopilotClient + + +@pytest.fixture +def exclude_list(request: Any) -> list[str]: + """Fixture that returns a list of environment variables to exclude.""" + return request.param if hasattr(request, "param") else [] + + +@pytest.fixture +def override_env_param_dict(request: Any) -> dict[str, str]: + """Fixture that returns a dict of environment variables to override.""" + return request.param if hasattr(request, "param") else {} + + +@pytest.fixture() +def copilot_studio_unit_test_env(monkeypatch, exclude_list, override_env_param_dict): # type: ignore + """Fixture to set environment variables for CopilotStudioSettings.""" + + if exclude_list is None: + exclude_list = [] + + if override_env_param_dict is None: + override_env_param_dict = {} + + env_vars = { + "COPILOTSTUDIOAGENT__ENVIRONMENTID": "test-environment-id", + "COPILOTSTUDIOAGENT__SCHEMANAME": "test-schema-name", + "COPILOTSTUDIOAGENT__AGENTAPPID": "test-client-id", + "COPILOTSTUDIOAGENT__TENANTID": "test-tenant-id", + } + + env_vars.update(override_env_param_dict) # type: ignore + + for key, value in env_vars.items(): + if key in exclude_list: + monkeypatch.delenv(key, raising=False) # type: ignore + continue + monkeypatch.setenv(key, value) # type: ignore + + return env_vars + + +@pytest.fixture +def mock_copilot_client() -> MagicMock: + """Mock CopilotClient for testing.""" + return MagicMock(spec=CopilotClient) + + +@pytest.fixture +def mock_pca() -> MagicMock: + """Mock PublicClientApplication for testing.""" + mock_pca = MagicMock() + + # Mock successful token response + mock_token_response = { + "access_token": "test-access-token-12345", + "token_type": "Bearer", + "expires_in": 3600, + } + + mock_pca.get_accounts.return_value = [] + mock_pca.acquire_token_interactive.return_value = mock_token_response + mock_pca.acquire_token_silent.return_value = mock_token_response + + return mock_pca + + +@pytest.fixture +def mock_activity() -> MagicMock: + """Mock Activity for testing.""" + mock_activity = MagicMock() + mock_activity.text = "Test response" + mock_activity.type = "message" + mock_activity.id = "test-activity-id" + mock_activity.from_property.name = "Test Bot" + return mock_activity + + +@pytest.fixture +def mock_conversation() -> MagicMock: + """Mock conversation for testing.""" + mock_conversation = MagicMock() + mock_conversation.id = "test-conversation-id" + return mock_conversation diff --git a/python/packages/copilotstudio/tests/test_acquire_token.py b/python/packages/copilotstudio/tests/test_acquire_token.py new file mode 100644 index 0000000000..3aeb6312c3 --- /dev/null +++ b/python/packages/copilotstudio/tests/test_acquire_token.py @@ -0,0 +1,243 @@ +# Copyright (c) Microsoft. All rights reserved. + +from unittest.mock import MagicMock, patch + +import pytest +from agent_framework.exceptions import ServiceException + +from agent_framework_copilotstudio._acquire_token import DEFAULT_SCOPES, acquire_token + + +class TestAcquireToken: + """Test class for token acquisition functionality.""" + + def test_acquire_token_missing_client_id(self) -> None: + """Test that acquire_token raises ServiceException when client_id is missing.""" + with pytest.raises(ServiceException, match="Client ID is required for token acquisition"): + acquire_token(client_id="", tenant_id="test-tenant-id") + + def test_acquire_token_missing_tenant_id(self) -> None: + """Test that acquire_token raises ServiceException when tenant_id is missing.""" + with pytest.raises(ServiceException, match="Tenant ID is required for token acquisition"): + acquire_token(client_id="test-client-id", tenant_id="") + + def test_acquire_token_none_client_id(self) -> None: + """Test that acquire_token raises ServiceException when client_id is None.""" + with pytest.raises(ServiceException, match="Client ID is required for token acquisition"): + acquire_token(client_id=None, tenant_id="test-tenant-id") # type: ignore + + def test_acquire_token_none_tenant_id(self) -> None: + """Test that acquire_token raises ServiceException when tenant_id is None.""" + with pytest.raises(ServiceException, match="Tenant ID is required for token acquisition"): + acquire_token(client_id="test-client-id", tenant_id=None) # type: ignore + + @patch("agent_framework_copilotstudio._acquire_token.PublicClientApplication") + def test_acquire_token_silent_success(self, mock_pca_class: MagicMock) -> None: + """Test successful silent token acquisition.""" + mock_pca = MagicMock() + mock_pca_class.return_value = mock_pca + + mock_account = MagicMock() + mock_pca.get_accounts.return_value = [mock_account] + + mock_token_response = {"access_token": "test-access-token-12345"} + mock_pca.acquire_token_silent.return_value = mock_token_response + + result = acquire_token( + client_id="test-client-id", + tenant_id="test-tenant-id", + ) + + assert result == "test-access-token-12345" + mock_pca_class.assert_called_once_with( + client_id="test-client-id", + authority="https://login.microsoftonline.com/test-tenant-id", + token_cache=None, + ) + mock_pca.get_accounts.assert_called_once_with(username=None) + mock_pca.acquire_token_silent.assert_called_once_with(scopes=DEFAULT_SCOPES, account=mock_account) + + @patch("agent_framework_copilotstudio._acquire_token.PublicClientApplication") + def test_acquire_token_silent_success_with_username(self, mock_pca_class: MagicMock) -> None: + """Test successful silent token acquisition with username.""" + mock_pca = MagicMock() + mock_pca_class.return_value = mock_pca + + mock_account = MagicMock() + mock_pca.get_accounts.return_value = [mock_account] + + mock_token_response = {"access_token": "test-access-token-12345"} + mock_pca.acquire_token_silent.return_value = mock_token_response + + result = acquire_token( + client_id="test-client-id", + tenant_id="test-tenant-id", + username="test-user@example.com", + ) + + assert result == "test-access-token-12345" + mock_pca.get_accounts.assert_called_once_with(username="test-user@example.com") + mock_pca.acquire_token_silent.assert_called_once_with(scopes=DEFAULT_SCOPES, account=mock_account) + + @patch("agent_framework_copilotstudio._acquire_token.PublicClientApplication") + def test_acquire_token_silent_success_with_custom_scopes(self, mock_pca_class: MagicMock) -> None: + """Test successful silent token acquisition with custom scopes.""" + # Setup + mock_pca = MagicMock() + mock_pca_class.return_value = mock_pca + + mock_account = MagicMock() + mock_pca.get_accounts.return_value = [mock_account] + + mock_token_response = {"access_token": "test-access-token-12345"} + mock_pca.acquire_token_silent.return_value = mock_token_response + + custom_scopes = ["https://custom.api.com/.default"] + + result = acquire_token( + client_id="test-client-id", + tenant_id="test-tenant-id", + scopes=custom_scopes, + ) + + assert result == "test-access-token-12345" + mock_pca.acquire_token_silent.assert_called_once_with(scopes=custom_scopes, account=mock_account) + + @patch("agent_framework_copilotstudio._acquire_token.PublicClientApplication") + def test_acquire_token_interactive_success_no_accounts(self, mock_pca_class: MagicMock) -> None: + """Test successful interactive token acquisition when no cached accounts exist.""" + # Setup + mock_pca = MagicMock() + mock_pca_class.return_value = mock_pca + + mock_pca.get_accounts.return_value = [] # No cached accounts + + mock_token_response = {"access_token": "test-interactive-token-67890"} + mock_pca.acquire_token_interactive.return_value = mock_token_response + + result = acquire_token( + client_id="test-client-id", + tenant_id="test-tenant-id", + ) + + assert result == "test-interactive-token-67890" + mock_pca.acquire_token_interactive.assert_called_once_with(scopes=DEFAULT_SCOPES) + + @patch("agent_framework_copilotstudio._acquire_token.PublicClientApplication") + def test_acquire_token_fallback_to_interactive_after_silent_fails(self, mock_pca_class: MagicMock) -> None: + """Test fallback to interactive authentication when silent acquisition fails.""" + mock_pca = MagicMock() + mock_pca_class.return_value = mock_pca + + mock_account = MagicMock() + mock_pca.get_accounts.return_value = [mock_account] + + # Silent acquisition fails with error response + mock_silent_error_response = {"error": "invalid_grant", "error_description": "Token expired"} + mock_pca.acquire_token_silent.return_value = mock_silent_error_response + + # Interactive acquisition succeeds + mock_interactive_response = {"access_token": "test-interactive-token-67890"} + mock_pca.acquire_token_interactive.return_value = mock_interactive_response + + result = acquire_token( + client_id="test-client-id", + tenant_id="test-tenant-id", + ) + + assert result == "test-interactive-token-67890" + mock_pca.acquire_token_silent.assert_called_once_with(scopes=DEFAULT_SCOPES, account=mock_account) + mock_pca.acquire_token_interactive.assert_called_once_with(scopes=DEFAULT_SCOPES) + + @patch("agent_framework_copilotstudio._acquire_token.PublicClientApplication") + def test_acquire_token_fallback_to_interactive_after_silent_exception(self, mock_pca_class: MagicMock) -> None: + """Test fallback to interactive authentication when silent acquisition throws exception.""" + mock_pca = MagicMock() + mock_pca_class.return_value = mock_pca + + mock_account = MagicMock() + mock_pca.get_accounts.return_value = [mock_account] + + # Silent acquisition throws exception + mock_pca.acquire_token_silent.side_effect = Exception("Network error") + + # Interactive acquisition succeeds + mock_interactive_response = {"access_token": "test-interactive-token-67890"} + mock_pca.acquire_token_interactive.return_value = mock_interactive_response + + result = acquire_token( + client_id="test-client-id", + tenant_id="test-tenant-id", + ) + + assert result == "test-interactive-token-67890" + mock_pca.acquire_token_silent.assert_called_once_with(scopes=DEFAULT_SCOPES, account=mock_account) + mock_pca.acquire_token_interactive.assert_called_once_with(scopes=DEFAULT_SCOPES) + + @patch("agent_framework_copilotstudio._acquire_token.PublicClientApplication") + def test_acquire_token_interactive_error_response(self, mock_pca_class: MagicMock) -> None: + """Test that acquire_token handles error responses from interactive authentication.""" + mock_pca = MagicMock() + mock_pca_class.return_value = mock_pca + + mock_pca.get_accounts.return_value = [] # No cached accounts + + # Interactive acquisition returns error + mock_error_response = {"error": "access_denied", "error_description": "User denied consent"} + mock_pca.acquire_token_interactive.return_value = mock_error_response + + with pytest.raises(ServiceException, match="Authentication token cannot be acquired"): + acquire_token( + client_id="test-client-id", + tenant_id="test-tenant-id", + ) + + @patch("agent_framework_copilotstudio._acquire_token.PublicClientApplication") + def test_acquire_token_interactive_exception(self, mock_pca_class: MagicMock) -> None: + """Test that acquire_token handles exceptions from interactive authentication.""" + mock_pca = MagicMock() + mock_pca_class.return_value = mock_pca + + mock_pca.get_accounts.return_value = [] # No cached accounts + + # Interactive acquisition throws exception + mock_pca.acquire_token_interactive.side_effect = Exception("Authentication service unavailable") + + with pytest.raises(ServiceException, match="Failed to acquire authentication token"): + acquire_token( + client_id="test-client-id", + tenant_id="test-tenant-id", + ) + + @patch("agent_framework_copilotstudio._acquire_token.PublicClientApplication") + def test_acquire_token_with_token_cache(self, mock_pca_class: MagicMock) -> None: + """Test acquire_token with custom token cache.""" + mock_pca = MagicMock() + mock_pca_class.return_value = mock_pca + + mock_account = MagicMock() + mock_pca.get_accounts.return_value = [mock_account] + + mock_token_response = {"access_token": "test-cached-token"} + mock_pca.acquire_token_silent.return_value = mock_token_response + + mock_token_cache = MagicMock() + + result = acquire_token( + client_id="test-client-id", + tenant_id="test-tenant-id", + token_cache=mock_token_cache, + ) + + assert result == "test-cached-token" + mock_pca_class.assert_called_once_with( + client_id="test-client-id", + authority="https://login.microsoftonline.com/test-tenant-id", + token_cache=mock_token_cache, + ) + + def test_default_scopes_constant(self) -> None: + """Test that DEFAULT_SCOPES constant is properly defined.""" + assert DEFAULT_SCOPES == ["https://api.powerplatform.com/.default"] + assert isinstance(DEFAULT_SCOPES, list) + assert len(DEFAULT_SCOPES) == 1 diff --git a/python/packages/copilotstudio/tests/test_agent.py b/python/packages/copilotstudio/tests/test_agent.py new file mode 100644 index 0000000000..1126d71dfc --- /dev/null +++ b/python/packages/copilotstudio/tests/test_agent.py @@ -0,0 +1,167 @@ +# Copyright (c) Microsoft. All rights reserved. + +from unittest.mock import MagicMock, patch + +import pytest +from agent_framework.exceptions import ServiceInitializationError + +from agent_framework_copilotstudio._agent import CopilotStudioSettings + + +class TestCopilotStudioSettings: + """Test class for CopilotStudioSettings.""" + + def test_copilot_studio_settings_with_env_vars(self, copilot_studio_unit_test_env: dict[str, str]) -> None: + """Test CopilotStudioSettings initialization with environment variables.""" + settings = CopilotStudioSettings() + + assert settings.environmentid == "test-environment-id" + assert settings.schemaname == "test-schema-name" + assert settings.agentappid == "test-client-id" + assert settings.tenantid == "test-tenant-id" + + def test_copilot_studio_settings_direct_values(self) -> None: + """Test CopilotStudioSettings initialization with direct values.""" + settings = CopilotStudioSettings( + environmentid="direct-env-id", + schemaname="direct-schema-name", + agentappid="direct-client-id", + tenantid="direct-tenant-id", + ) + + assert settings.environmentid == "direct-env-id" + assert settings.schemaname == "direct-schema-name" + assert settings.agentappid == "direct-client-id" + assert settings.tenantid == "direct-tenant-id" + + def test_copilot_studio_settings_none_values(self) -> None: + """Test CopilotStudioSettings with None values.""" + with patch.dict("os.environ", {}, clear=True): + settings = CopilotStudioSettings() + + assert settings.environmentid is None + assert settings.schemaname is None + assert settings.agentappid is None + assert settings.tenantid is None + + +class TestCopilotStudioAgentInitializationLogic: + """Test class for CopilotStudioAgent initialization logic without full agent creation.""" + + @patch("agent_framework_copilotstudio._agent.acquire_token") + def test_token_acquisition_called_with_correct_parameters( + self, mock_acquire_token: MagicMock, copilot_studio_unit_test_env: dict[str, str] + ) -> None: + """Test that token acquisition is called with correct parameters during initialization.""" + from agent_framework_copilotstudio._agent import CopilotStudioAgent + + mock_acquire_token.return_value = "test-token" + + with ( + patch.object(CopilotStudioAgent, "__init__", return_value=None), + patch("agent_framework_copilotstudio._agent.CopilotClient"), + patch("agent_framework_copilotstudio._agent.ConnectionSettings"), + ): + settings = CopilotStudioSettings() + + from agent_framework_copilotstudio._agent import acquire_token + + token = acquire_token( + client_id=settings.agentappid or "test-client-id", + tenant_id=settings.tenantid or "test-tenant-id", + username=None, + token_cache=None, + scopes=None, + ) + + assert token == "test-token" + mock_acquire_token.assert_called_once_with( + client_id="test-client-id", + tenant_id="test-tenant-id", + username=None, + token_cache=None, + scopes=None, + ) + + @pytest.mark.parametrize("exclude_list", [["COPILOTSTUDIOAGENT__ENVIRONMENTID"]], indirect=True) + def test_missing_environment_id_validation(self, copilot_studio_unit_test_env: dict[str, str]) -> None: + """Test that missing environment ID is properly validated.""" + settings = CopilotStudioSettings() + assert settings.environmentid is None + + if not settings.environmentid: + with pytest.raises(ServiceInitializationError, match="Copilot Studio environment ID is required"): + raise ServiceInitializationError( + "Copilot Studio environment ID is required. Set via 'environment_id' parameter " + "or 'COPILOTSTUDIOAGENT__ENVIRONMENTID' environment variable." + ) + + @pytest.mark.parametrize("exclude_list", [["COPILOTSTUDIOAGENT__SCHEMANAME"]], indirect=True) + def test_missing_schema_name_validation(self, copilot_studio_unit_test_env: dict[str, str]) -> None: + """Test that missing schema name is properly validated.""" + settings = CopilotStudioSettings() + assert settings.schemaname is None + + if not settings.schemaname: + with pytest.raises( + ServiceInitializationError, match="Copilot Studio agent identifier/schema name is required" + ): + raise ServiceInitializationError( + "Copilot Studio agent identifier/schema name is required. Set via 'agent_identifier' parameter " + "or 'COPILOTSTUDIOAGENT__SCHEMANAME' environment variable." + ) + + @pytest.mark.parametrize("exclude_list", [["COPILOTSTUDIOAGENT__AGENTAPPID"]], indirect=True) + def test_missing_client_id_validation(self, copilot_studio_unit_test_env: dict[str, str]) -> None: + """Test that missing client ID is properly validated.""" + settings = CopilotStudioSettings() + assert settings.agentappid is None + + if not settings.agentappid: + with pytest.raises(ServiceInitializationError, match="Copilot Studio client ID is required"): + raise ServiceInitializationError( + "Copilot Studio client ID is required. Set via 'client_id' parameter " + "or 'COPILOTSTUDIOAGENT__AGENTAPPID' environment variable." + ) + + @pytest.mark.parametrize("exclude_list", [["COPILOTSTUDIOAGENT__TENANTID"]], indirect=True) + def test_missing_tenant_id_validation(self, copilot_studio_unit_test_env: dict[str, str]) -> None: + """Test that missing tenant ID is properly validated.""" + settings = CopilotStudioSettings() + assert settings.tenantid is None + + if not settings.tenantid: + with pytest.raises(ServiceInitializationError, match="Copilot Studio tenant ID is required"): + raise ServiceInitializationError( + "Copilot Studio tenant ID is required. Set via 'tenant_id' parameter " + "or 'COPILOTSTUDIOAGENT__TENANTID' environment variable." + ) + + +class TestCopilotStudioAgentMethods: + """Test class for individual methods that can be tested without full initialization.""" + + @patch("agent_framework_copilotstudio._agent.acquire_token") + def test_token_acquisition_with_custom_parameters(self, mock_acquire_token: MagicMock) -> None: + """Test token acquisition with custom parameters.""" + from agent_framework_copilotstudio._agent import acquire_token + + mock_acquire_token.return_value = "custom-token" + + # Test the acquire_token function directly + token = acquire_token( + client_id="custom-client-id", + tenant_id="custom-tenant-id", + username="custom-user@example.com", + token_cache="custom-cache", + scopes=["custom-scope"], + ) + + assert token == "custom-token" + mock_acquire_token.assert_called_once_with( + client_id="custom-client-id", + tenant_id="custom-tenant-id", + username="custom-user@example.com", + token_cache="custom-cache", + scopes=["custom-scope"], + )