Ryan Malloy
ce785029f4
Blocks generated by gr-mcp now have a first-class path to GRC's graphical editor. Previously, .block.yml was only produced as a side-effect of full OOT export — now clients can generate YAML standalone to preview, customize, or use without a full OOT module. - New MCP tool: generate_grc_yaml (registered in block dev mode) - New model: GrcYamlResult with yaml_content, notes, filename - Enhanced YAML generation: flags, asserts from min/max bounds, parameter descriptions, docstring extraction - New prompt resource: prompts://block-generation/grc-yaml - 10 new unit tests for GRC YAML generation (475 total, 7 skipped)
439 lines
16 KiB
Python
439 lines
16 KiB
Python
"""Unit tests for OOTExporterMiddleware."""
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from gnuradio_mcp.middlewares.oot_exporter import OOTExporterMiddleware
|
|
from gnuradio_mcp.models import BlockParameter, GeneratedBlockCode, SignatureItem
|
|
|
|
|
|
class TestOOTExporterMiddleware:
|
|
"""Tests for OOT module export functionality."""
|
|
|
|
@pytest.fixture
|
|
def exporter(self):
|
|
"""Create an OOT exporter instance."""
|
|
return OOTExporterMiddleware()
|
|
|
|
@pytest.fixture
|
|
def sample_block(self):
|
|
"""Create a sample generated block for testing."""
|
|
return GeneratedBlockCode(
|
|
source_code='''
|
|
import numpy
|
|
from gnuradio import gr
|
|
|
|
class blk(gr.sync_block):
|
|
def __init__(self, gain=1.0):
|
|
gr.sync_block.__init__(
|
|
self,
|
|
name="test_gain",
|
|
in_sig=[numpy.float32],
|
|
out_sig=[numpy.float32],
|
|
)
|
|
self.gain = gain
|
|
|
|
def work(self, input_items, output_items):
|
|
output_items[0][:] = input_items[0] * self.gain
|
|
return len(output_items[0])
|
|
''',
|
|
block_name="test_gain",
|
|
block_class="sync_block",
|
|
inputs=[SignatureItem(dtype="float", vlen=1)],
|
|
outputs=[SignatureItem(dtype="float", vlen=1)],
|
|
is_valid=True,
|
|
)
|
|
|
|
def test_generate_oot_skeleton_creates_directory(self, exporter):
|
|
"""Generate OOT skeleton creates proper directory structure."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# Create skeleton - output_dir is the module root itself
|
|
module_dir = Path(tmpdir) / "gr-custom"
|
|
result = exporter.generate_oot_skeleton(
|
|
module_name="custom",
|
|
output_dir=str(module_dir),
|
|
author="Test Author",
|
|
)
|
|
|
|
# Check result
|
|
assert result.success is True
|
|
assert result.module_name == "custom"
|
|
|
|
# Check directory exists
|
|
assert module_dir.exists()
|
|
|
|
# Check required files
|
|
assert (module_dir / "CMakeLists.txt").exists()
|
|
assert (module_dir / "python" / "custom" / "__init__.py").exists()
|
|
assert (module_dir / "grc").exists()
|
|
|
|
def test_generate_oot_skeleton_creates_cmake(self, exporter):
|
|
"""Generate OOT skeleton creates valid CMakeLists.txt."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
module_dir = Path(tmpdir) / "gr-mymodule"
|
|
exporter.generate_oot_skeleton(
|
|
module_name="mymodule",
|
|
output_dir=str(module_dir),
|
|
)
|
|
|
|
cmake_path = module_dir / "CMakeLists.txt"
|
|
cmake_content = cmake_path.read_text()
|
|
|
|
assert "cmake_minimum_required" in cmake_content
|
|
assert "project(" in cmake_content
|
|
assert "find_package(Gnuradio" in cmake_content
|
|
|
|
def test_generate_oot_skeleton_creates_python_init(self, exporter):
|
|
"""Generate OOT skeleton creates Python __init__.py."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
module_dir = Path(tmpdir) / "gr-testmod"
|
|
exporter.generate_oot_skeleton(
|
|
module_name="testmod",
|
|
output_dir=str(module_dir),
|
|
)
|
|
|
|
init_path = module_dir / "python" / "testmod" / "__init__.py"
|
|
init_content = init_path.read_text()
|
|
|
|
# Should have basic module setup
|
|
assert "testmod" in init_content or "__init__" in str(init_path)
|
|
|
|
def test_export_block_to_oot(self, exporter, sample_block):
|
|
"""Export a generated block to OOT module structure."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
module_dir = Path(tmpdir) / "gr-custom"
|
|
result = exporter.export_block_to_oot(
|
|
generated=sample_block,
|
|
module_name="custom",
|
|
output_dir=str(module_dir),
|
|
)
|
|
|
|
assert result.success is True
|
|
|
|
# Check block file exists
|
|
block_path = module_dir / "python" / "custom" / "test_gain.py"
|
|
assert block_path.exists()
|
|
|
|
# Check GRC yaml exists
|
|
yaml_files = list((module_dir / "grc").glob("*.block.yml"))
|
|
assert len(yaml_files) > 0
|
|
|
|
def test_export_block_creates_yaml(self, exporter, sample_block):
|
|
"""Export creates valid GRC YAML file."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
module_dir = Path(tmpdir) / "gr-custom"
|
|
exporter.export_block_to_oot(
|
|
generated=sample_block,
|
|
module_name="custom",
|
|
output_dir=str(module_dir),
|
|
)
|
|
|
|
yaml_path = module_dir / "grc" / "custom_test_gain.block.yml"
|
|
if yaml_path.exists():
|
|
yaml_content = yaml_path.read_text()
|
|
|
|
assert "id:" in yaml_content
|
|
assert "label:" in yaml_content
|
|
assert "templates:" in yaml_content
|
|
|
|
def test_export_to_existing_module(self, exporter, sample_block):
|
|
"""Export to an existing OOT module adds the block."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
module_dir = Path(tmpdir) / "gr-existing"
|
|
|
|
# First create the skeleton
|
|
exporter.generate_oot_skeleton(
|
|
module_name="existing",
|
|
output_dir=str(module_dir),
|
|
)
|
|
|
|
# Then export a block to it
|
|
result = exporter.export_block_to_oot(
|
|
generated=sample_block,
|
|
module_name="existing",
|
|
output_dir=str(module_dir),
|
|
)
|
|
|
|
assert result.success is True
|
|
|
|
# Block should be added
|
|
block_path = module_dir / "python" / "existing" / "test_gain.py"
|
|
assert block_path.exists()
|
|
|
|
|
|
class TestOOTExporterEdgeCases:
|
|
"""Tests for edge cases in OOT export."""
|
|
|
|
@pytest.fixture
|
|
def exporter(self):
|
|
return OOTExporterMiddleware()
|
|
|
|
def test_sanitize_module_name_strips_gr_prefix(self, exporter):
|
|
"""Module names have gr- prefix stripped."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
module_dir = Path(tmpdir) / "gr-mytest"
|
|
result = exporter.generate_oot_skeleton(
|
|
module_name="gr-mytest", # Has gr- prefix which should be removed
|
|
output_dir=str(module_dir),
|
|
)
|
|
|
|
assert result.success is True
|
|
# The sanitized module_name should not have gr- prefix
|
|
assert result.module_name == "mytest"
|
|
|
|
def test_sanitize_module_name_replaces_dashes(self, exporter):
|
|
"""Module names with dashes are sanitized to underscores."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
module_dir = Path(tmpdir) / "gr-my-module"
|
|
result = exporter.generate_oot_skeleton(
|
|
module_name="my-module-name",
|
|
output_dir=str(module_dir),
|
|
)
|
|
|
|
assert result.success is True
|
|
# Dashes replaced with underscores for valid Python identifier
|
|
assert "_" in result.module_name or result.module_name.isalnum()
|
|
|
|
def test_export_complex_block(self, exporter):
|
|
"""Export a block with complex I/O."""
|
|
complex_block = GeneratedBlockCode(
|
|
source_code='''
|
|
import numpy
|
|
from gnuradio import gr
|
|
|
|
class blk(gr.sync_block):
|
|
def __init__(self):
|
|
gr.sync_block.__init__(
|
|
self,
|
|
name="complex_processor",
|
|
in_sig=[numpy.complex64],
|
|
out_sig=[numpy.complex64],
|
|
)
|
|
|
|
def work(self, input_items, output_items):
|
|
output_items[0][:] = input_items[0] * 1j
|
|
return len(output_items[0])
|
|
''',
|
|
block_name="complex_processor",
|
|
block_class="sync_block",
|
|
inputs=[SignatureItem(dtype="complex", vlen=1)],
|
|
outputs=[SignatureItem(dtype="complex", vlen=1)],
|
|
is_valid=True,
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
module_dir = Path(tmpdir) / "gr-complex_test"
|
|
result = exporter.export_block_to_oot(
|
|
generated=complex_block,
|
|
module_name="complex_test",
|
|
output_dir=str(module_dir),
|
|
)
|
|
|
|
assert result.success is True
|
|
|
|
def test_export_block_with_parameters(self, exporter):
|
|
"""Export a block with multiple parameters."""
|
|
from gnuradio_mcp.models import BlockParameter
|
|
|
|
param_block = GeneratedBlockCode(
|
|
source_code='''
|
|
import numpy
|
|
from gnuradio import gr
|
|
|
|
class blk(gr.sync_block):
|
|
def __init__(self, gain=1.0, offset=0.0, threshold=0.5):
|
|
gr.sync_block.__init__(
|
|
self,
|
|
name="multi_param",
|
|
in_sig=[numpy.float32],
|
|
out_sig=[numpy.float32],
|
|
)
|
|
self.gain = gain
|
|
self.offset = offset
|
|
self.threshold = threshold
|
|
|
|
def work(self, input_items, output_items):
|
|
scaled = input_items[0] * self.gain + self.offset
|
|
output_items[0][:] = numpy.where(scaled > self.threshold, scaled, 0.0)
|
|
return len(output_items[0])
|
|
''',
|
|
block_name="multi_param",
|
|
block_class="sync_block",
|
|
inputs=[SignatureItem(dtype="float", vlen=1)],
|
|
outputs=[SignatureItem(dtype="float", vlen=1)],
|
|
parameters=[
|
|
BlockParameter(name="gain", dtype="float", default=1.0),
|
|
BlockParameter(name="offset", dtype="float", default=0.0),
|
|
BlockParameter(name="threshold", dtype="float", default=0.5),
|
|
],
|
|
is_valid=True,
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
module_dir = Path(tmpdir) / "gr-param_test"
|
|
result = exporter.export_block_to_oot(
|
|
generated=param_block,
|
|
module_name="param_test",
|
|
output_dir=str(module_dir),
|
|
)
|
|
|
|
assert result.success is True
|
|
|
|
|
|
class TestOOTExporterYAMLGeneration:
|
|
"""Tests specifically for YAML file generation."""
|
|
|
|
@pytest.fixture
|
|
def exporter(self):
|
|
return OOTExporterMiddleware()
|
|
|
|
def test_yaml_has_required_fields(self, exporter):
|
|
"""Generated YAML has all required GRC fields."""
|
|
block = GeneratedBlockCode(
|
|
source_code="...",
|
|
block_name="my_block",
|
|
block_class="sync_block",
|
|
inputs=[SignatureItem(dtype="float", vlen=1)],
|
|
outputs=[SignatureItem(dtype="float", vlen=1)],
|
|
is_valid=True,
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
module_dir = Path(tmpdir) / "gr-test"
|
|
exporter.export_block_to_oot(
|
|
generated=block,
|
|
module_name="test",
|
|
output_dir=str(module_dir),
|
|
)
|
|
|
|
yaml_path = module_dir / "grc" / "test_my_block.block.yml"
|
|
if yaml_path.exists():
|
|
yaml_content = yaml_path.read_text()
|
|
|
|
# Required GRC YAML fields
|
|
assert "id:" in yaml_content
|
|
assert "label:" in yaml_content
|
|
assert "category:" in yaml_content
|
|
assert "templates:" in yaml_content
|
|
|
|
|
|
class TestGenerateGrcYaml:
|
|
"""Tests for the standalone generate_grc_yaml method."""
|
|
|
|
@pytest.fixture
|
|
def exporter(self):
|
|
return OOTExporterMiddleware()
|
|
|
|
@pytest.fixture
|
|
def gain_block(self):
|
|
return GeneratedBlockCode(
|
|
source_code='''
|
|
import numpy
|
|
from gnuradio import gr
|
|
|
|
class blk(gr.sync_block):
|
|
"""Configurable gain block."""
|
|
def __init__(self, gain=1.0):
|
|
gr.sync_block.__init__(
|
|
self, name="configurable_gain",
|
|
in_sig=[numpy.float32], out_sig=[numpy.float32],
|
|
)
|
|
self.gain = gain
|
|
|
|
def set_gain(self, gain):
|
|
self.gain = gain
|
|
|
|
def work(self, input_items, output_items):
|
|
output_items[0][:] = input_items[0] * self.gain
|
|
return len(output_items[0])
|
|
''',
|
|
block_name="configurable_gain",
|
|
block_class="sync_block",
|
|
inputs=[SignatureItem(dtype="float", vlen=1)],
|
|
outputs=[SignatureItem(dtype="float", vlen=1)],
|
|
parameters=[
|
|
BlockParameter(name="gain", dtype="float", default=1.0,
|
|
description="Gain factor", min_value=0.0),
|
|
],
|
|
is_valid=True,
|
|
)
|
|
|
|
def test_returns_success(self, exporter, gain_block):
|
|
"""generate_grc_yaml returns a successful result."""
|
|
result = exporter.generate_grc_yaml(gain_block)
|
|
assert result.success is True
|
|
assert result.yaml_content
|
|
assert result.filename == "custom_configurable_gain.block.yml"
|
|
assert result.block_id == "custom_configurable_gain"
|
|
|
|
def test_yaml_has_flags(self, exporter, gain_block):
|
|
"""Generated YAML includes flags: [ python ]."""
|
|
result = exporter.generate_grc_yaml(gain_block)
|
|
assert "flags: [ python ]" in result.yaml_content
|
|
|
|
def test_yaml_has_callbacks(self, exporter, gain_block):
|
|
"""Setter methods are detected as GRC callbacks."""
|
|
result = exporter.generate_grc_yaml(gain_block)
|
|
assert "set_gain(${gain})" in result.yaml_content
|
|
assert "callbacks:" in result.yaml_content
|
|
|
|
def test_yaml_has_asserts_from_min_value(self, exporter, gain_block):
|
|
"""Parameter min_value generates GRC asserts."""
|
|
result = exporter.generate_grc_yaml(gain_block)
|
|
assert "gain >= 0.0" in result.yaml_content
|
|
|
|
def test_yaml_has_parameter_dtype_mapping(self, exporter, gain_block):
|
|
"""Python 'float' maps to GRC 'real' dtype."""
|
|
result = exporter.generate_grc_yaml(gain_block)
|
|
assert "dtype: real" in result.yaml_content
|
|
|
|
def test_yaml_has_documentation(self, exporter, gain_block):
|
|
"""Docstring is included as documentation."""
|
|
result = exporter.generate_grc_yaml(gain_block)
|
|
assert "Configurable gain block" in result.yaml_content
|
|
|
|
def test_custom_module_name(self, exporter, gain_block):
|
|
"""Module name changes block ID and import."""
|
|
result = exporter.generate_grc_yaml(gain_block, module_name="dsp")
|
|
assert result.block_id == "dsp_configurable_gain"
|
|
assert "from dsp import configurable_gain" in result.yaml_content
|
|
|
|
def test_custom_category(self, exporter, gain_block):
|
|
"""Category can be overridden."""
|
|
result = exporter.generate_grc_yaml(gain_block, category="Signal Processing")
|
|
assert "category: [Signal Processing]" in result.yaml_content
|
|
|
|
def test_notes_for_no_setters(self, exporter):
|
|
"""Notes warn about missing setter methods."""
|
|
block = GeneratedBlockCode(
|
|
source_code='''
|
|
import numpy
|
|
from gnuradio import gr
|
|
|
|
class blk(gr.sync_block):
|
|
def __init__(self, gain=1.0):
|
|
gr.sync_block.__init__(self, name="no_setters",
|
|
in_sig=[numpy.float32], out_sig=[numpy.float32])
|
|
self.gain = gain
|
|
|
|
def work(self, input_items, output_items):
|
|
output_items[0][:] = input_items[0] * self.gain
|
|
return len(output_items[0])
|
|
''',
|
|
block_name="no_setters",
|
|
block_class="sync_block",
|
|
inputs=[SignatureItem(dtype="float", vlen=1)],
|
|
outputs=[SignatureItem(dtype="float", vlen=1)],
|
|
parameters=[BlockParameter(name="gain", dtype="float", default=1.0)],
|
|
is_valid=True,
|
|
)
|
|
result = exporter.generate_grc_yaml(block)
|
|
assert any("set_*()" in note for note in result.notes)
|
|
|
|
def test_file_format_1(self, exporter, gain_block):
|
|
"""YAML includes file_format: 1."""
|
|
result = exporter.generate_grc_yaml(gain_block)
|
|
assert "file_format: 1" in result.yaml_content
|