From 88ab7c3683be4b712501dbb2d22c7e590c649516 Mon Sep 17 00:00:00 2001 From: detti456 Date: Tue, 9 Dec 2025 11:18:10 +0100 Subject: [PATCH 1/4] test: add integration tests for AWS MCP --- examples/mcp-client/langchain/main.py | 6 - .../middleware/initialize_middleware.py | 2 +- tests/integ/conftest.py | 12 ++ tests/integ/mcp/simple_mcp_client.py | 2 +- tests/integ/test_aws_mcp_server.py | 138 ++++++++++++++++++ tests/integ/test_aws_mcp_server_negative.py | 132 +++++++++++++++++ uv.lock | 4 +- 7 files changed, 286 insertions(+), 10 deletions(-) create mode 100644 tests/integ/test_aws_mcp_server.py create mode 100644 tests/integ/test_aws_mcp_server_negative.py diff --git a/examples/mcp-client/langchain/main.py b/examples/mcp-client/langchain/main.py index 88b00d0..ac54cf6 100644 --- a/examples/mcp-client/langchain/main.py +++ b/examples/mcp-client/langchain/main.py @@ -26,12 +26,6 @@ - MCP_REGION: AWS region where the MCP server is hosted (e.g., "us-west-2") 3. Run: `uv run main.py` -Example .env file: -================== -MCP_SERVER_URL=https://example.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp -MCP_SERVER_AWS_SERVICE=bedrock-agentcore -MCP_SERVER_REGION=us-west-2 - Example .env file: ================== MCP_SERVER_URL=https://example.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp diff --git a/mcp_proxy_for_aws/middleware/initialize_middleware.py b/mcp_proxy_for_aws/middleware/initialize_middleware.py index 06fa277..b39edd5 100644 --- a/mcp_proxy_for_aws/middleware/initialize_middleware.py +++ b/mcp_proxy_for_aws/middleware/initialize_middleware.py @@ -9,7 +9,7 @@ class InitializeMiddleware(Middleware): - """Intecept MCP initialize request and initialize the proxy client.""" + """Intercept MCP initialize request and initialize the proxy client.""" def __init__(self, client_factory: AWSMCPProxyClientFactory) -> None: """Create a middleware with client factory.""" diff --git a/tests/integ/conftest.py b/tests/integ/conftest.py index 441d1ce..5205002 100644 --- a/tests/integ/conftest.py +++ b/tests/integ/conftest.py @@ -90,3 +90,15 @@ def _build_endpoint_environment_remote_configuration(): endpoint=remote_endpoint_url, region_name=region_name, ) + + +@pytest_asyncio.fixture(loop_scope="module", scope="module") +async def aws_mcp_client(): + """Create MCP Client for AWS MCP Server.""" + client = build_mcp_client( + endpoint="https://aws-mcp.us-east-1.api.aws/mcp", + region_name="us-east-1", + ) + + async with client: + yield client \ No newline at end of file diff --git a/tests/integ/mcp/simple_mcp_client.py b/tests/integ/mcp/simple_mcp_client.py index 6657c63..07eaa25 100644 --- a/tests/integ/mcp/simple_mcp_client.py +++ b/tests/integ/mcp/simple_mcp_client.py @@ -27,7 +27,7 @@ def build_mcp_client( **_build_mcp_config(endpoint=endpoint, region_name=region_name, metadata=metadata) ), elicitation_handler=_basic_elicitation_handler, - timeout=10.0, # seconds + timeout=20.0, # seconds ) diff --git a/tests/integ/test_aws_mcp_server.py b/tests/integ/test_aws_mcp_server.py new file mode 100644 index 0000000..db02525 --- /dev/null +++ b/tests/integ/test_aws_mcp_server.py @@ -0,0 +1,138 @@ +"""Integration tests for AWS MCP Server at https://aws-mcp.us-east-1.api.aws/mcp.""" + +import fastmcp +import logging +import pytest +from fastmcp.client.client import CallToolResult + +from tests.integ.test_proxy_simple_mcp_server import get_text_content +import json + + +logger = logging.getLogger(__name__) + + +@pytest.mark.asyncio(loop_scope="module") +async def test_aws_mcp_ping(aws_mcp_client: fastmcp.Client): + """Test ping to AWS MCP Server.""" + await aws_mcp_client.ping() + +@pytest.mark.asyncio(loop_scope="module") +async def test_aws_mcp_list_tools(aws_mcp_client: fastmcp.Client): + """Test list tools from AWS MCP Server.""" + tools = await aws_mcp_client.list_tools() + + assert len(tools) > 0, f"AWS MCP Server should have tools (got {len(tools)})" + + +def verify_response_content(response: CallToolResult): + """Verify that a tool call response is successful and contains text content. + + Args: + response: The CallToolResult from an MCP tool call + + Returns: + str: The extracted text content from the response + + Raises: + AssertionError: If response indicates an error or has empty content + """ + assert ( + response.is_error is False + ), f"is_error returned true. Returned response body: {response}." + assert len(response.content) > 0, f"Empty result list in response. Response: {response}" + + response_text = get_text_content(response) + assert len(response_text) > 0, f"Empty response text. Response: {response}" + + return response_text + +def verify_json_response(response: CallToolResult): + """Verify that a tool call response is successful and contains valid JSON data. + + Args: + response: The CallToolResult from an MCP tool call + + Raises: + AssertionError: If response indicates an error, has empty content, + contains invalid JSON, or JSON data is empty + """ + response_text = verify_response_content(response) + + # Verify response_text is valid JSON + try: + response_data = json.loads(response_text) + except json.JSONDecodeError: + raise AssertionError(f"Response text is not valid JSON. Response text: {response_text}") + + assert len(response_data) > 0, f"Empty JSON content in response. Response: {response}" + + +@pytest.mark.parametrize( + "tool_name,params", + [ + ("aws___list_regions", {}), + ("aws___suggest_aws_commands", {"query": "how to list my lambda functions"}), + ("aws___search_documentation", {"search_phrase": "S3 bucket versioning"}), + ( + "aws___recommend", + {"url": "bad_url"}, + ), + ( + "aws___read_documentation", + {"url": "https://docs.aws.amazon.com/lambda/latest/dg/lambda-invocation.html"}, + ), + ( + "aws___get_regional_availability", + {"resource_type": "cfn", "region": "us-east-1"}, + ), + ("aws___call_aws", {"cli_command": "aws s3 ls", "max_results": 50}), + ], + ids=[ + "list_regions", + "suggest_aws_commands", + "search_documentation", + "recommend", + "read_documentation", + "get_regional_availability", + "call_aws", + ], +) +@pytest.mark.asyncio(loop_scope="module") +async def test_aws_mcp_tools(aws_mcp_client: fastmcp.Client, tool_name: str, params: dict): + """Test AWS MCP tools with minimal valid params.""" + response = await aws_mcp_client.call_tool(tool_name, params) + verify_json_response(response) + + +@pytest.mark.asyncio(loop_scope="module") +async def test_aws_mcp_tools_retrieve_agent_sop(aws_mcp_client: fastmcp.Client): + """Test aws___retrieve_agent_sop by retrieving the list of available SOPs.""" + + # Step 1: Call retrieve_agent_sop with empty params to get list of available SOPs + list_sops_response = await aws_mcp_client.call_tool("aws___retrieve_agent_sop") + + list_sops_response_text = verify_response_content(list_sops_response) + + # Parse SOP names from text (format: "* sop_name : description") + sop_names = [] + for line in list_sops_response_text.split("\n"): + line = line.strip() + if line.startswith("*") and ":" in line: + # Extract the SOP name between '*' and ':' + sop_name = line.split("*", 1)[1].split(":", 1)[0].strip() + if sop_name: + sop_names.append(sop_name) + + assert ( + len(sop_names) > 0 + ), f"No SOPs found in response. Response: {list_sops_response_text[:200]}..." + logger.info(f"Found {len(sop_names)} SOPs: {sop_names}") + + # Step 2: Test retrieving the first SOP + test_script = sop_names[0] + logger.info(f"Testing with SOP: {test_script}") + + response = await aws_mcp_client.call_tool("aws___retrieve_agent_sop", {"sop_name": test_script}) + + verify_response_content(response) diff --git a/tests/integ/test_aws_mcp_server_negative.py b/tests/integ/test_aws_mcp_server_negative.py new file mode 100644 index 0000000..ed39a8a --- /dev/null +++ b/tests/integ/test_aws_mcp_server_negative.py @@ -0,0 +1,132 @@ +"""Negative integration tests for AWS MCP Server at https://aws-mcp.us-east-1.api.aws/mcp.""" + +import fastmcp +import logging +import pytest +import boto3 +from fastmcp.client import StdioTransport +from datetime import datetime, timedelta + +logger = logging.getLogger(__name__) + + +@pytest.mark.asyncio(loop_scope="module") +async def test_nonexistent_tool(aws_mcp_client: fastmcp.Client): + """Test that calling a nonexistent tool raises an exception. + + This test will: + - PASS when calling a nonexistent tool raises an exception + - FAIL if the nonexistent tool somehow succeeds + """ + exception_raised = False + exception_message = None + + try: + response = await aws_mcp_client.call_tool("aws___nonexistent_tool_12345", {}) + logger.info(f"Unexpected success, response: {response}") + except Exception as e: + exception_raised = True + exception_message = str(e) + logger.info(f"Exception raised as expected: {type(e).__name__}: {exception_message}") + + # Assert that an exception was raised + assert exception_raised, ( + f"Expected exception when calling nonexistent tool 'aws___nonexistent_tool_12345', " + f"but call succeeded unexpectedly." + ) + + # Verify the exception mentions the tool not being found + error_message_lower = exception_message.lower() + tool_error_patterns = [ + "not found", + "unknown", + "does not exist", + "invalid tool", + "tool", + "unknown tool", + ] + + assert any(pattern in error_message_lower for pattern in tool_error_patterns), ( + f"Exception was raised but doesn't appear to be about a missing tool. " + f"Expected one of {tool_error_patterns}, but got: {exception_message[:200]}" + ) + + logger.info(f"Test passed: Nonexistent tool correctly raised exception") + + +@pytest.mark.asyncio(loop_scope="module") +async def test_expired_credentials(): + """Test that expired credentials are properly rejected. + + This test uses real AWS credentials but modifies the session token to simulate + an expired token, which should result in an 'expired token' error message. + + This test will: + - PASS when expired credentials are rejected with appropriate error + - FAIL if the modified credentials somehow work + """ + + # Get real credentials from boto3 + session = boto3.Session() + creds = session.get_credentials() + + # Use real access key and secret, but modify the token to simulate expiration by changing a few characters + expired_token = creds.token[:-20] + "EXPIRED_TOKEN_12345" + + expired_client = fastmcp.Client( + StdioTransport( + command="mcp-proxy-for-aws", + args=[ + "https://aws-mcp.us-east-1.api.aws/mcp", + "--log-level", + "DEBUG", + "--region", + "us-east-1", + ], + env={ + "AWS_REGION": "us-east-1", + "AWS_ACCESS_KEY_ID": creds.access_key, + "AWS_SECRET_ACCESS_KEY": creds.secret_key, + "AWS_SESSION_TOKEN": expired_token, + }, + ), + timeout=30.0, + ) + + exception_raised = False + exception_message = None + + try: + async with expired_client: + response = await expired_client.call_tool("aws___list_regions") + logger.info(f"Tool call completed without exception: is_error={response.is_error}") + except Exception as e: + exception_raised = True + exception_message = str(e) + logger.info(f"Exception raised as expected: {type(e).__name__}: {exception_message}") + + # Assert that an exception was raised (credentials are invalid) + assert exception_raised, ( + f"Expected authentication exception with invalid credentials, " f"but tool call succeeded." + ) + + # Verify the exception is related to authentication/credentials + error_message_lower = exception_message.lower() + auth_error_patterns = [ + "credential", + "authentication", + "authorization", + "access denied", + "unauthorized", + "invalid", + "expired", + "signature", + "401", + ] + + assert any(pattern in error_message_lower for pattern in auth_error_patterns), ( + f"Exception was raised but doesn't appear to be authentication-related. " + f"Expected one of {auth_error_patterns}, but got: {exception_message[:200]}" + ) + + logger.info(f"Test passed: Invalid credentials correctly rejected") diff --git a/uv.lock b/uv.lock index df83e59..45ee02a 100644 --- a/uv.lock +++ b/uv.lock @@ -2458,8 +2458,8 @@ dev = [ [package.metadata] requires-dist = [ - { name = "boto3", specifier = ">=1.34.0" }, - { name = "botocore", extras = ["crt"], specifier = ">=1.34.0" }, + { name = "boto3", specifier = ">=1.41.0" }, + { name = "botocore", extras = ["crt"], specifier = ">=1.41.0" }, { name = "fastmcp", specifier = ">=2.13.1" }, ] From ae9ca478de1f43db0587a561fec7c53215fdc0b4 Mon Sep 17 00:00:00 2001 From: detti456 Date: Tue, 9 Dec 2025 13:31:48 +0100 Subject: [PATCH 2/4] test: fix AWS MCP integration test setup --- ...r.py => test_aws_mcp_server_happy_path.py} | 6 +-- tests/integ/test_aws_mcp_server_negative.py | 45 ------------------- 2 files changed, 3 insertions(+), 48 deletions(-) rename tests/integ/{test_aws_mcp_server.py => test_aws_mcp_server_happy_path.py} (95%) diff --git a/tests/integ/test_aws_mcp_server.py b/tests/integ/test_aws_mcp_server_happy_path.py similarity index 95% rename from tests/integ/test_aws_mcp_server.py rename to tests/integ/test_aws_mcp_server_happy_path.py index db02525..0598f6d 100644 --- a/tests/integ/test_aws_mcp_server.py +++ b/tests/integ/test_aws_mcp_server_happy_path.py @@ -1,4 +1,4 @@ -"""Integration tests for AWS MCP Server at https://aws-mcp.us-east-1.api.aws/mcp.""" +"""Happy path integration tests for AWS MCP Server at https://aws-mcp.us-east-1.api.aws/mcp.""" import fastmcp import logging @@ -76,7 +76,7 @@ def verify_json_response(response: CallToolResult): ("aws___search_documentation", {"search_phrase": "S3 bucket versioning"}), ( "aws___recommend", - {"url": "bad_url"}, + {"url": "https://docs.aws.amazon.com/lambda/latest/dg/lambda-invocation.html"}, ), ( "aws___read_documentation", @@ -86,7 +86,7 @@ def verify_json_response(response: CallToolResult): "aws___get_regional_availability", {"resource_type": "cfn", "region": "us-east-1"}, ), - ("aws___call_aws", {"cli_command": "aws s3 ls", "max_results": 50}), + ("aws___call_aws", {"cli_command": "aws s3 ls", "max_results": 10}), ], ids=[ "list_regions", diff --git a/tests/integ/test_aws_mcp_server_negative.py b/tests/integ/test_aws_mcp_server_negative.py index ed39a8a..e514809 100644 --- a/tests/integ/test_aws_mcp_server_negative.py +++ b/tests/integ/test_aws_mcp_server_negative.py @@ -9,51 +9,6 @@ logger = logging.getLogger(__name__) - -@pytest.mark.asyncio(loop_scope="module") -async def test_nonexistent_tool(aws_mcp_client: fastmcp.Client): - """Test that calling a nonexistent tool raises an exception. - - This test will: - - PASS when calling a nonexistent tool raises an exception - - FAIL if the nonexistent tool somehow succeeds - """ - exception_raised = False - exception_message = None - - try: - response = await aws_mcp_client.call_tool("aws___nonexistent_tool_12345", {}) - logger.info(f"Unexpected success, response: {response}") - except Exception as e: - exception_raised = True - exception_message = str(e) - logger.info(f"Exception raised as expected: {type(e).__name__}: {exception_message}") - - # Assert that an exception was raised - assert exception_raised, ( - f"Expected exception when calling nonexistent tool 'aws___nonexistent_tool_12345', " - f"but call succeeded unexpectedly." - ) - - # Verify the exception mentions the tool not being found - error_message_lower = exception_message.lower() - tool_error_patterns = [ - "not found", - "unknown", - "does not exist", - "invalid tool", - "tool", - "unknown tool", - ] - - assert any(pattern in error_message_lower for pattern in tool_error_patterns), ( - f"Exception was raised but doesn't appear to be about a missing tool. " - f"Expected one of {tool_error_patterns}, but got: {exception_message[:200]}" - ) - - logger.info(f"Test passed: Nonexistent tool correctly raised exception") - - @pytest.mark.asyncio(loop_scope="module") async def test_expired_credentials(): """Test that expired credentials are properly rejected. From 5d686cc84985d93b479bd08b535945b25667f100 Mon Sep 17 00:00:00 2001 From: detti456 Date: Tue, 9 Dec 2025 14:15:03 +0100 Subject: [PATCH 3/4] style: fix AWS MCP integration tests formatting --- tests/integ/conftest.py | 8 +- tests/integ/test_aws_mcp_server_happy_path.py | 88 ++++++++++--------- tests/integ/test_aws_mcp_server_negative.py | 62 ++++++------- 3 files changed, 80 insertions(+), 78 deletions(-) diff --git a/tests/integ/conftest.py b/tests/integ/conftest.py index 5205002..d43b80e 100644 --- a/tests/integ/conftest.py +++ b/tests/integ/conftest.py @@ -92,13 +92,13 @@ def _build_endpoint_environment_remote_configuration(): ) -@pytest_asyncio.fixture(loop_scope="module", scope="module") +@pytest_asyncio.fixture(loop_scope='module', scope='module') async def aws_mcp_client(): """Create MCP Client for AWS MCP Server.""" client = build_mcp_client( - endpoint="https://aws-mcp.us-east-1.api.aws/mcp", - region_name="us-east-1", + endpoint='https://aws-mcp.us-east-1.api.aws/mcp', + region_name='us-east-1', ) async with client: - yield client \ No newline at end of file + yield client diff --git a/tests/integ/test_aws_mcp_server_happy_path.py b/tests/integ/test_aws_mcp_server_happy_path.py index 0598f6d..cbffece 100644 --- a/tests/integ/test_aws_mcp_server_happy_path.py +++ b/tests/integ/test_aws_mcp_server_happy_path.py @@ -1,28 +1,28 @@ """Happy path integration tests for AWS MCP Server at https://aws-mcp.us-east-1.api.aws/mcp.""" import fastmcp +import json import logging import pytest from fastmcp.client.client import CallToolResult - from tests.integ.test_proxy_simple_mcp_server import get_text_content -import json logger = logging.getLogger(__name__) -@pytest.mark.asyncio(loop_scope="module") +@pytest.mark.asyncio(loop_scope='module') async def test_aws_mcp_ping(aws_mcp_client: fastmcp.Client): """Test ping to AWS MCP Server.""" await aws_mcp_client.ping() -@pytest.mark.asyncio(loop_scope="module") + +@pytest.mark.asyncio(loop_scope='module') async def test_aws_mcp_list_tools(aws_mcp_client: fastmcp.Client): """Test list tools from AWS MCP Server.""" tools = await aws_mcp_client.list_tools() - assert len(tools) > 0, f"AWS MCP Server should have tools (got {len(tools)})" + assert len(tools) > 0, f'AWS MCP Server should have tools (got {len(tools)})' def verify_response_content(response: CallToolResult): @@ -37,16 +37,17 @@ def verify_response_content(response: CallToolResult): Raises: AssertionError: If response indicates an error or has empty content """ - assert ( - response.is_error is False - ), f"is_error returned true. Returned response body: {response}." - assert len(response.content) > 0, f"Empty result list in response. Response: {response}" + assert response.is_error is False, ( + f'is_error returned true. Returned response body: {response}.' + ) + assert len(response.content) > 0, f'Empty result list in response. Response: {response}' response_text = get_text_content(response) - assert len(response_text) > 0, f"Empty response text. Response: {response}" + assert len(response_text) > 0, f'Empty response text. Response: {response}' return response_text + def verify_json_response(response: CallToolResult): """Verify that a tool call response is successful and contains valid JSON data. @@ -63,76 +64,77 @@ def verify_json_response(response: CallToolResult): try: response_data = json.loads(response_text) except json.JSONDecodeError: - raise AssertionError(f"Response text is not valid JSON. Response text: {response_text}") + raise AssertionError(f'Response text is not valid JSON. Response text: {response_text}') - assert len(response_data) > 0, f"Empty JSON content in response. Response: {response}" + assert len(response_data) > 0, f'Empty JSON content in response. Response: {response}' @pytest.mark.parametrize( - "tool_name,params", + 'tool_name,params', [ - ("aws___list_regions", {}), - ("aws___suggest_aws_commands", {"query": "how to list my lambda functions"}), - ("aws___search_documentation", {"search_phrase": "S3 bucket versioning"}), + ('aws___list_regions', {}), + ('aws___suggest_aws_commands', {'query': 'how to list my lambda functions'}), + ('aws___search_documentation', {'search_phrase': 'S3 bucket versioning'}), ( - "aws___recommend", - {"url": "https://docs.aws.amazon.com/lambda/latest/dg/lambda-invocation.html"}, + 'aws___recommend', + {'url': 'https://docs.aws.amazon.com/lambda/latest/dg/lambda-invocation.html'}, ), ( - "aws___read_documentation", - {"url": "https://docs.aws.amazon.com/lambda/latest/dg/lambda-invocation.html"}, + 'aws___read_documentation', + {'url': 'https://docs.aws.amazon.com/lambda/latest/dg/lambda-invocation.html'}, ), ( - "aws___get_regional_availability", - {"resource_type": "cfn", "region": "us-east-1"}, + 'aws___get_regional_availability', + {'resource_type': 'cfn', 'region': 'us-east-1'}, ), - ("aws___call_aws", {"cli_command": "aws s3 ls", "max_results": 10}), + ('aws___call_aws', {'cli_command': 'aws s3 ls', 'max_results': 10}), ], ids=[ - "list_regions", - "suggest_aws_commands", - "search_documentation", - "recommend", - "read_documentation", - "get_regional_availability", - "call_aws", + 'list_regions', + 'suggest_aws_commands', + 'search_documentation', + 'recommend', + 'read_documentation', + 'get_regional_availability', + 'call_aws', ], ) -@pytest.mark.asyncio(loop_scope="module") +@pytest.mark.asyncio(loop_scope='module') async def test_aws_mcp_tools(aws_mcp_client: fastmcp.Client, tool_name: str, params: dict): """Test AWS MCP tools with minimal valid params.""" response = await aws_mcp_client.call_tool(tool_name, params) verify_json_response(response) -@pytest.mark.asyncio(loop_scope="module") +@pytest.mark.asyncio(loop_scope='module') async def test_aws_mcp_tools_retrieve_agent_sop(aws_mcp_client: fastmcp.Client): """Test aws___retrieve_agent_sop by retrieving the list of available SOPs.""" - # Step 1: Call retrieve_agent_sop with empty params to get list of available SOPs - list_sops_response = await aws_mcp_client.call_tool("aws___retrieve_agent_sop") + list_sops_response = await aws_mcp_client.call_tool('aws___retrieve_agent_sop') list_sops_response_text = verify_response_content(list_sops_response) # Parse SOP names from text (format: "* sop_name : description") sop_names = [] - for line in list_sops_response_text.split("\n"): + for line in list_sops_response_text.split('\n'): line = line.strip() - if line.startswith("*") and ":" in line: + if line.startswith('*') and ':' in line: # Extract the SOP name between '*' and ':' - sop_name = line.split("*", 1)[1].split(":", 1)[0].strip() + sop_name = line.split('*', 1)[1].split(':', 1)[0].strip() if sop_name: sop_names.append(sop_name) - assert ( - len(sop_names) > 0 - ), f"No SOPs found in response. Response: {list_sops_response_text[:200]}..." - logger.info(f"Found {len(sop_names)} SOPs: {sop_names}") + assert len(sop_names) > 0, ( + f'No SOPs found in response. Response: {list_sops_response_text[:200]}...' + ) + logger.info('Found %d SOPs: %s', len(sop_names), sop_names) # Step 2: Test retrieving the first SOP test_script = sop_names[0] - logger.info(f"Testing with SOP: {test_script}") + logger.info('Testing with SOP: %s', test_script) - response = await aws_mcp_client.call_tool("aws___retrieve_agent_sop", {"sop_name": test_script}) + response = await aws_mcp_client.call_tool( + 'aws___retrieve_agent_sop', {'sop_name': test_script} + ) verify_response_content(response) diff --git a/tests/integ/test_aws_mcp_server_negative.py b/tests/integ/test_aws_mcp_server_negative.py index e514809..55c64f7 100644 --- a/tests/integ/test_aws_mcp_server_negative.py +++ b/tests/integ/test_aws_mcp_server_negative.py @@ -1,15 +1,16 @@ """Negative integration tests for AWS MCP Server at https://aws-mcp.us-east-1.api.aws/mcp.""" +import boto3 import fastmcp import logging import pytest -import boto3 from fastmcp.client import StdioTransport -from datetime import datetime, timedelta + logger = logging.getLogger(__name__) -@pytest.mark.asyncio(loop_scope="module") + +@pytest.mark.asyncio(loop_scope='module') async def test_expired_credentials(): """Test that expired credentials are properly rejected. @@ -20,68 +21,67 @@ async def test_expired_credentials(): - PASS when expired credentials are rejected with appropriate error - FAIL if the modified credentials somehow work """ - # Get real credentials from boto3 session = boto3.Session() creds = session.get_credentials() # Use real access key and secret, but modify the token to simulate expiration by changing a few characters - expired_token = creds.token[:-20] + "EXPIRED_TOKEN_12345" + expired_token = 'EXPIRED_TOKEN_12345' expired_client = fastmcp.Client( StdioTransport( - command="mcp-proxy-for-aws", + command='mcp-proxy-for-aws', args=[ - "https://aws-mcp.us-east-1.api.aws/mcp", - "--log-level", - "DEBUG", - "--region", - "us-east-1", + 'https://aws-mcp.us-east-1.api.aws/mcp', + '--log-level', + 'DEBUG', + '--region', + 'us-east-1', ], env={ - "AWS_REGION": "us-east-1", - "AWS_ACCESS_KEY_ID": creds.access_key, - "AWS_SECRET_ACCESS_KEY": creds.secret_key, - "AWS_SESSION_TOKEN": expired_token, + 'AWS_REGION': 'us-east-1', + 'AWS_ACCESS_KEY_ID': creds.access_key, + 'AWS_SECRET_ACCESS_KEY': creds.secret_key, + 'AWS_SESSION_TOKEN': expired_token, }, ), timeout=30.0, ) exception_raised = False - exception_message = None + exception_message = '' try: async with expired_client: - response = await expired_client.call_tool("aws___list_regions") - logger.info(f"Tool call completed without exception: is_error={response.is_error}") + response = await expired_client.call_tool('aws___list_regions') + logger.info('Tool call completed without exception. Response: %s', response) except Exception as e: exception_raised = True exception_message = str(e) - logger.info(f"Exception raised as expected: {type(e).__name__}: {exception_message}") + logger.info('Exception raised as expected: %s: %s', type(e).__name__, exception_message) # Assert that an exception was raised (credentials are invalid) assert exception_raised, ( - f"Expected authentication exception with invalid credentials, " f"but tool call succeeded." + 'Expected authentication exception with invalid credentials, but tool call succeeded.' ) # Verify the exception is related to authentication/credentials error_message_lower = exception_message.lower() auth_error_patterns = [ - "credential", - "authentication", - "authorization", - "access denied", - "unauthorized", - "invalid", - "expired", - "signature", - "401", + 'credential', + 'authentication', + 'authorization', + 'access denied', + 'unauthorized', + 'invalid', + 'expired', + 'signature', + '401', ] assert any(pattern in error_message_lower for pattern in auth_error_patterns), ( f"Exception was raised but doesn't appear to be authentication-related. " - f"Expected one of {auth_error_patterns}, but got: {exception_message[:200]}" + f'Expected one of {auth_error_patterns}, but got: {exception_message[:200]}' ) - logger.info(f"Test passed: Invalid credentials correctly rejected") + logger.info('Test passed: Invalid credentials correctly rejected') From 1b5d22a9f5897232f36c8f2af13b47aa0d967e7f Mon Sep 17 00:00:00 2001 From: detti456 Date: Tue, 9 Dec 2025 15:20:42 +0100 Subject: [PATCH 4/4] style: add license header to newly created files --- tests/integ/conftest.py | 14 ++++++++++++++ tests/integ/mcp/simple_mcp_client.py | 14 ++++++++++++++ tests/integ/test_aws_mcp_server_happy_path.py | 14 ++++++++++++++ tests/integ/test_aws_mcp_server_negative.py | 14 ++++++++++++++ 4 files changed, 56 insertions(+) diff --git a/tests/integ/conftest.py b/tests/integ/conftest.py index d43b80e..d03feaa 100644 --- a/tests/integ/conftest.py +++ b/tests/integ/conftest.py @@ -1,3 +1,17 @@ +# 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. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + import logging import os import pytest diff --git a/tests/integ/mcp/simple_mcp_client.py b/tests/integ/mcp/simple_mcp_client.py index 07eaa25..db2c133 100644 --- a/tests/integ/mcp/simple_mcp_client.py +++ b/tests/integ/mcp/simple_mcp_client.py @@ -1,3 +1,17 @@ +# 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. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + import boto3 import fastmcp import logging diff --git a/tests/integ/test_aws_mcp_server_happy_path.py b/tests/integ/test_aws_mcp_server_happy_path.py index cbffece..ac1ca31 100644 --- a/tests/integ/test_aws_mcp_server_happy_path.py +++ b/tests/integ/test_aws_mcp_server_happy_path.py @@ -1,3 +1,17 @@ +# 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. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + """Happy path integration tests for AWS MCP Server at https://aws-mcp.us-east-1.api.aws/mcp.""" import fastmcp diff --git a/tests/integ/test_aws_mcp_server_negative.py b/tests/integ/test_aws_mcp_server_negative.py index 55c64f7..f07a79c 100644 --- a/tests/integ/test_aws_mcp_server_negative.py +++ b/tests/integ/test_aws_mcp_server_negative.py @@ -1,3 +1,17 @@ +# 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. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + """Negative integration tests for AWS MCP Server at https://aws-mcp.us-east-1.api.aws/mcp.""" import boto3