Files
2026-07-13 12:38:34 +08:00

591 lines
23 KiB
Python

"""Tests for auth configs management."""
from unittest.mock import Mock
import pytest
from composio.client.types import (
auth_config_create_response,
auth_config_list_response,
auth_config_retrieve_response,
)
from composio.core.models.auth_configs import AuthConfigs
class TestAuthConfigs:
"""Test suite for AuthConfigs class."""
@pytest.fixture
def mock_client(self):
"""Create a mock client for testing."""
client = Mock()
client.auth_configs = Mock()
client.auth_configs.list = Mock()
client.auth_configs.create = Mock()
client.auth_configs.retrieve = Mock()
client.auth_configs.update = Mock()
client.auth_configs.delete = Mock()
client.auth_configs.update_status = Mock()
client.not_given = object() # Sentinel value for optional params
return client
@pytest.fixture
def auth_configs(self, mock_client):
"""Create an AuthConfigs instance with mock client."""
return AuthConfigs(client=mock_client)
@pytest.fixture
def mock_auth_config_response(self):
"""Mock auth config retrieve response."""
response = Mock(spec=auth_config_retrieve_response.AuthConfigRetrieveResponse)
response.id = "auth_12345"
response.name = "Test Auth Config"
response.no_of_connections = 5
response.status = "ENABLED"
response.toolkit = Mock()
response.toolkit.logo = "https://example.com/logo.png"
response.toolkit.slug = "github"
response.uuid = "uuid-12345"
response.auth_scheme = "OAUTH2"
response.credentials = {
"client_id": "test_client_id",
"client_secret": "test_client_secret",
}
response.expected_input_fields = [
{"name": "client_id", "type": "string"},
{"name": "client_secret", "type": "string"},
]
response.is_composio_managed = True
response.created_by = "user_123"
response.created_at = "2023-01-01T00:00:00Z"
response.last_updated_at = "2023-01-01T00:00:00Z"
return response
def test_constructor_creates_instance(self, auth_configs, mock_client):
"""Test that AuthConfigs instance is created successfully."""
assert isinstance(auth_configs, AuthConfigs)
assert auth_configs._client is mock_client
# List tests
def test_list_without_params(self, auth_configs, mock_client):
"""Test listing auth configs without query parameters."""
mock_response = Mock(spec=auth_config_list_response.AuthConfigListResponse)
mock_response.items = []
mock_response.next_cursor = None
mock_response.total_pages = 0
mock_client.auth_configs.list.return_value = mock_response
result = auth_configs.list()
mock_client.auth_configs.list.assert_called_once_with()
assert result == mock_response
def test_list_with_params(self, auth_configs, mock_client):
"""Test listing auth configs with query parameters."""
mock_response = Mock(spec=auth_config_list_response.AuthConfigListResponse)
mock_response.items = []
mock_client.auth_configs.list.return_value = mock_response
result = auth_configs.list(
cursor="cursor_123",
is_composio_managed=True,
limit=10,
toolkit_slug="github",
)
mock_client.auth_configs.list.assert_called_once_with(
cursor="cursor_123",
is_composio_managed=True,
limit=10,
toolkit_slug="github",
)
assert result == mock_response
def test_list_with_empty_result(self, auth_configs, mock_client):
"""Test listing auth configs returns empty list."""
mock_response = Mock(spec=auth_config_list_response.AuthConfigListResponse)
mock_response.items = []
mock_response.next_cursor = None
mock_response.total_pages = 0
mock_client.auth_configs.list.return_value = mock_response
result = auth_configs.list()
assert len(result.items) == 0
assert result.next_cursor is None
assert result.total_pages == 0
# Create tests
def test_create_with_default_composio_managed_auth(self, auth_configs, mock_client):
"""Test creating auth config with default Composio managed type."""
mock_auth_config = Mock(spec=auth_config_create_response.AuthConfig)
mock_auth_config.id = "auth_12345"
mock_auth_config.auth_scheme = "OAUTH2"
mock_auth_config.is_composio_managed = True
mock_response = Mock(spec=auth_config_create_response.AuthConfigCreateResponse)
mock_response.auth_config = mock_auth_config
mock_client.auth_configs.create.return_value = mock_response
options = {
"type": "use_composio_managed_auth",
"name": "My GitHub Config",
}
result = auth_configs.create("github", options)
mock_client.auth_configs.create.assert_called_once()
call_args = mock_client.auth_configs.create.call_args
assert call_args.kwargs["toolkit"] == {"slug": "github"}
assert call_args.kwargs["auth_config"] == options
assert result == mock_auth_config
def test_create_with_custom_auth_and_credentials(self, auth_configs, mock_client):
"""Test creating custom auth config with credentials."""
mock_auth_config = Mock(spec=auth_config_create_response.AuthConfig)
mock_auth_config.id = "auth_12345"
mock_auth_config.auth_scheme = "OAUTH2"
mock_auth_config.is_composio_managed = False
mock_response = Mock(spec=auth_config_create_response.AuthConfigCreateResponse)
mock_response.auth_config = mock_auth_config
mock_client.auth_configs.create.return_value = mock_response
options = {
"type": "use_custom_auth",
"name": "Custom GitHub Auth",
"auth_scheme": "OAUTH2",
"credentials": {
"client_id": "test_client_id",
"client_secret": "test_client_secret",
},
}
result = auth_configs.create("github", options)
mock_client.auth_configs.create.assert_called_once()
call_args = mock_client.auth_configs.create.call_args
assert call_args.kwargs["toolkit"] == {"slug": "github"}
assert call_args.kwargs["auth_config"] == options
assert result.is_composio_managed is False
assert result.auth_scheme == "OAUTH2"
def test_create_with_tool_access_config(self, auth_configs, mock_client):
"""Test creating auth config with tool access configuration."""
mock_auth_config = Mock(spec=auth_config_create_response.AuthConfig)
mock_auth_config.id = "auth_12345"
mock_response = Mock(spec=auth_config_create_response.AuthConfigCreateResponse)
mock_response.auth_config = mock_auth_config
mock_client.auth_configs.create.return_value = mock_response
options = {
"type": "use_composio_managed_auth",
"name": "Config with Tool Access",
"tool_access_config": {
"tools_for_connected_account_creation": ["GITHUB_CREATE_ISSUE"]
},
}
result = auth_configs.create("github", options)
mock_client.auth_configs.create.assert_called_once()
assert result == mock_auth_config
# Get tests
def test_get_retrieves_auth_config_by_id(
self, auth_configs, mock_client, mock_auth_config_response
):
"""Test retrieving auth config by ID."""
mock_client.auth_configs.retrieve.return_value = mock_auth_config_response
result = auth_configs.get("auth_12345")
mock_client.auth_configs.retrieve.assert_called_once_with("auth_12345")
assert result == mock_auth_config_response
assert result.id == "auth_12345"
assert result.name == "Test Auth Config"
def test_get_handles_not_found_error(self, auth_configs, mock_client):
"""Test get handles API error when auth config not found."""
mock_client.auth_configs.retrieve.side_effect = Exception(
"Auth config not found"
)
with pytest.raises(Exception) as exc_info:
auth_configs.get("nonexistent_auth")
assert "Auth config not found" in str(exc_info.value)
def test_get_with_is_enabled_for_tool_router_true(
self, auth_configs, mock_client, mock_auth_config_response
):
"""Test retrieving auth config with isEnabledForToolRouter set to true."""
mock_auth_config_response.is_enabled_for_tool_router = True
mock_client.auth_configs.retrieve.return_value = mock_auth_config_response
result = auth_configs.get("auth_12345")
assert hasattr(result, "is_enabled_for_tool_router")
assert result.is_enabled_for_tool_router is True
def test_get_with_is_enabled_for_tool_router_false(
self, auth_configs, mock_client, mock_auth_config_response
):
"""Test retrieving auth config with isEnabledForToolRouter set to false."""
mock_auth_config_response.is_enabled_for_tool_router = False
mock_client.auth_configs.retrieve.return_value = mock_auth_config_response
result = auth_configs.get("auth_12345")
assert hasattr(result, "is_enabled_for_tool_router")
assert result.is_enabled_for_tool_router is False
def test_get_with_is_enabled_for_tool_router_undefined(
self, auth_configs, mock_client, mock_auth_config_response
):
"""Test retrieving auth config with isEnabledForToolRouter undefined."""
# Don't set the attribute at all to simulate undefined
if hasattr(mock_auth_config_response, "is_enabled_for_tool_router"):
delattr(mock_auth_config_response, "is_enabled_for_tool_router")
mock_client.auth_configs.retrieve.return_value = mock_auth_config_response
result = auth_configs.get("auth_12345")
# Should not have the attribute or it should be None
assert (
not hasattr(result, "is_enabled_for_tool_router")
or result.is_enabled_for_tool_router is None
)
# Update tests
def test_update_custom_auth_config_with_credentials(
self, auth_configs, mock_client
):
"""Test updating custom auth config with credentials."""
mock_response = {"id": "auth_12345", "status": "success"}
mock_client.auth_configs.update.return_value = mock_response
options = {
"type": "custom",
"credentials": {
"client_id": "new_client_id",
"client_secret": "new_client_secret",
},
}
result = auth_configs.update("auth_12345", options=options)
mock_client.auth_configs.update.assert_called_once()
call_args = mock_client.auth_configs.update.call_args
assert call_args.kwargs["nanoid"] == "auth_12345"
assert call_args.kwargs["type"] == "custom"
assert call_args.kwargs["credentials"] == options["credentials"]
assert result == mock_response
def test_update_default_auth_config_with_scopes(self, auth_configs, mock_client):
"""Test updating default auth config with scopes."""
mock_response = {"id": "auth_12345", "status": "success"}
mock_client.auth_configs.update.return_value = mock_response
options = {
"type": "default",
"scopes": "read:user,repo",
}
result = auth_configs.update("auth_12345", options=options)
mock_client.auth_configs.update.assert_called_once()
# Check that scopes are not directly passed but other fields are
call_args = mock_client.auth_configs.update.call_args
assert call_args.kwargs["nanoid"] == "auth_12345"
assert call_args.kwargs["type"] == "default"
assert result == mock_response
def test_update_with_is_enabled_for_tool_router(self, auth_configs, mock_client):
"""Test updating auth config with isEnabledForToolRouter."""
mock_response = {"id": "auth_12345", "status": "success"}
mock_client.auth_configs.update.return_value = mock_response
options = {
"type": "custom",
"credentials": {"api_key": "new_key"},
"is_enabled_for_tool_router": True,
}
result = auth_configs.update("auth_12345", options=options)
call_args = mock_client.auth_configs.update.call_args
assert call_args.kwargs["is_enabled_for_tool_router"] is True
assert result == mock_response
def test_update_with_tool_access_config(self, auth_configs, mock_client):
"""Test updating auth config with tool access configuration."""
mock_response = {"id": "auth_12345", "status": "success"}
mock_client.auth_configs.update.return_value = mock_response
options = {
"type": "custom",
"credentials": {"api_key": "new_key"},
"tool_access_config": {
"tools_for_connected_account_creation": ["GITHUB_CREATE_ISSUE"]
},
}
result = auth_configs.update("auth_12345", options=options)
call_args = mock_client.auth_configs.update.call_args
assert call_args.kwargs["tool_access_config"] == options["tool_access_config"]
assert result == mock_response
def test_update_with_large_credential_object(self, auth_configs, mock_client):
"""Test updating auth config with large credential object."""
mock_response = {"id": "auth_12345", "status": "success"}
mock_client.auth_configs.update.return_value = mock_response
large_credentials = {
"field1": "value1",
"field2": "value2",
"field3": {"nested": "object"},
"field4": ["array", "values"],
"field5": 12345,
"field6": True,
}
options = {
"type": "custom",
"credentials": large_credentials,
}
result = auth_configs.update("auth_12345", options=options)
call_args = mock_client.auth_configs.update.call_args
assert call_args.kwargs["credentials"] == large_credentials
assert result == mock_response
def test_update_handles_api_error(self, auth_configs, mock_client):
"""Test update handles API errors."""
mock_client.auth_configs.update.side_effect = Exception("Update failed")
options = {
"type": "custom",
"credentials": {"api_key": "key"},
}
with pytest.raises(Exception) as exc_info:
auth_configs.update("auth_12345", options=options)
assert "Update failed" in str(exc_info.value)
# Delete tests
def test_delete_auth_config_by_id(self, auth_configs, mock_client):
"""Test deleting auth config by ID."""
mock_response = {"id": "auth_12345", "status": "deleted"}
mock_client.auth_configs.delete.return_value = mock_response
result = auth_configs.delete("auth_12345")
mock_client.auth_configs.delete.assert_called_once_with("auth_12345")
assert result == mock_response
def test_delete_handles_api_error(self, auth_configs, mock_client):
"""Test delete handles API errors."""
mock_client.auth_configs.delete.side_effect = Exception("Delete failed")
with pytest.raises(Exception) as exc_info:
auth_configs.delete("auth_12345")
assert "Delete failed" in str(exc_info.value)
# Enable/Disable tests
def test_enable_auth_config(self, auth_configs, mock_client):
"""Test enabling auth config."""
mock_response = {"id": "auth_12345", "status": "ENABLED"}
mock_client.auth_configs.update_status.return_value = mock_response
result = auth_configs.enable("auth_12345")
mock_client.auth_configs.update_status.assert_called_once_with(
"ENABLED", nanoid="auth_12345"
)
assert result == mock_response
def test_disable_auth_config(self, auth_configs, mock_client):
"""Test disabling auth config."""
mock_response = {"id": "auth_12345", "status": "DISABLED"}
mock_client.auth_configs.update_status.return_value = mock_response
result = auth_configs.disable("auth_12345")
mock_client.auth_configs.update_status.assert_called_once_with(
"DISABLED", nanoid="auth_12345"
)
assert result == mock_response
def test_enable_handles_api_error(self, auth_configs, mock_client):
"""Test enable handles API errors."""
mock_client.auth_configs.update_status.side_effect = Exception("Enable failed")
with pytest.raises(Exception) as exc_info:
auth_configs.enable("auth_12345")
assert "Enable failed" in str(exc_info.value)
def test_disable_handles_api_error(self, auth_configs, mock_client):
"""Test disable handles API errors."""
mock_client.auth_configs.update_status.side_effect = Exception("Disable failed")
with pytest.raises(Exception) as exc_info:
auth_configs.disable("auth_12345")
assert "Disable failed" in str(exc_info.value)
# Edge cases
def test_update_with_missing_optional_fields(self, auth_configs, mock_client):
"""Test update works with only required fields."""
mock_response = {"id": "auth_12345", "status": "success"}
mock_client.auth_configs.update.return_value = mock_response
options = {
"type": "custom",
"credentials": {"api_key": "key"},
}
result = auth_configs.update("auth_12345", options=options)
call_args = mock_client.auth_configs.update.call_args
# Verify that optional fields use the sentinel value
assert call_args.kwargs["is_enabled_for_tool_router"] == mock_client.not_given
assert call_args.kwargs["tool_access_config"] == mock_client.not_given
assert result == mock_response
def test_create_with_minimal_options(self, auth_configs, mock_client):
"""Test creating auth config with minimal options."""
mock_auth_config = Mock(spec=auth_config_create_response.AuthConfig)
mock_auth_config.id = "auth_12345"
mock_response = Mock(spec=auth_config_create_response.AuthConfigCreateResponse)
mock_response.auth_config = mock_auth_config
mock_client.auth_configs.create.return_value = mock_response
# Just the toolkit, using default type
result = auth_configs.create("github", {"type": "use_composio_managed_auth"})
mock_client.auth_configs.create.assert_called_once()
assert result == mock_auth_config
def test_list_with_pagination(self, auth_configs, mock_client):
"""Test listing auth configs with pagination."""
mock_response = Mock(spec=auth_config_list_response.AuthConfigListResponse)
mock_response.items = [Mock(), Mock(), Mock()]
mock_response.next_cursor = "next_page_cursor"
mock_response.total_pages = 5
mock_client.auth_configs.list.return_value = mock_response
result = auth_configs.list(cursor="current_cursor", limit=3)
mock_client.auth_configs.list.assert_called_once()
assert len(result.items) == 3
assert result.next_cursor == "next_page_cursor"
assert result.total_pages == 5
def test_get_minimal_auth_config(self, auth_configs, mock_client):
"""Test retrieving auth config with minimal fields."""
minimal_response = Mock(
spec=auth_config_retrieve_response.AuthConfigRetrieveResponse
)
minimal_response.id = "auth_minimal"
minimal_response.name = "Minimal Config"
minimal_response.no_of_connections = 0
minimal_response.status = "DISABLED"
minimal_response.toolkit = Mock()
minimal_response.toolkit.logo = ""
minimal_response.toolkit.slug = "minimal-toolkit"
minimal_response.uuid = "uuid-minimal"
mock_client.auth_configs.retrieve.return_value = minimal_response
result = auth_configs.get("auth_minimal")
assert result.id == "auth_minimal"
assert result.name == "Minimal Config"
assert result.no_of_connections == 0
assert result.status == "DISABLED"
class TestAuthConfigsOverloads:
"""Test type overloads for create and update methods."""
@pytest.fixture
def mock_client(self):
"""Create a mock client for testing."""
client = Mock()
client.auth_configs = Mock()
client.auth_configs.create = Mock()
client.auth_configs.update = Mock()
client.not_given = object()
return client
@pytest.fixture
def auth_configs(self, mock_client):
"""Create an AuthConfigs instance with mock client."""
return AuthConfigs(client=mock_client)
def test_create_overload_use_composio_managed_auth(self, auth_configs, mock_client):
"""Test create with use_composio_managed_auth type."""
mock_auth_config = Mock(spec=auth_config_create_response.AuthConfig)
mock_response = Mock(spec=auth_config_create_response.AuthConfigCreateResponse)
mock_response.auth_config = mock_auth_config
mock_client.auth_configs.create.return_value = mock_response
options = {"type": "use_composio_managed_auth"}
result = auth_configs.create("github", options)
assert result == mock_auth_config
def test_create_overload_use_custom_auth(self, auth_configs, mock_client):
"""Test create with use_custom_auth type."""
mock_auth_config = Mock(spec=auth_config_create_response.AuthConfig)
mock_response = Mock(spec=auth_config_create_response.AuthConfigCreateResponse)
mock_response.auth_config = mock_auth_config
mock_client.auth_configs.create.return_value = mock_response
options = {
"type": "use_custom_auth",
"auth_scheme": "API_KEY",
"credentials": {"api_key": "test_key"},
}
result = auth_configs.create("github", options)
assert result == mock_auth_config
def test_update_overload_custom_type(self, auth_configs, mock_client):
"""Test update with custom type."""
mock_response = {"id": "auth_12345", "status": "success"}
mock_client.auth_configs.update.return_value = mock_response
options = {
"type": "custom",
"credentials": {"api_key": "new_key"},
}
result = auth_configs.update("auth_12345", options=options)
assert result == mock_response
def test_update_overload_default_type(self, auth_configs, mock_client):
"""Test update with default type."""
mock_response = {"id": "auth_12345", "status": "success"}
mock_client.auth_configs.update.return_value = mock_response
options = {
"type": "default",
"scopes": "read:user",
}
result = auth_configs.update("auth_12345", options=options)
assert result == mock_response