ExMCP.TestCase (ex_mcp v0.9.2)

Custom test case for ExMCP with MCP-specific testing utilities.

This module provides a comprehensive testing framework specifically designed for MCP (Model Context Protocol) applications, including:

Custom assertions for MCP protocol validation
Mock server and client infrastructure
Content testing utilities
DSL testing helpers
Performance testing tools
Integration testing support

Usage

defmodule MyMCPTest do
  use ExMCP.TestCase, async: true

  test "tool call validation" do
    tool_result = %{
      "content" => [%{"type" => "text", "text" => "Hello"}]
    }

    assert_valid_tool_result(tool_result)
    assert_content_type(tool_result, :text)
    assert_content_contains(tool_result, "Hello")
  end

  test "server integration" do
    with_mock_server(tools: [sample_tool()]) do |client|
      result = call_tool(client, "sample_tool", %{input: "test"})
      assert_success(result)
    end
  end
end

Summary

Functions

add_cleanup(context, cleanup_fun)

Adds a cleanup function to the test context.

capture_logs(list)

Captures log output during test execution.

concurrent_test(description, opts \\ [], list)

Creates a test that runs concurrently with other tests.

create_test_context(initial_context \\ %{})

Creates a test context with cleanup.

flush_messages()

Flushes all messages from the test process mailbox.

measure_time(list)

Measures execution time of a block.

repeat_test(times, list)

Runs a test multiple times to check for flakiness.

run_parallel(functions, opts \\ [])

Runs tests in parallel and collects results.

start_test_supervisor(child_specs)

Creates a supervision tree for testing.

unique_id(prefix, length \\ 8)

Generates unique test data with a prefix.

wait_until(condition_fun, opts \\ [])

Waits for a condition to become true within a timeout.

with_config(config, list)

Runs a test with specific configuration overrides.

with_mock_server(opts \\ [], list)

Runs a test with a mock MCP server.

with_temp_file(content, extension \\ "", list)

Creates a temporary file with content for testing.

with_test_server(server_module, opts \\ [], list)

Runs a test with a real MCP server process.

with_timeout(timeout_ms, list)

Runs a block with a timeout, failing the test if it takes too long.

Functions

add_cleanup(context, cleanup_fun)

@spec add_cleanup(map(), (-> any())) :: map()

Adds a cleanup function to the test context.

capture_logs(list)

(macro)

Captures log output during test execution.

Examples

logs = capture_logs do
  Logger.info("Test message")
  some_operation()
end

assert logs =~ "Test message"

concurrent_test(description, opts \\ [], list)

(macro)

Creates a test that runs concurrently with other tests.

Examples

concurrent_test "parallel operation", count: 5 do |index|
  result = parallel_operation(index)
  assert result.id == index
end

create_test_context(initial_context \\ %{})

@spec create_test_context(map()) :: map()

Creates a test context with cleanup.

Examples

context = create_test_context(%{
  server_port: 8080,
  temp_dir: "/tmp/test"
})

# Use context in tests
assert context.server_port == 8080

# Cleanup is automatic

flush_messages()

@spec flush_messages() :: [any()]

Flushes all messages from the test process mailbox.

measure_time(list)

(macro)

Measures execution time of a block.

Examples

{result, time_ms} = measure_time do
  heavy_computation()
end

assert time_ms < 1000  # Should complete in under 1 second

repeat_test(times, list)

(macro)

Runs a test multiple times to check for flakiness.

Examples

repeat_test 10 do
  result = potentially_flaky_operation()
  assert result == :ok
end

run_parallel(functions, opts \\ [])

@spec run_parallel(
  [(-> any())],
  keyword()
) :: [any()]

Runs tests in parallel and collects results.

Examples

results = run_parallel([
  fn -> test_operation_1() end,
  fn -> test_operation_2() end,
  fn -> test_operation_3() end
])

assert length(results) == 3
assert Enum.all?(results, &(&1 == :ok))

start_test_supervisor(child_specs)

@spec start_test_supervisor([Supervisor.child_spec()]) ::
  {:ok, pid()} | {:error, any()}

Creates a supervision tree for testing.

Examples

{:ok, supervisor} = start_test_supervisor([
  {MyWorker, [name: :test_worker]},
  {MyServer, [port: 0]}
])

# Test with supervised processes
assert Process.alive?(Process.whereis(:test_worker))

unique_id(prefix, length \\ 8)

@spec unique_id(String.t(), pos_integer()) :: String.t()

Generates unique test data with a prefix.

Examples

id = unique_id("test")          # "test_a1b2c3d4"
name = unique_id("user", 8)     # "user_1a2b3c4d"

wait_until(condition_fun, opts \\ [])

@spec wait_until(
  (-> boolean()),
  keyword()
) :: :ok | :timeout

Waits for a condition to become true within a timeout.

Examples

wait_until(fn -> Process.alive?(pid) end, timeout: 1000)

with_config(config, list)

(macro)

Runs a test with specific configuration overrides.

Examples

with_config([timeout: 10_000, retries: 3]) do
  test_with_custom_config()
end

with_mock_server(opts \\ [], list)

(macro)

Runs a test with a mock MCP server.

Examples

with_mock_server(tools: [sample_tool()], resources: [sample_resource()]) do |client|
  result = ExMCP.Client.list_tools(client)
  assert {:ok, %{"tools" => tools}} = result
  assert length(tools) == 1
end

with_temp_file(content, extension \\ "", list)

(macro)

Creates a temporary file with content for testing.

Examples

with_temp_file("test content", ".txt") do |file_path|
  content = File.read!(file_path)
  assert content == "test content"
end

with_test_server(server_module, opts \\ [], list)

(macro)

Runs a test with a real MCP server process.

Examples

with_test_server(MyServer, port: 8080) do |client|
  result = ExMCP.Client.list_tools(client)
  assert {:ok, %{"tools" => _}} = result
end

with_timeout(timeout_ms, list)

(macro)

Runs a block with a timeout, failing the test if it takes too long.

Examples

with_timeout 5000 do
  slow_operation()
end