Ryan Malloy
212832e7e4
Wire up protocol analysis (parse_protocol_spec, generate_decoder_chain, get_missing_oot_modules), signal analysis (analyze_iq_file), and OOT export (generate_oot_skeleton, export_block_to_oot, export_from_flowgraph) as MCP tools with integration tests. Security fixes from Hamilton review: - Remove `from __future__ import annotations` from tool registration files (breaks FastMCP schema generation) - Add blocklist guard to evaluate_expression (was unsandboxed eval) - Replace string interpolation with base64 encoding in Docker test harness (prevents code injection) - Add try/finally cleanup for temp files and Docker containers - Replace assert with proper ValueError in flowgraph block creation - Log OOT auto-discovery failures instead of swallowing silently Packaging: - Move entry point to src/gnuradio_mcp/server.py with script entry point (uv run gnuradio-mcp) - Add PyPI metadata (authors, license, classifiers, urls) - Add MIT LICENSE file - Rewrite README for current feature set (80+ tools) - Document single-session limitation
53 lines
1.7 KiB
Python
53 lines
1.7 KiB
Python
import pytest
|
|
from fastmcp import Client
|
|
|
|
from gnuradio_mcp.server import app as mcp_app
|
|
|
|
|
|
@pytest.fixture
|
|
async def main_mcp_client():
|
|
async with Client(mcp_app) as mcp_client:
|
|
yield mcp_client
|
|
|
|
|
|
async def test_get_blocks(main_mcp_client: Client):
|
|
"""Test retrieving available blocks."""
|
|
blocks = await main_mcp_client.call_tool(name="get_blocks")
|
|
assert blocks.data is not None
|
|
# We don't know exactly what blocks are there, but should get a list
|
|
assert isinstance(blocks.data, list)
|
|
|
|
|
|
async def test_make_and_remove_block(main_mcp_client: Client):
|
|
"""Test making a block and then removing it."""
|
|
block_type = "analog_sig_source_x"
|
|
|
|
# helper to check if block exists in the flowgraph
|
|
async def get_block_names():
|
|
current_blocks = await main_mcp_client.call_tool(name="get_blocks")
|
|
# FastMCP may return dicts or Pydantic models depending on serialization
|
|
return [b["name"] if isinstance(b, dict) else b.name for b in current_blocks.data]
|
|
|
|
# 1. Create a block
|
|
result = await main_mcp_client.call_tool(
|
|
name="make_block", arguments={"block_name": block_type}
|
|
)
|
|
assert result.data is not None
|
|
# The output is the new block name, likely something like "analog_sig_source_x_0"
|
|
new_block_name = str(result.data)
|
|
assert block_type in new_block_name
|
|
|
|
# Verify it exists
|
|
block_names = await get_block_names()
|
|
assert new_block_name in block_names
|
|
|
|
# 2. Remove the block
|
|
remove_result = await main_mcp_client.call_tool(
|
|
name="remove_block", arguments={"block_name": new_block_name}
|
|
)
|
|
assert remove_result.data is True
|
|
|
|
# Verify it's gone
|
|
block_names = await get_block_names()
|
|
assert new_block_name not in block_names
|