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
576 lines
22 KiB
Python
576 lines
22 KiB
Python
"""Integration tests for McpBlockDevProvider."""
|
|
|
|
import pytest
|
|
from fastmcp import Client, FastMCP
|
|
|
|
from gnuradio_mcp.providers.mcp_block_dev import McpBlockDevProvider
|
|
|
|
|
|
@pytest.fixture
|
|
def mcp_app():
|
|
"""Create a FastMCP app with block dev provider."""
|
|
app = FastMCP("Block Dev Test")
|
|
McpBlockDevProvider.create(app)
|
|
return app
|
|
|
|
|
|
class TestDynamicBlockDevMode:
|
|
"""Tests for dynamic block development mode registration."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_block_dev_mode_starts_disabled(self, mcp_app):
|
|
"""Block dev mode should be disabled by default."""
|
|
async with Client(mcp_app) as client:
|
|
result = await client.call_tool(name="get_block_dev_mode")
|
|
assert result.data.enabled is False
|
|
assert result.data.tools_registered == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enable_block_dev_mode_registers_tools(self, mcp_app):
|
|
"""Enabling block dev mode should register generation tools."""
|
|
async with Client(mcp_app) as client:
|
|
result = await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
assert result.data.enabled is True
|
|
assert "generate_sync_block" in result.data.tools_registered
|
|
assert "validate_block_code" in result.data.tools_registered
|
|
assert "parse_block_prompt" in result.data.tools_registered
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disable_block_dev_mode_removes_tools(self, mcp_app):
|
|
"""Disabling block dev mode should remove generation tools."""
|
|
async with Client(mcp_app) as client:
|
|
# Enable first
|
|
await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
# Then disable
|
|
result = await client.call_tool(name="disable_block_dev_mode")
|
|
|
|
assert result.data.enabled is False
|
|
assert result.data.tools_registered == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enable_block_dev_mode_idempotent(self, mcp_app):
|
|
"""Enabling block dev mode multiple times should be idempotent."""
|
|
async with Client(mcp_app) as client:
|
|
result1 = await client.call_tool(name="enable_block_dev_mode")
|
|
result2 = await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
assert result1.data.enabled == result2.data.enabled
|
|
assert set(result1.data.tools_registered) == set(result2.data.tools_registered)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disable_block_dev_mode_idempotent(self, mcp_app):
|
|
"""Disabling block dev mode multiple times should be idempotent."""
|
|
async with Client(mcp_app) as client:
|
|
result1 = await client.call_tool(name="disable_block_dev_mode")
|
|
result2 = await client.call_tool(name="disable_block_dev_mode")
|
|
|
|
assert result1.data.enabled is False
|
|
assert result2.data.enabled is False
|
|
|
|
|
|
class TestBlockDevTools:
|
|
"""Tests for block development tools when enabled."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_sync_block_creates_valid_code(self, mcp_app):
|
|
"""Generate a sync block and verify it validates."""
|
|
async with Client(mcp_app) as client:
|
|
# Enable block dev mode
|
|
await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
# Generate a block
|
|
result = await client.call_tool(
|
|
name="generate_sync_block",
|
|
arguments={
|
|
"name": "test_gain",
|
|
"description": "Multiply by gain factor",
|
|
"inputs": [{"dtype": "float", "vlen": 1}],
|
|
"outputs": [{"dtype": "float", "vlen": 1}],
|
|
"parameters": [{"name": "gain", "dtype": "float", "default": 1.0}],
|
|
"work_template": "gain",
|
|
},
|
|
)
|
|
|
|
assert result.data.is_valid is True
|
|
assert "gr.sync_block" in result.data.source_code
|
|
assert "self.gain" in result.data.source_code
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_basic_block_creates_valid_code(self, mcp_app):
|
|
"""Generate a basic block and verify it validates."""
|
|
async with Client(mcp_app) as client:
|
|
await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
result = await client.call_tool(
|
|
name="generate_basic_block",
|
|
arguments={
|
|
"name": "packet_extract",
|
|
"description": "Extract packets",
|
|
"inputs": [{"dtype": "byte", "vlen": 1}],
|
|
"outputs": [{"dtype": "byte", "vlen": 1}],
|
|
"parameters": [],
|
|
"work_logic": "self.consume_each(1); return 1",
|
|
},
|
|
)
|
|
|
|
assert result.data.is_valid is True
|
|
assert "gr.basic_block" in result.data.source_code
|
|
assert "general_work" in result.data.source_code
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_block_code_success(self, mcp_app):
|
|
"""Validate syntactically correct code."""
|
|
async with Client(mcp_app) as client:
|
|
await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
valid_code = '''
|
|
import numpy
|
|
from gnuradio import gr
|
|
|
|
class blk(gr.sync_block):
|
|
def __init__(self):
|
|
gr.sync_block.__init__(
|
|
self, name="test", in_sig=[numpy.float32], out_sig=[numpy.float32]
|
|
)
|
|
|
|
def work(self, input_items, output_items):
|
|
output_items[0][:] = input_items[0]
|
|
return len(output_items[0])
|
|
'''
|
|
result = await client.call_tool(
|
|
name="validate_block_code",
|
|
arguments={"source_code": valid_code},
|
|
)
|
|
|
|
assert result.data.is_valid is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_block_code_syntax_error(self, mcp_app):
|
|
"""Validate code with syntax errors."""
|
|
async with Client(mcp_app) as client:
|
|
await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
invalid_code = """
|
|
def broken(:
|
|
pass
|
|
"""
|
|
result = await client.call_tool(
|
|
name="validate_block_code",
|
|
arguments={"source_code": invalid_code},
|
|
)
|
|
|
|
assert result.data.is_valid is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_interp_block(self, mcp_app):
|
|
"""Generate an interpolating block."""
|
|
async with Client(mcp_app) as client:
|
|
await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
result = await client.call_tool(
|
|
name="generate_interp_block",
|
|
arguments={
|
|
"name": "upsample_2x",
|
|
"description": "Upsample by 2",
|
|
"inputs": [{"dtype": "float", "vlen": 1}],
|
|
"outputs": [{"dtype": "float", "vlen": 1}],
|
|
"interpolation": 2,
|
|
"parameters": [],
|
|
},
|
|
)
|
|
|
|
assert result.data.is_valid is True
|
|
assert "gr.interp_block" in result.data.source_code
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_decim_block(self, mcp_app):
|
|
"""Generate a decimating block."""
|
|
async with Client(mcp_app) as client:
|
|
await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
result = await client.call_tool(
|
|
name="generate_decim_block",
|
|
arguments={
|
|
"name": "downsample_4x",
|
|
"description": "Downsample by 4",
|
|
"inputs": [{"dtype": "float", "vlen": 1}],
|
|
"outputs": [{"dtype": "float", "vlen": 1}],
|
|
"decimation": 4,
|
|
"parameters": [],
|
|
},
|
|
)
|
|
|
|
assert result.data.is_valid is True
|
|
assert "gr.decim_block" in result.data.source_code
|
|
|
|
|
|
class TestBlockDevResources:
|
|
"""Tests for block dev prompt template resources."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_block_prompt_resource(self, mcp_app):
|
|
"""Verify sync block prompt resource is available."""
|
|
async with Client(mcp_app) as client:
|
|
resources = await client.list_resources()
|
|
resource_uris = [r.uri for r in resources]
|
|
|
|
# Check that our prompt resources are registered
|
|
assert any("sync-block" in str(uri) for uri in resource_uris)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_basic_block_prompt_resource(self, mcp_app):
|
|
"""Verify basic block prompt resource is available."""
|
|
async with Client(mcp_app) as client:
|
|
resources = await client.list_resources()
|
|
resource_uris = [r.uri for r in resources]
|
|
|
|
assert any("basic-block" in str(uri) for uri in resource_uris)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_decoder_chain_prompt_resource(self, mcp_app):
|
|
"""Verify decoder chain prompt resource is available."""
|
|
async with Client(mcp_app) as client:
|
|
resources = await client.list_resources()
|
|
resource_uris = [r.uri for r in resources]
|
|
|
|
assert any("decoder-chain" in str(uri) for uri in resource_uris)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_common_patterns_prompt_resource(self, mcp_app):
|
|
"""Verify common patterns prompt resource is available."""
|
|
async with Client(mcp_app) as client:
|
|
resources = await client.list_resources()
|
|
resource_uris = [r.uri for r in resources]
|
|
|
|
assert any("common-patterns" in str(uri) for uri in resource_uris)
|
|
|
|
|
|
class TestToolNotAvailableWhenDisabled:
|
|
"""Tests that tools are not available when block dev mode is disabled."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_sync_block_not_available_when_disabled(self, mcp_app):
|
|
"""generate_sync_block should not be callable when mode is disabled."""
|
|
async with Client(mcp_app) as client:
|
|
# Ensure disabled
|
|
await client.call_tool(name="disable_block_dev_mode")
|
|
|
|
# List available tools
|
|
tools = await client.list_tools()
|
|
tool_names = [t.name for t in tools]
|
|
|
|
# Generation tools should not be in the list
|
|
assert "generate_sync_block" not in tool_names
|
|
assert "validate_block_code" not in tool_names
|
|
assert "generate_basic_block" not in tool_names
|
|
|
|
# But mode control tools should be
|
|
assert "get_block_dev_mode" in tool_names
|
|
assert "enable_block_dev_mode" in tool_names
|
|
assert "disable_block_dev_mode" in tool_names
|
|
|
|
|
|
class TestProtocolAnalysisTools:
|
|
"""Tests for protocol analysis and signal detection tools."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parse_protocol_spec_gfsk(self, mcp_app):
|
|
"""Parse a GFSK protocol specification."""
|
|
async with Client(mcp_app) as client:
|
|
await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
# Note: Parser expects "Xk baud" and "deviation: Xkhz" format
|
|
result = await client.call_tool(
|
|
name="parse_protocol_spec",
|
|
arguments={
|
|
"spec_text": "GFSK signal at 250k baud, deviation: 160khz"
|
|
},
|
|
)
|
|
|
|
assert result.data.modulation.scheme == "GFSK"
|
|
assert result.data.modulation.symbol_rate == 250000.0
|
|
assert result.data.modulation.deviation == 160000.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parse_protocol_spec_lora(self, mcp_app):
|
|
"""Parse a LoRa/CSS protocol specification."""
|
|
async with Client(mcp_app) as client:
|
|
await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
result = await client.call_tool(
|
|
name="parse_protocol_spec",
|
|
arguments={
|
|
"spec_text": """
|
|
Protocol: LoRa
|
|
Modulation: CSS (Chirp Spread Spectrum)
|
|
Bandwidth: 125 kHz
|
|
Preamble: 8 upchirps
|
|
Sync word: 0x34
|
|
"""
|
|
},
|
|
)
|
|
|
|
assert result.data.name == "LoRa"
|
|
assert result.data.modulation.scheme == "CSS"
|
|
assert result.data.modulation.bandwidth == 125000.0
|
|
assert result.data.framing is not None
|
|
assert result.data.framing.sync_word == "0x34"
|
|
assert result.data.framing.preamble_length == 8
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parse_protocol_spec_with_fec(self, mcp_app):
|
|
"""Parse protocol with FEC encoding."""
|
|
async with Client(mcp_app) as client:
|
|
await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
result = await client.call_tool(
|
|
name="parse_protocol_spec",
|
|
arguments={
|
|
"spec_text": """
|
|
FSK at 9600 baud
|
|
Hamming FEC with rate 3/4
|
|
Data whitening enabled
|
|
"""
|
|
},
|
|
)
|
|
|
|
assert result.data.modulation.scheme == "FSK"
|
|
assert result.data.encoding is not None
|
|
assert result.data.encoding.fec_type == "hamming"
|
|
assert result.data.encoding.fec_rate == "3/4"
|
|
assert result.data.encoding.whitening is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_decoder_chain_gfsk(self, mcp_app):
|
|
"""Generate decoder chain for GFSK signal."""
|
|
async with Client(mcp_app) as client:
|
|
await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
# First parse a protocol (using parser's expected format)
|
|
parse_result = await client.call_tool(
|
|
name="parse_protocol_spec",
|
|
arguments={
|
|
"spec_text": "GFSK at 50k baud, deviation: 25khz"
|
|
},
|
|
)
|
|
|
|
# Generate decoder chain from parsed protocol
|
|
# Note: Use structured_content (already a dict) for passing to next tool
|
|
result = await client.call_tool(
|
|
name="generate_decoder_chain",
|
|
arguments={
|
|
"protocol": parse_result.structured_content,
|
|
"sample_rate": 2000000.0,
|
|
},
|
|
)
|
|
|
|
# Should have demodulation blocks
|
|
block_types = [b.block_type for b in result.data.blocks]
|
|
assert "analog_quadrature_demod_cf" in block_types
|
|
assert "digital_symbol_sync_ff" in block_types
|
|
assert "digital_binary_slicer_fb" in block_types
|
|
|
|
# Should have connections
|
|
assert len(result.data.connections) >= 2
|
|
|
|
# Should have sample rate variable
|
|
# Note: Access via structured_content since data wraps nested objects
|
|
assert result.structured_content["variables"]["samp_rate"] == 2000000.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_decoder_chain_with_framing(self, mcp_app):
|
|
"""Generate decoder with sync word correlation."""
|
|
async with Client(mcp_app) as client:
|
|
await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
# Use parser's expected format for baud rate
|
|
parse_result = await client.call_tool(
|
|
name="parse_protocol_spec",
|
|
arguments={
|
|
"spec_text": """
|
|
FSK at 9.6k baud
|
|
Sync word: 0x2DD4
|
|
Preamble: 10101010 pattern
|
|
"""
|
|
},
|
|
)
|
|
|
|
result = await client.call_tool(
|
|
name="generate_decoder_chain",
|
|
arguments={"protocol": parse_result.structured_content},
|
|
)
|
|
|
|
# Should have correlator for sync word
|
|
block_types = [b.block_type for b in result.data.blocks]
|
|
assert "digital_correlate_access_code_tag_bb" in block_types
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_missing_oot_modules_lora(self, mcp_app):
|
|
"""Identify missing OOT modules for LoRa decoder."""
|
|
async with Client(mcp_app) as client:
|
|
await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
# Parse LoRa protocol (requires gr-lora_sdr)
|
|
parse_result = await client.call_tool(
|
|
name="parse_protocol_spec",
|
|
arguments={
|
|
"spec_text": """
|
|
Protocol: LoRa
|
|
CSS modulation
|
|
Bandwidth: 125 kHz
|
|
"""
|
|
},
|
|
)
|
|
|
|
pipeline_result = await client.call_tool(
|
|
name="generate_decoder_chain",
|
|
arguments={"protocol": parse_result.structured_content},
|
|
)
|
|
|
|
# Check for missing OOT modules
|
|
result = await client.call_tool(
|
|
name="get_missing_oot_modules",
|
|
arguments={"pipeline": pipeline_result.structured_content},
|
|
)
|
|
|
|
# LoRa blocks require gr-lora_sdr OOT module
|
|
# The pipeline should indicate lora_sdr_demod is missing
|
|
# which maps to gr-lora_sdr module
|
|
# (Only if not installed - test checks the mapping works)
|
|
assert isinstance(result.data, list)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_protocol_analysis_tools_registered(self, mcp_app):
|
|
"""Verify protocol analysis tools are registered when enabled."""
|
|
async with Client(mcp_app) as client:
|
|
result = await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
tool_names = result.data.tools_registered
|
|
|
|
# Protocol analysis tools (Phase 3)
|
|
assert "parse_protocol_spec" in tool_names
|
|
assert "generate_decoder_chain" in tool_names
|
|
assert "get_missing_oot_modules" in tool_names
|
|
|
|
# Signal analysis tools (Phase 4)
|
|
assert "analyze_iq_file" in tool_names
|
|
|
|
# OOT export tools (Phase 5)
|
|
assert "generate_oot_skeleton" in tool_names
|
|
assert "export_block_to_oot" in tool_names
|
|
|
|
|
|
class TestOOTExportTools:
|
|
"""Tests for OOT module export workflow."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_oot_skeleton(self, mcp_app, tmp_path):
|
|
"""Generate an empty OOT module skeleton."""
|
|
async with Client(mcp_app) as client:
|
|
await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
output_dir = str(tmp_path / "gr-test")
|
|
|
|
result = await client.call_tool(
|
|
name="generate_oot_skeleton",
|
|
arguments={
|
|
"module_name": "test",
|
|
"output_dir": output_dir,
|
|
"author": "Test Author",
|
|
"description": "Test module",
|
|
},
|
|
)
|
|
|
|
assert result.data.success is True
|
|
assert result.data.module_name == "test"
|
|
|
|
# Check files were created
|
|
assert (tmp_path / "gr-test" / "CMakeLists.txt").exists()
|
|
assert (tmp_path / "gr-test" / "python" / "test" / "__init__.py").exists()
|
|
assert (tmp_path / "gr-test" / "grc" / "CMakeLists.txt").exists()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_export_block_to_oot(self, mcp_app, tmp_path):
|
|
"""Export a generated block to OOT module."""
|
|
async with Client(mcp_app) as client:
|
|
await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
# First generate a block
|
|
block_result = await client.call_tool(
|
|
name="generate_sync_block",
|
|
arguments={
|
|
"name": "my_gain",
|
|
"description": "Multiply by gain",
|
|
"inputs": [{"dtype": "float", "vlen": 1}],
|
|
"outputs": [{"dtype": "float", "vlen": 1}],
|
|
"parameters": [{"name": "gain", "dtype": "float", "default": 1.0}],
|
|
"work_template": "gain",
|
|
},
|
|
)
|
|
|
|
# Export to OOT
|
|
output_dir = str(tmp_path / "gr-custom")
|
|
|
|
result = await client.call_tool(
|
|
name="export_block_to_oot",
|
|
arguments={
|
|
"generated": block_result.structured_content,
|
|
"module_name": "custom",
|
|
"output_dir": output_dir,
|
|
"author": "Test Author",
|
|
},
|
|
)
|
|
|
|
assert result.data.success is True
|
|
assert result.data.module_name == "custom"
|
|
assert result.data.block_name == "my_gain"
|
|
|
|
# Check block files exist
|
|
assert (tmp_path / "gr-custom" / "python" / "custom" / "my_gain.py").exists()
|
|
assert (tmp_path / "gr-custom" / "grc" / "custom_my_gain.block.yml").exists()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_export_full_workflow(self, mcp_app, tmp_path):
|
|
"""Full workflow: parse protocol → generate chain → export blocks."""
|
|
async with Client(mcp_app) as client:
|
|
await client.call_tool(name="enable_block_dev_mode")
|
|
|
|
# Generate a custom block for the protocol
|
|
block_result = await client.call_tool(
|
|
name="generate_sync_block",
|
|
arguments={
|
|
"name": "pm_demod",
|
|
"description": "Phase demodulator for Apollo USB",
|
|
"inputs": [{"dtype": "complex", "vlen": 1}],
|
|
"outputs": [{"dtype": "float", "vlen": 1}],
|
|
"parameters": [{"name": "sensitivity", "dtype": "float", "default": 1.0}],
|
|
"work_logic": "output_items[0][:] = numpy.angle(input_items[0]) * self.sensitivity",
|
|
},
|
|
)
|
|
|
|
assert block_result.data.is_valid is True
|
|
|
|
# Export to OOT module
|
|
output_dir = str(tmp_path / "gr-apollo")
|
|
|
|
result = await client.call_tool(
|
|
name="export_block_to_oot",
|
|
arguments={
|
|
"generated": block_result.structured_content,
|
|
"module_name": "apollo",
|
|
"output_dir": output_dir,
|
|
},
|
|
)
|
|
|
|
assert result.data.success is True
|
|
assert result.data.build_ready is True
|
|
|
|
# Verify the exported Python source contains our work logic
|
|
block_py = tmp_path / "gr-apollo" / "python" / "apollo" / "pm_demod.py"
|
|
assert block_py.exists()
|
|
content = block_py.read_text()
|
|
assert "numpy.angle" in content
|
|
assert "sensitivity" in content
|