mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Add unit tests
This commit is contained in:
Dmytro Struk
@@ -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
|
||||
@@ -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
|
||||
@@ -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"],
|
||||
)
|
||||
Reference in New Issue
Block a user