Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 64 additions & 0 deletions tests/test_integrations.py
Original file line number Diff line number Diff line change
Expand Up @@ -609,3 +609,67 @@ async def test_call_tool_empty_arguments_uses_empty_dict(self):

call_args = mock_session.call_tool.call_args
assert call_args.args[1] == {}

"""
Integration tests for CapiscioMCPServer.connect() factory.

Verifies that the factory method:
- Correctly loads identity from environment variables
- Forwards kwargs to the underlying constructor
- Behaves predictably under an existing event loop
- Handles missing/invalid env vars gracefully
"""

import os
import pytest
from unittest.mock import patch


class TestCapiscioMCPServerConnect:
"""Tests for CapiscioMCPServer.connect() factory."""

def test_connect_requires_env(self):
"""connect() should raise when env vars are missing."""
from capiscio_mcp.integrations.mcp import CapiscioMCPServer

with patch.dict(os.environ, {}, clear=True):
with pytest.raises(ValueError, match=".*CAPISCIO_API_KEY.*"):
CapiscioMCPServer.connect()

def test_connect_with_env(self):
"""connect() should succeed with minimal env vars set."""
from capiscio_mcp.integrations.mcp import CapiscioMCPServer

env_vars = {
"CAPISCIO_API_KEY": "test-api-key",
"CAPISCIO_SERVER_ID": "test-server",
}
with patch.dict(os.environ, env_vars, clear=True):
server = CapiscioMCPServer.connect()
assert server is not None
assert server.did is not None
assert "test-server" in server.did

def test_connect_forwards_kwargs(self):
"""connect() should forward extra kwargs to the constructor."""
from capiscio_mcp.integrations.mcp import CapiscioMCPServer

env_vars = {
"CAPISCIO_API_KEY": "test-api-key",
"CAPISCIO_SERVER_ID": "test-server",
}
with patch.dict(os.environ, env_vars, clear=True):
server = CapiscioMCPServer.connect(
default_min_trust_level=2,
name="custom-name",
)
assert server.default_min_trust_level == 2
assert server.name == "custom-name"

def test_aconnect_syntax_is_async(self):
"""aconnect() should be an async classmethod."""
from capiscio_mcp.integrations.mcp import CapiscioMCPServer

assert hasattr(CapiscioMCPServer, "aconnect")
import inspect
assert inspect.iscoroutinefunction(CapiscioMCPServer.aconnect)