FrankfurterMCP

import asyncio import json import logging import pytest from fastmcp import Client, FastMCP from frankfurtermcp.common import AppMetadata from frankfurtermcp.middleware import StripUnknownArgumentsMiddleware from frankfurtermcp.model import ResponseMetadata from frankfurtermcp.server import FrankfurterMCP, app as frankfurtermcp_app logger = logging.getLogger(__name__) class TestMCPServer: """Test suite for the Frankfurter MCP server.""" @pytest.fixture(scope="class", autouse=True) @classmethod def mcp_server(cls): """Fixture to register features in an MCP server.""" server = FastMCP() mcp_obj = FrankfurterMCP() server_with_features = mcp_obj.register_features(server) server_with_features.add_middleware(StripUnknownArgumentsMiddleware()) return server_with_features @pytest.fixture(scope="class") @classmethod def mcp_server_bogus_config(cls): """Fixture to register features in an MCP server with a bogus configuration.""" server = FastMCP() mcp_obj = FrankfurterMCP() mcp_obj.frankfurter_api_url = "http://127.0.0.1:12345/nonexistent_endpoint" server_with_features = mcp_obj.register_features(server) # We don't care about the middleware for this bogus config test return server_with_features @pytest.fixture(scope="class", autouse=True) @classmethod def mcp_client(cls, mcp_server): """Fixture to create a client for the MCP server.""" mcp_client = Client( transport=mcp_server, timeout=60, ) return mcp_client @pytest.fixture(scope="class") @classmethod def mcp_client_bogus_config(cls, mcp_server_bogus_config): """Fixture to create a client for the MCP server with a bogus config.""" mcp_client = Client( transport=mcp_server_bogus_config, timeout=60, ) return mcp_client async def call_tool(self, tool_name: str, mcp_client: Client, **kwargs): """Helper method to call a tool on the MCP server.""" logger.debug(f"Calling tool '{tool_name}' with arguments: {kwargs}") async with mcp_client: result = await mcp_client.call_tool(tool_name, arguments=kwargs) await mcp_client.close() logger.debug(f"Tool '{tool_name}' returned result: {result}") return result def test_app_creation(self): """Test the creation of the FastMCP application.""" mcp_server = frankfurtermcp_app() assert isinstance(mcp_server, FastMCP), "Expected mcp_server to be an instance of FastMCP" tools_list = asyncio.run(mcp_server.get_tools()) assert len(tools_list) == 5, ( f"Expected exactly 5 tools to be registered in the MCP server, got {len(tools_list)}" ) def test_get_supported_currencies(self, mcp_client, mcp_client_bogus_config): """Test the get_supported_currencies function to ensure it returns a list of supported currencies.""" test_method = "get_supported_currencies" response = asyncio.run( self.call_tool( tool_name=test_method, mcp_client=mcp_client, ) ) assert isinstance(response.meta, dict), ( f"Expected result from MCP tool call, {test_method}, to contain a metadata dictionary" ) try: ResponseMetadata.model_validate(response.meta[AppMetadata.PACKAGE_NAME]) except Exception as e: # pragma: no cover assert False, f"Metadata validation failed with error: {e}" json_result: dict = json.loads(response.content[0].text) assert len(json_result.keys()) > 0, "Expected non-empty list of currencies" assert all((isinstance(code, str) and len(code) == 3) for code in json_result.keys()), ( "All currency codes should be 3-character strings" ) with pytest.raises(Exception) as exc_info: asyncio.run( self.call_tool( tool_name=test_method, mcp_client=mcp_client_bogus_config, ) ) assert "Connection refused" in str(exc_info.value), ( "Expected exception message to indicate failure to connect to bogus API endpoint" ) def test_convert_currency_latest(self, mcp_client): """Test the convert_currency_latest function to ensure it returns a list of supported currencies.""" test_method = "convert_currency_latest" response = asyncio.run( self.call_tool( tool_name=test_method, mcp_client=mcp_client, from_currency="GBP", to_currency="JPY", amount=100.0, ) ) assert isinstance(response.meta, dict), ( f"Expected result from MCP tool call, {test_method}, to contain a metadata dictionary" ) try: ResponseMetadata.model_validate(response.meta[AppMetadata.PACKAGE_NAME]) except Exception as e: # pragma: no cover assert False, f"Metadata validation failed with error: {e}" json_result: dict = json.loads(response.content[0].text) logger.info(f"{test_method} response: {json_result}") assert isinstance(json_result["converted_amount"], float), "Expected float value for converted amount" assert json_result["converted_amount"] > 100.0, "The exchange rate for GBP to JPY should be greater than 1.0" # Run again to test cache hit response = asyncio.run( self.call_tool( tool_name=test_method, mcp_client=mcp_client, from_currency="GBP", to_currency="JPY", amount=100.0, ) ) parsed_metadata = ResponseMetadata.model_validate(response.meta[AppMetadata.PACKAGE_NAME]) assert parsed_metadata.cached_response is True, "Expected cached_response to be True on second call" # Test with the same currency for from and to to raise an expected exception with pytest.raises(Exception) as exc_info: asyncio.run( self.call_tool( tool_name=test_method, mcp_client=mcp_client, from_currency="GBP", to_currency="GBP", amount=100.0, ) ) assert "Source currency 'GBP' and target currency 'GBP' are the same." in str(exc_info.value), ( "Expected exception message to indicate failure to retrieve exchange rates" ) def test_get_latest_exchange_rates(self, mcp_client, mcp_client_bogus_config, benchmark): """Test the get_latest_exchange_rates function to ensure that it returns the list of latest rates with other currencies.""" test_method = "get_latest_exchange_rates" def bench_func(): return asyncio.run( self.call_tool( tool_name=test_method, mcp_client=mcp_client, base_currency="JPY", symbols=["EUR", "GBP", "CHF", "NZD"], ) ) response = benchmark.pedantic(bench_func, iterations=1, rounds=5, warmup_rounds=1) assert isinstance(response.meta, dict), ( f"Expected result from MCP tool call, {test_method}, to contain a metadata dictionary" ) try: ResponseMetadata.model_validate(response.meta[AppMetadata.PACKAGE_NAME]) except Exception as e: assert False, f"Metadata validation failed with error: {e}" json_result: dict = json.loads(response.content[0].text) assert len(json_result["rates"].keys()) > 0, "Expected non-empty list of currency rates" assert all((isinstance(code, str) and len(code) == 3) for code in json_result["rates"].keys()), ( "All currency codes for exchange rates should be 3-character strings" ) # Run again to test cache hit response = asyncio.run( self.call_tool( tool_name=test_method, mcp_client=mcp_client, base_currency="JPY", symbols=["EUR", "GBP", "CHF", "NZD"], ) ) parsed_metadata = ResponseMetadata.model_validate(response.meta[AppMetadata.PACKAGE_NAME]) assert parsed_metadata.cached_response is True, "Expected cached_response to be True on second call" with pytest.raises(Exception) as exc_info: asyncio.run( self.call_tool( tool_name=test_method, mcp_client=mcp_client_bogus_config, base_currency="JPY", symbols=["EUR", "GBP", "CHF", "NZD"], ) ) assert "Connection refused" in str(exc_info.value), ( "Expected exception message to indicate failure to connect to bogus API endpoint" ) def test_get_historical_exchange_rates(self, mcp_client, mcp_client_bogus_config, benchmark): """Test the get_historical_exchange_rates function to ensure that it returns the list of historical rates with other currencies.""" test_method = "get_historical_exchange_rates" def bench_func(): return asyncio.run( self.call_tool( tool_name=test_method, mcp_client=mcp_client, base_currency="JPY", start_date="2025-06-01", end_date="2025-06-19", symbols=["EUR", "GBP", "CHF", "NZD"], ) ) response = benchmark.pedantic(bench_func, iterations=1, rounds=5, warmup_rounds=1) assert isinstance(response.meta, dict), ( f"Expected result from MCP tool call, {test_method}, to contain a metadata dictionary" ) try: ResponseMetadata.model_validate(response.meta[AppMetadata.PACKAGE_NAME]) except Exception as e: assert False, f"Metadata validation failed with error: {e}" json_result: dict = json.loads(response.content[0].text) assert all(len(rates_for_date) > 0 for _, rates_for_date in json_result["rates"].items()), ( "Expected non-empty list of currency rates" ) assert all( ((isinstance(code, str) and len(code) == 3) for code in rates_for_date.keys()) for _, rates_for_date in json_result["rates"].items() ), "All currency codes for exchange rates should be 3-character strings" try: asyncio.run( self.call_tool( tool_name=test_method, mcp_client=mcp_client, base_currency="JPY", start_date="2025-06-01", end_date="2025-06-19", symbols="EUR", ) ) except Exception as e: assert False, f"MCP tool call with single currency symbol should not have raised exception: {e}" try: asyncio.run( self.call_tool( tool_name=test_method, mcp_client=mcp_client, base_currency="JPY", start_date="2025-08-31", symbols=["NOK"], ) ) except Exception as e: assert False, f"MCP tool call with only start_date and no end_date should not have raised exception: {e}" with pytest.raises(Exception) as exc_info: asyncio.run( self.call_tool( tool_name=test_method, mcp_client=mcp_client, base_currency="JPY", symbols=["EUR", "GBP", "CHF", "NZD"], ) ) assert "You must provide either a specific date, a start date, or a date range." in str(exc_info.value), ( "Expected exception message to indicate failure to retrieve exchange rates" ) with pytest.raises(Exception) as exc_info: asyncio.run( self.call_tool( tool_name=test_method, mcp_client=mcp_client_bogus_config, base_currency="JPY", start_date="2025-06-01", end_date="2025-06-19", symbols=["EUR", "GBP", "CHF", "NZD"], ) ) assert "Connection refused" in str(exc_info.value), ( "Expected exception message to indicate failure to connect to bogus API endpoint" ) def test_get_latest_exchange_rates_for_single_currency(self, mcp_client): """Test the get_latest_exchange_rates function to ensure that it returns the latest rates for a single currency.""" test_method = "get_latest_exchange_rates" response = asyncio.run( self.call_tool( tool_name=test_method, mcp_client=mcp_client, base_currency="JPY", symbols="GBP", ) ) assert isinstance(response.meta, dict), ( f"Expected result from MCP tool call, {test_method}, to contain a metadata dictionary" ) try: ResponseMetadata.model_validate(response.meta[AppMetadata.PACKAGE_NAME]) except Exception as e: assert False, f"Metadata validation failed with error: {e}" json_result: dict = json.loads(response.content[0].text) assert len(json_result["rates"].keys()) > 0, "Expected non-empty list of currency rates" assert all((isinstance(code, str) and len(code) == 3) for code in json_result["rates"].keys()), ( "All currency codes for exchange rates should be 3-character strings" ) def test_get_historical_exchange_rates_for_a_single_currency(self, mcp_client): """Test the get_historical_exchange_rates function to ensure that it returns the historical rates for a single currency.""" test_method = "get_historical_exchange_rates" response = asyncio.run( self.call_tool( tool_name=test_method, mcp_client=mcp_client, base_currency="JPY", start_date="2025-06-01", end_date="2025-06-19", symbols=["EUR", "GBP", "CHF", "NZD"], ) ) assert isinstance(response.meta, dict), ( f"Expected result from MCP tool call, {test_method}, to contain a metadata dictionary" ) try: ResponseMetadata.model_validate(response.meta[AppMetadata.PACKAGE_NAME]) except Exception as e: assert False, f"Metadata validation failed with error: {e}" json_result: dict = json.loads(response.content[0].text) assert all(len(rates_for_date) > 0 for _, rates_for_date in json_result["rates"].items()), ( "Expected non-empty list of currency rates" ) assert all( ((isinstance(code, str) and len(code) == 3) for code in rates_for_date.keys()) for _, rates_for_date in json_result["rates"].items() ), "All currency codes for exchange rates should be 3-character strings" def test_convert_currency_specific_date(self, mcp_client): """Test the convert_currency_specific_date function to ensure it returns a list of supported currencies.""" test_method = "convert_currency_specific_date" response = asyncio.run( self.call_tool( tool_name=test_method, mcp_client=mcp_client, from_currency="GBP", to_currency="JPY", amount=100.0, specific_date="2025-06-01", ) ) assert isinstance(response.meta, dict), ( f"Expected result from MCP tool call, {test_method}, to contain a metadata dictionary" ) try: ResponseMetadata.model_validate(response.meta[AppMetadata.PACKAGE_NAME]) except Exception as e: assert False, f"Metadata validation failed with error: {e}" json_result: dict = json.loads(response.content[0].text) print(f"{test_method} response: {json_result}") assert isinstance(json_result["converted_amount"], float), "Expected float value for converted amount" assert json_result["converted_amount"] > 100.0, "The exchange rate for GBP to JPY should be greater than 1.0" # Run again to test cache hit response = asyncio.run( self.call_tool( tool_name=test_method, mcp_client=mcp_client, from_currency="GBP", to_currency="JPY", amount=100.0, specific_date="2025-06-01", ) ) parsed_metadata = ResponseMetadata.model_validate(response.meta[AppMetadata.PACKAGE_NAME]) assert parsed_metadata.cached_response is True, "Expected cached_response to be True on second call" # Test with the same currency for from and to to raise an expected exception with pytest.raises(Exception) as exc_info: asyncio.run( self.call_tool( tool_name=test_method, mcp_client=mcp_client, from_currency="GBP", to_currency="GBP", amount=100.0, specific_date="2025-06-01", ) ) assert "Source currency 'GBP' and target currency 'GBP' are the same." in str(exc_info.value), ( "Expected exception message to indicate failure to retrieve exchange rates" )