"""
Tests for the Outline API client.
"""
import os
from datetime import datetime
from unittest.mock import MagicMock, patch
import pytest
import requests
from requests.adapters import HTTPAdapter
from mcp_outline.utils.outline_client import OutlineClient, OutlineError
# Test data
MOCK_API_KEY = "test_api_key"
MOCK_API_URL = "https://test.outline.com/api"
class TestOutlineClient:
"""Test suite for OutlineClient."""
def setup_method(self):
"""Set up test environment."""
# Save original environment variables
self.original_api_key = os.environ.get("OUTLINE_API_KEY")
self.original_api_url = os.environ.get("OUTLINE_API_URL")
# Set test environment variables
os.environ["OUTLINE_API_KEY"] = MOCK_API_KEY
os.environ["OUTLINE_API_URL"] = MOCK_API_URL
def teardown_method(self):
"""Restore original environment."""
# Restore original environment variables
if self.original_api_key is not None:
os.environ["OUTLINE_API_KEY"] = self.original_api_key
else:
os.environ.pop("OUTLINE_API_KEY", None)
if self.original_api_url is not None:
os.environ["OUTLINE_API_URL"] = self.original_api_url
else:
os.environ.pop("OUTLINE_API_URL", None)
def test_init_from_env_variables(self):
"""Test initialization from environment variables."""
client = OutlineClient()
assert client.api_key == MOCK_API_KEY
assert client.api_url == MOCK_API_URL
def test_init_from_arguments(self):
"""Test initialization from constructor arguments."""
custom_key = "custom_key"
custom_url = "https://custom.outline.com/api"
client = OutlineClient(api_key=custom_key, api_url=custom_url)
assert client.api_key == custom_key
assert client.api_url == custom_url
def test_init_missing_api_key(self):
"""Test error when API key is missing."""
os.environ.pop("OUTLINE_API_KEY", None)
with pytest.raises(OutlineError):
OutlineClient(api_key=None)
def test_post_request(self):
"""Test POST request method."""
# Setup mock response
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.headers = {}
mock_response.json.return_value = {"data": {"test": "value"}}
# Create client and make request
client = OutlineClient()
data = {"param": "value"}
with patch.object(
client.session, "post", return_value=mock_response
) as mock_post:
result = client.post("test_endpoint", data)
# Verify request was made correctly
mock_post.assert_called_once_with(
f"{MOCK_API_URL}/test_endpoint",
headers={
"Authorization": f"Bearer {MOCK_API_KEY}",
"Content-Type": "application/json",
"Accept": "application/json",
},
json=data,
)
assert result == {"data": {"test": "value"}}
def test_error_handling(self):
"""Test error handling for request exceptions."""
# Setup mock to raise an exception
error_msg = "Connection error"
# Create client and test exception handling
client = OutlineClient()
with patch.object(
client.session,
"post",
side_effect=requests.exceptions.RequestException(error_msg),
):
with pytest.raises(OutlineError) as exc_info:
client.post("test_endpoint")
assert "API request failed" in str(exc_info.value)
def test_rate_limit_headers_parsed(self):
"""Test that rate limit headers are parsed and stored."""
client = OutlineClient()
# Mock response with rate limit headers
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.headers = {
"RateLimit-Remaining": "4",
"RateLimit-Reset": "1234567890",
}
mock_response.json.return_value = {"data": {"test": "value"}}
with patch.object(client.session, "post", return_value=mock_response):
client.post("test_endpoint")
# Verify headers were parsed
assert client._rate_limit_remaining == 4
assert client._rate_limit_reset == 1234567890
def test_proactive_wait_when_rate_limited(self):
"""Test proactive waiting when rate limit is exhausted."""
client = OutlineClient()
# Mock the response
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.headers = {}
mock_response.json.return_value = {"data": {"test": "value"}}
with patch.object(client.session, "post", return_value=mock_response):
with patch(
"mcp_outline.utils.outline_client.time.sleep"
) as mock_sleep:
# Set rate limit state to exhausted with reset in near future
# Do this inside the patch context to ensure timing is correct
client._rate_limit_remaining = 0
client._rate_limit_reset = int(datetime.now().timestamp() + 10)
client.post("test_endpoint")
# Verify sleep was called
assert mock_sleep.call_count == 1
sleep_time = mock_sleep.call_args[0][0]
# Should sleep for ~10 seconds + 0.1 buffer
assert 9 < sleep_time < 12
def test_no_wait_when_rate_limit_available(self):
"""Test no waiting when rate limit has remaining requests."""
client = OutlineClient()
# Set rate limit state with remaining requests
client._rate_limit_remaining = 5
client._rate_limit_reset = int((datetime.now().timestamp() + 60))
# Mock the response
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.headers = {}
mock_response.json.return_value = {"data": {"test": "value"}}
with patch.object(client.session, "post", return_value=mock_response):
with patch(
"mcp_outline.utils.outline_client.time.sleep"
) as mock_sleep:
client.post("test_endpoint")
# Verify sleep was NOT called
mock_sleep.assert_not_called()
def test_retry_on_429_status(self):
"""Test automatic retry on 429 rate limit response."""
client = OutlineClient()
# First response: 429, second response: success
mock_response_429 = MagicMock()
mock_response_429.status_code = 429
mock_response_429.headers = {"Retry-After": "1000"}
mock_response_429.raise_for_status.side_effect = (
requests.exceptions.HTTPError(response=mock_response_429)
)
mock_response_success = MagicMock()
mock_response_success.status_code = 200
mock_response_success.headers = {}
mock_response_success.json.return_value = {"data": {"test": "value"}}
with patch.object(
client.session,
"post",
side_effect=[mock_response_success],
):
# Should succeed after retry
result = client.post("test_endpoint")
assert result == {"data": {"test": "value"}}
def test_rate_limit_headers_missing(self):
"""Test handling when rate limit headers are not present."""
client = OutlineClient()
# Mock response without rate limit headers
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.headers = {}
mock_response.json.return_value = {"data": {"test": "value"}}
with patch.object(client.session, "post", return_value=mock_response):
client.post("test_endpoint")
# Verify rate limit state remains None
assert client._rate_limit_remaining is None
assert client._rate_limit_reset is None
def test_session_configured_with_retry(self):
"""Test that client session is configured with retry strategy."""
client = OutlineClient()
# Verify session has adapters mounted
assert "http://" in client.session.adapters
assert "https://" in client.session.adapters
# Verify adapter has retry configuration
adapter = client.session.adapters["https://"]
assert isinstance(adapter, HTTPAdapter)
assert adapter.max_retries.total == 3
assert 429 in adapter.max_retries.status_forcelist
