Skip to content
GR-MCP

Code Coverage

GR-MCP can collect Python code coverage from flowgraphs running in Docker containers. This is useful for measuring test coverage of GNU Radio applications and embedded Python blocks.

How It Works

  1. Launch with enable_coverage=True — container runs flowgraph under coverage.py
  2. Run your test scenarios via XML-RPC variable control
  3. Stop the flowgraph gracefully — coverage data is written to disk
  4. Collect and analyze — generate reports in HTML, XML, or JSON
┌─────────────────────────────────────────────────────────────┐
│  Docker Container                                           │
│  ┌─────────────────────────────────────────────────────────┐│
│  │  coverage run flowgraph.py                              ││
│  │    └─ writes .coverage.* files on exit                  ││
│  └─────────────────────────────────────────────────────────┘│
│            │                                                │
│            ▼                                                │
│  /tmp/gr-mcp-coverage/{container_name}/                     │
│    ├─ .coverage                                             │
│    ├─ .coverage.{hostname}.{pid}.*                          │
│    └─ htmlcov/                                              │
└─────────────────────────────────────────────────────────────┘

Prerequisites

Build the coverage-enabled Docker image:

Terminal window
docker build -f docker/Dockerfile.gnuradio-coverage \
  -t gnuradio-coverage:latest docker/

Collect Coverage

  1. Launch with coverage enabled

    enable_runtime_mode()
    

    launch_flowgraph(
        flowgraph_path="/tmp/flowgraph.py",
        name="coverage-test",
        enable_coverage=True
    )

  2. Run your test scenario

    connect_to_container(name="coverage-test")
    

    # Exercise the flowgraph
    set_variable(name="freq", value=100e6)
    time.sleep(2)
    set_variable(name="freq", value=200e6)
    time.sleep(2)
    # ... more test actions ...

  3. Stop gracefully (required for coverage data)

    stop_flowgraph(name="coverage-test")

  4. Collect the coverage data

    collect_coverage(name="coverage-test")
    # Returns CoverageDataModel:
    #   container_name: "coverage-test"
    #   coverage_file: "/tmp/gr-mcp-coverage/coverage-test/.coverage"
    #   summary: "Name    Stmts   Miss  Cover\n..."
    #   lines_covered: 150
    #   lines_total: 200
    #   coverage_percent: 75.0

  5. Generate a report

    generate_coverage_report(
        name="coverage-test",
        format="html"
    )
    # Returns CoverageReportModel:
    #   report_path: "/tmp/gr-mcp-coverage/coverage-test/htmlcov/index.html"

Report Formats

/tmp/gr-mcp-coverage/coverage-test/htmlcov/index.html
generate_coverage_report(name="coverage-test", format="html")

Best for visual inspection — shows line-by-line coverage with highlighting.

Combine Multiple Runs

Aggregate coverage from multiple test scenarios:

# Run first scenario
launch_flowgraph(..., name="test-1", enable_coverage=True)
# ... exercise ...
stop_flowgraph(name="test-1")


# Run second scenario
launch_flowgraph(..., name="test-2", enable_coverage=True)
# ... exercise ...
stop_flowgraph(name="test-2")


# Combine results
combined = combine_coverage(names=["test-1", "test-2"])
# Returns CoverageDataModel for combined data


# Generate combined report
generate_coverage_report(name="combined", format="html")

Clean Up Coverage Data

# Delete specific container's coverage
delete_coverage(name="coverage-test")
# Returns: 1 (directories deleted)


# Delete old coverage data
delete_coverage(older_than_days=7)
# Returns: number of directories deleted


# Delete all coverage data
delete_coverage()
# Returns: number of directories deleted

Example: Test Suite with Coverage

#!/usr/bin/env python3
"""Run a test suite with coverage collection."""


import asyncio
import time
from fastmcp import Client


TEST_SCENARIOS = [
    {"name": "low-freq", "freq": 50e6, "duration": 5},
    {"name": "mid-freq", "freq": 500e6, "duration": 5},
    {"name": "high-freq", "freq": 2.4e9, "duration": 5},
]


async def run_test_suite():
    async with Client("gr-mcp") as client:
        await client.call_tool("enable_runtime_mode", {})


        container_names = []


        for scenario in TEST_SCENARIOS:
            name = f"cov-{scenario['name']}"
            container_names.append(name)


            print(f"Running scenario: {scenario['name']}")


            # Launch with coverage
            await client.call_tool("launch_flowgraph", {
                "flowgraph_path": "/tmp/radio_app.py",
                "name": name,
                "enable_coverage": True
            })


            time.sleep(3)


            # Connect and run scenario
            await client.call_tool("connect_to_container", {"name": name})
            await client.call_tool("set_variable", {
                "name": "freq",
                "value": scenario["freq"]
            })


            time.sleep(scenario["duration"])


            # Stop gracefully for coverage
            await client.call_tool("stop_flowgraph", {"name": name})


            # Collect this run's coverage
            result = await client.call_tool("collect_coverage", {"name": name})
            print(f"  Coverage: {result.data.coverage_percent}%")


        # Combine all runs
        print("\nCombining coverage from all scenarios...")
        combined = await client.call_tool("combine_coverage", {
            "names": container_names
        })
        print(f"Combined coverage: {combined.data.coverage_percent}%")


        # Generate final report
        await client.call_tool("generate_coverage_report", {
            "name": "combined",
            "format": "html"
        })
        print("HTML report: /tmp/gr-mcp-coverage/combined/htmlcov/index.html")


        # Cleanup containers
        for name in container_names:
            await client.call_tool("remove_flowgraph", {"name": name})


if __name__ == "__main__":
    asyncio.run(run_test_suite())

Coverage for Embedded Python Blocks

Coverage includes any embedded Python blocks in your flowgraph:

# Create an embedded block
source = """
import numpy as np
from gnuradio import gr


class my_block(gr.sync_block):
    def __init__(self):
        gr.sync_block.__init__(self, ...)


    def work(self, input_items, output_items):
        # This code path will show in coverage
        if input_items[0][0] > 0.5:
            output_items[0][:] = input_items[0] * 2
        else:
            output_items[0][:] = input_items[0]
        return len(output_items[0])
"""


create_embedded_python_block(source_code=source)

Coverage reports will show which branches of your embedded block were exercised.

Next Steps