Blockscout MCP Server

from unittest.mock import AsyncMock, patch import pytest from blockscout_mcp_server.models import ToolResponse, TransactionSummaryData from blockscout_mcp_server.tools.transaction.transaction_summary import transaction_summary @pytest.mark.asyncio async def test_transaction_summary_without_wrapper(mock_ctx): """ Test a transaction tool that doesn't use the periodic progress wrapper for comparison. This helps verify our testing approach for wrapper vs non-wrapper tools. """ # ARRANGE chain_id = "1" tx_hash = "0x123abc" mock_base_url = "https://eth.blockscout.com" summary_obj = {"template": "This is a test transaction summary.", "vars": {}} mock_api_response = {"data": {"summaries": [summary_obj]}} with ( patch( "blockscout_mcp_server.tools.transaction.transaction_summary.get_blockscout_base_url", new_callable=AsyncMock, ) as mock_get_url, patch( "blockscout_mcp_server.tools.transaction.transaction_summary.make_blockscout_request", new_callable=AsyncMock, ) as mock_request, ): mock_get_url.return_value = mock_base_url mock_request.return_value = mock_api_response # ACT result = await transaction_summary(chain_id=chain_id, transaction_hash=tx_hash, ctx=mock_ctx) # ASSERT assert isinstance(result, ToolResponse) assert isinstance(result.data, TransactionSummaryData) assert result.data.summary == [summary_obj] mock_get_url.assert_called_once_with(chain_id) mock_request.assert_called_once_with(base_url=mock_base_url, api_path=f"/api/v2/transactions/{tx_hash}/summary") # This tool should have 3 progress reports (start, after URL, completion) assert mock_ctx.report_progress.await_count == 3 assert mock_ctx.info.await_count == 3 # Verify progress values and message content progress_vals = [c.kwargs["progress"] for c in mock_ctx.report_progress.await_args_list] total_vals = [c.kwargs["total"] for c in mock_ctx.report_progress.await_args_list] assert progress_vals == [0.0, 1.0, 2.0] assert total_vals == [2.0, 2.0, 2.0] info_messages = [c.args[0] for c in mock_ctx.info.await_args_list] assert f"Starting to fetch transaction summary for {tx_hash}" in info_messages[0] assert "Resolved Blockscout instance URL" in info_messages[1] assert "Successfully fetched transaction summary." in info_messages[2] @pytest.mark.asyncio async def test_transaction_summary_no_summary_available(mock_ctx): """ Test transaction_summary when no summary is available in the response. """ # ARRANGE chain_id = "1" tx_hash = "0x123abc" mock_base_url = "https://eth.blockscout.com" # Response with no summary data mock_api_response = {"data": {}} with ( patch( "blockscout_mcp_server.tools.transaction.transaction_summary.get_blockscout_base_url", new_callable=AsyncMock, ) as mock_get_url, patch( "blockscout_mcp_server.tools.transaction.transaction_summary.make_blockscout_request", new_callable=AsyncMock, ) as mock_request, ): mock_get_url.return_value = mock_base_url mock_request.return_value = mock_api_response # ACT result = await transaction_summary(chain_id=chain_id, transaction_hash=tx_hash, ctx=mock_ctx) # ASSERT assert isinstance(result, ToolResponse) assert isinstance(result.data, TransactionSummaryData) assert result.data.summary is None mock_get_url.assert_called_once_with(chain_id) mock_request.assert_called_once_with(base_url=mock_base_url, api_path=f"/api/v2/transactions/{tx_hash}/summary") assert mock_ctx.report_progress.await_count == 3 assert mock_ctx.info.await_count == 3 @pytest.mark.asyncio async def test_transaction_summary_missing_data_key_treated_as_none(mock_ctx): chain_id = "1" tx_hash = "0xno_data" mock_base_url = "https://eth.blockscout.com" mock_api_response = {} with ( patch( "blockscout_mcp_server.tools.transaction.transaction_summary.get_blockscout_base_url", new_callable=AsyncMock, ) as mock_get_url, patch( "blockscout_mcp_server.tools.transaction.transaction_summary.make_blockscout_request", new_callable=AsyncMock, ) as mock_request, ): mock_get_url.return_value = mock_base_url mock_request.return_value = mock_api_response result = await transaction_summary(chain_id=chain_id, transaction_hash=tx_hash, ctx=mock_ctx) assert isinstance(result, ToolResponse) assert isinstance(result.data, TransactionSummaryData) assert result.data.summary is None mock_get_url.assert_called_once_with(chain_id) mock_request.assert_called_once_with(base_url=mock_base_url, api_path=f"/api/v2/transactions/{tx_hash}/summary") assert mock_ctx.report_progress.await_count == 3 assert mock_ctx.info.await_count == 3 @pytest.mark.asyncio async def test_transaction_summary_summary_explicit_none(mock_ctx): chain_id = "1" tx_hash = "0xnone" mock_base_url = "https://eth.blockscout.com" mock_api_response = {"data": {"summaries": None}} with ( patch( "blockscout_mcp_server.tools.transaction.transaction_summary.get_blockscout_base_url", new_callable=AsyncMock, ) as mock_get_url, patch( "blockscout_mcp_server.tools.transaction.transaction_summary.make_blockscout_request", new_callable=AsyncMock, ) as mock_request, ): mock_get_url.return_value = mock_base_url mock_request.return_value = mock_api_response result = await transaction_summary(chain_id=chain_id, transaction_hash=tx_hash, ctx=mock_ctx) assert isinstance(result, ToolResponse) assert isinstance(result.data, TransactionSummaryData) assert result.data.summary is None assert mock_ctx.report_progress.await_count == 3 assert mock_ctx.info.await_count == 3 @pytest.mark.asyncio async def test_transaction_summary_handles_list_summary(mock_ctx): """Verify transaction_summary correctly handles a list-of-dicts summary.""" # ARRANGE chain_id = "1" tx_hash = "0xcomplex" mock_base_url = "https://eth.blockscout.com" complex_summary = [ {"template": "Summary 1", "vars": {"a": 1}}, {"template": "Summary 2", "vars": {"b": 2}}, ] mock_api_response = {"data": {"summaries": complex_summary}} with ( patch( "blockscout_mcp_server.tools.transaction.transaction_summary.get_blockscout_base_url", new_callable=AsyncMock, ) as mock_get_url, patch( "blockscout_mcp_server.tools.transaction.transaction_summary.make_blockscout_request", new_callable=AsyncMock, ) as mock_request, ): mock_get_url.return_value = mock_base_url mock_request.return_value = mock_api_response # ACT result = await transaction_summary(chain_id=chain_id, transaction_hash=tx_hash, ctx=mock_ctx) # ASSERT assert isinstance(result, ToolResponse) assert isinstance(result.data, TransactionSummaryData) assert result.data.summary == complex_summary # Assert it's the original list mock_get_url.assert_called_once_with(chain_id) mock_request.assert_called_once_with(base_url=mock_base_url, api_path=f"/api/v2/transactions/{tx_hash}/summary") assert mock_ctx.report_progress.await_count == 3 assert mock_ctx.info.await_count == 3 @pytest.mark.asyncio async def test_transaction_summary_handles_empty_list(mock_ctx): """Return an empty list when Blockscout summarizes to nothing.""" chain_id = "1" tx_hash = "0xempty" mock_base_url = "https://eth.blockscout.com" mock_api_response = {"data": {"summaries": []}} with ( patch( "blockscout_mcp_server.tools.transaction.transaction_summary.get_blockscout_base_url", new_callable=AsyncMock, ) as mock_get_url, patch( "blockscout_mcp_server.tools.transaction.transaction_summary.make_blockscout_request", new_callable=AsyncMock, ) as mock_request, ): mock_get_url.return_value = mock_base_url mock_request.return_value = mock_api_response result = await transaction_summary(chain_id=chain_id, transaction_hash=tx_hash, ctx=mock_ctx) assert isinstance(result, ToolResponse) assert isinstance(result.data, TransactionSummaryData) assert result.data.summary == [] mock_get_url.assert_called_once_with(chain_id) mock_request.assert_called_once_with(base_url=mock_base_url, api_path=f"/api/v2/transactions/{tx_hash}/summary") assert mock_ctx.report_progress.await_count == 3 assert mock_ctx.info.await_count == 3 @pytest.mark.asyncio async def test_transaction_summary_invalid_format(mock_ctx): """Raise RuntimeError when Blockscout returns unexpected summary format.""" chain_id = "1" tx_hash = "0xdeadbeef" mock_base_url = "https://eth.blockscout.com" mock_api_response = {"data": {"summaries": "unexpected"}} with ( patch( "blockscout_mcp_server.tools.transaction.transaction_summary.get_blockscout_base_url", new_callable=AsyncMock, ) as mock_get_url, patch( "blockscout_mcp_server.tools.transaction.transaction_summary.make_blockscout_request", new_callable=AsyncMock, ) as mock_request, ): mock_get_url.return_value = mock_base_url mock_request.return_value = mock_api_response with pytest.raises(RuntimeError, match="unexpected format"): await transaction_summary(chain_id=chain_id, transaction_hash=tx_hash, ctx=mock_ctx) mock_get_url.assert_called_once_with(chain_id) mock_request.assert_called_once_with( base_url=mock_base_url, api_path=f"/api/v2/transactions/{tx_hash}/summary", ) assert mock_ctx.report_progress.await_count == 3 assert mock_ctx.info.await_count == 3 @pytest.mark.asyncio async def test_transaction_summary_request_error_propagates_and_reports_partial_progress(mock_ctx): chain_id = "1" tx_hash = "0xerr" mock_base_url = "https://eth.blockscout.com" with ( patch( "blockscout_mcp_server.tools.transaction.transaction_summary.get_blockscout_base_url", new_callable=AsyncMock, ) as mock_get_url, patch( "blockscout_mcp_server.tools.transaction.transaction_summary.make_blockscout_request", new_callable=AsyncMock, ) as mock_request, ): mock_get_url.return_value = mock_base_url mock_request.side_effect = RuntimeError("network error") with pytest.raises(RuntimeError, match="network error"): await transaction_summary(chain_id=chain_id, transaction_hash=tx_hash, ctx=mock_ctx) # Only the start and URL resolution steps should be reported when the request fails. assert mock_ctx.report_progress.await_count == 2 progress_vals = [c.kwargs["progress"] for c in mock_ctx.report_progress.await_args_list] total_vals = [c.kwargs["total"] for c in mock_ctx.report_progress.await_args_list] assert progress_vals == [0.0, 1.0] assert total_vals == [2.0, 2.0] assert mock_ctx.info.await_count == 2 info_messages = [call.args[0] for call in mock_ctx.info.await_args_list] assert not any(message.startswith("Successfully fetched transaction summary.") for message in info_messages)