OpenTelemetry Documentation MCP Server
by ryu1maniwa
test_server.py•8.62 kB
# Copyright 2025 ryu1maniwa. All Rights Reserved.
#
# This file is derived from awslabs.aws-documentation-mcp-server, which is licensed as follows:
#
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"). You may not use this file except in compliance
# with the License. A copy of the License is located at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# or in the 'license' file accompanying this file. This file is distributed on an 'AS IS' BASIS, WITHOUT WARRANTIES
# OR CONDITIONS OF ANY KIND, express or implied. See the License for the specific language governing permissions
# and limitations under the License.
"""Tests for the OpenTelemetry Documentation MCP Server."""
import httpx
import pytest
from opentelemetry_documentation_mcp_server.server import (
read_documentation,
search_documentation,
)
from opentelemetry_documentation_mcp_server.util import extract_content_from_html
from unittest.mock import AsyncMock, MagicMock, patch
class MockContext:
"""Mock context for testing."""
async def error(self, message):
"""Mock error method."""
print(f'Error: {message}')
class TestExtractContentFromHTML:
"""Tests for the extract_content_from_html function."""
@patch('opentelemetry_documentation_mcp_server.util.BeautifulSoup')
@patch('opentelemetry_documentation_mcp_server.util.markdownify.markdownify')
def test_extract_content_from_html(self, mock_markdownify, mock_bs):
"""Test extracting content from HTML."""
html = '<html><body><h1>Test</h1><p>This is a test.</p></body></html>'
# Setup mock
mock_soup_instance = mock_bs.return_value
mock_soup_instance.body = mock_soup_instance
mock_soup_instance.select_one.return_value = None
# Setup mock markdownify return value
mock_markdownify.return_value = '# Test\n\nThis is a test.'
# Call function
result = extract_content_from_html(html)
# Check result
assert result == '# Test\n\nThis is a test.'
mock_bs.assert_called_once()
mock_markdownify.assert_called_once()
@patch('opentelemetry_documentation_mcp_server.util.BeautifulSoup')
def test_extract_content_from_html_no_content(self, mock_bs):
"""Test extracting content from HTML with no content."""
html = '<html><body></body></html>'
# Setup mock
mock_soup_instance = mock_bs.return_value
mock_soup_instance.body = None
# Call function
result = extract_content_from_html(html)
# Check result
assert '<e>' in result
mock_bs.assert_called_once()
class TestReadDocumentation:
"""Tests for the read_documentation function."""
@pytest.mark.asyncio
async def test_read_documentation(self):
"""Test reading OpenTelemetry documentation."""
url = 'https://opentelemetry.io/docs/concepts/overview/'
ctx = MockContext()
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.text = '<html><body><h1>Test</h1><p>This is a test.</p></body></html>'
mock_response.headers = {'content-type': 'text/html'}
with patch('httpx.AsyncClient.get', new_callable=AsyncMock) as mock_get:
mock_get.return_value = mock_response
with patch(
'opentelemetry_documentation_mcp_server.server.extract_content_from_html'
) as mock_extract:
mock_extract.return_value = '# Test\n\nThis is a test.'
result = await read_documentation(ctx, url=url, max_length=10000, start_index=0)
assert 'OpenTelemetry Documentation from' in result
assert '# Test\n\nThis is a test.' in result
mock_get.assert_called_once()
mock_extract.assert_called_once()
@pytest.mark.asyncio
async def test_read_documentation_error(self):
"""Test reading OpenTelemetry documentation with an error."""
url = 'https://opentelemetry.io/docs/concepts/overview/'
ctx = MockContext()
with patch('httpx.AsyncClient.get', new_callable=AsyncMock) as mock_get:
mock_get.side_effect = httpx.HTTPError('Connection error')
result = await read_documentation(ctx, url=url, max_length=10000, start_index=0)
assert 'Failed to fetch' in result
assert 'Connection error' in result
mock_get.assert_called_once()
@pytest.mark.asyncio
async def test_read_documentation_invalid_url(self):
"""Test reading OpenTelemetry documentation with an invalid URL."""
url = 'https://invalid-domain.com/docs/'
ctx = MockContext()
with pytest.raises(ValueError):
await read_documentation(ctx, url=url, max_length=10000, start_index=0)
class TestSearchDocumentation:
"""Tests for the search_documentation function."""
@pytest.mark.asyncio
async def test_search_documentation(self):
"""Test searching OpenTelemetry documentation."""
search_phrase = 'opentelemetry tracing instrumentation'
ctx = MockContext()
# Mock Google Custom Search API response
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"items": [
{
"title": "Tracing Instrumentation",
"link": "https://opentelemetry.io/docs/instrumentation/",
"snippet": "This page describes tracing instrumentation in OpenTelemetry."
},
{
"title": "Concepts",
"link": "https://opentelemetry.io/docs/concepts/",
"snippet": "OpenTelemetry concepts and terminology."
}
]
}
with patch('os.getenv') as mock_getenv:
mock_getenv.return_value = 'fake-api-key' # Mock the API key
with patch('httpx.AsyncClient.get', new_callable=AsyncMock) as mock_get:
mock_get.return_value = mock_response
results = await search_documentation(ctx, search_phrase=search_phrase, limit=10)
assert len(results) == 2
assert results[0].rank_order == 1
assert results[0].url == 'https://opentelemetry.io/docs/instrumentation/'
assert results[0].title == 'Tracing Instrumentation'
assert results[0].context == 'This page describes tracing instrumentation in OpenTelemetry.'
assert results[1].rank_order == 2
assert results[1].url == 'https://opentelemetry.io/docs/concepts/'
assert results[1].title == 'Concepts'
assert results[1].context == 'OpenTelemetry concepts and terminology.'
mock_get.assert_called_once()
@pytest.mark.asyncio
async def test_search_documentation_error(self):
"""Test searching OpenTelemetry documentation with an error."""
search_phrase = 'opentelemetry tracing instrumentation'
ctx = MockContext()
with patch('os.getenv') as mock_getenv:
mock_getenv.return_value = 'fake-api-key' # Mock the API key
with patch('httpx.AsyncClient.get', new_callable=AsyncMock) as mock_get:
mock_get.side_effect = httpx.HTTPError('Connection error')
results = await search_documentation(ctx, search_phrase=search_phrase, limit=10)
# Even with error, the function should return a list with an error message
assert len(results) == 1
assert results[0].rank_order == 1
assert 'Error' in results[0].title
assert results[0].url == ''
mock_get.assert_called_once()
@pytest.mark.asyncio
async def test_search_documentation_no_api_key(self):
"""Test searching OpenTelemetry documentation without API key."""
search_phrase = 'opentelemetry tracing instrumentation'
ctx = MockContext()
with patch('os.getenv') as mock_getenv:
mock_getenv.return_value = None # No API key
results = await search_documentation(ctx, search_phrase=search_phrase, limit=10)
# Should return error message about missing API key
assert len(results) == 1
assert results[0].rank_order == 1
assert 'Google API key not found' in results[0].title
assert results[0].url == ''
Latest Blog Posts
- MCP Moves to the Linux Foundation: Neutral Stewardship for Agentic InfrastructureBy Om-Shree-0709 on .mcpanthropicLinux Foundation
MCP directory API
We provide all the information about MCP servers via our MCP API.
curl -X GET 'https://glama.ai/api/mcp/v1/servers/ryu1maniwa/opentelemetry-documentation-mcp-server'
If you have feedback or need assistance with the MCP directory API, please join our Discord server