MCP/gr-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: MCP:e63f6e1ba060c16d33b6dee458050746269fe016
Branches Tags
MCP:main
..
compare: MCP:521c30617350aa2f5878be197260780959097f86
Branches Tags
MCP:main

No commits in common. "e63f6e1ba060c16d33b6dee458050746269fe016" and "521c30617350aa2f5878be197260780959097f86" have entirely different histories.
e63f6e1ba0 ... 521c306173

7 changed files with 80 additions and 1726 deletions
Show Stats Expand all files Collapse all files

502
src/gnuradio_mcp/middlewares/oot.py
View File

@ -4,20 +4,11 @@ import io
import json import json
import logging import logging
import subprocess import subprocess
import tarfile
from datetime import datetime, timezone from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
import re from gnuradio_mcp.models import OOTImageInfo, OOTInstallResult
from gnuradio_mcp.models import (
ComboImageInfo,
ComboImageResult,
OOTDetectionResult,
OOTImageInfo,
OOTInstallResult,
)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -34,11 +25,8 @@ RUN apt-get update && apt-get install -y --no-install-recommends \\
# Clone and build # Clone and build
WORKDIR /build WORKDIR /build
COPY fix_binding_hashes.py /tmp/fix_binding_hashes.py
RUN git clone --depth 1 --branch {branch} {git_url} && \\ RUN git clone --depth 1 --branch {branch} {git_url} && \\
cd {repo_dir} && \\ cd {repo_dir} && mkdir build && cd build && \\
python3 /tmp/fix_binding_hashes.py . && \\
mkdir build && cd build && \\
cmake -DCMAKE_INSTALL_PREFIX=/usr {cmake_args}.. && \\ cmake -DCMAKE_INSTALL_PREFIX=/usr {cmake_args}.. && \\
make -j$(nproc) && make install && \\ make -j$(nproc) && make install && \\
ldconfig && \\ ldconfig && \\
@ -50,47 +38,6 @@ WORKDIR /flowgraphs
ENV PYTHONPATH="/usr/lib/python3.11/site-packages:${{PYTHONPATH}}" ENV PYTHONPATH="/usr/lib/python3.11/site-packages:${{PYTHONPATH}}"
""" """
# Standalone script injected into OOT Docker builds to fix stale
# pybind11 binding hashes that would otherwise trigger castxml regen.
FIX_BINDING_HASHES_SCRIPT = """\
#!/usr/bin/env python3
\"\"\"Fix stale BINDTOOL_HEADER_FILE_HASH in pybind11 binding files.
GNU Radio's GR_PYBIND_MAKE_OOT cmake macro compares MD5 hashes of C++
headers against values stored in the binding .cc files. When they
differ it tries to regenerate via castxml, which often fails in minimal
Docker images. This script updates the hashes to match the actual
headers so cmake skips the regeneration step.
\"\"\"
import hashlib, pathlib, re, sys
root = pathlib.Path(sys.argv[1]) if len(sys.argv) > 1 else pathlib.Path(".")
# GR 3.9-: python/bindings/ | GR 3.10+: python/<module>/bindings/
binding_dirs = list(root.joinpath("python").glob("**/bindings"))
if not binding_dirs:
sys.exit(0)
for bindings in binding_dirs:
for cc in sorted(bindings.glob("*_python.cc")):
text = cc.read_text()
m = re.search(r"BINDTOOL_HEADER_FILE\\((\\S+)\\)", text)
if not m:
continue
header = next(root.joinpath("include").rglob(m.group(1)), None)
if not header:
continue
actual = hashlib.md5(header.read_bytes()).hexdigest()
new_text = re.sub(
r"BINDTOOL_HEADER_FILE_HASH\\([a-f0-9]+\\)",
f"BINDTOOL_HEADER_FILE_HASH({actual})",
text,
)
if new_text != text:
cc.write_text(new_text)
print(f"Fixed binding hash: {cc.name}")
"""
class OOTInstallerMiddleware: class OOTInstallerMiddleware:
"""Builds OOT modules into Docker images from git repos. """Builds OOT modules into Docker images from git repos.
@ -109,8 +56,6 @@ class OOTInstallerMiddleware:
self._base_image = base_image self._base_image = base_image
self._registry_path = Path.home() / ".gr-mcp" / "oot-registry.json" self._registry_path = Path.home() / ".gr-mcp" / "oot-registry.json"
self._registry: dict[str, OOTImageInfo] = self._load_registry() self._registry: dict[str, OOTImageInfo] = self._load_registry()
self._combo_registry_path = Path.home() / ".gr-mcp" / "oot-combo-registry.json"
self._combo_registry: dict[str, ComboImageInfo] = self._load_combo_registry()
# ────────────────────────────────────────── # ──────────────────────────────────────────
# Public API # Public API
@ -136,23 +81,6 @@ class OOTInstallerMiddleware:
# Idempotent: skip if image already exists # Idempotent: skip if image already exists
if not force and self._image_exists(image_tag): if not force and self._image_exists(image_tag):
existing = self._registry.get(module_name) existing = self._registry.get(module_name)
if existing is None:
# Image exists but registry entry missing — re-register
existing = OOTImageInfo(
module_name=module_name,
image_tag=image_tag,
git_url=git_url,
branch=branch,
git_commit=commit,
base_image=effective_base,
built_at=datetime.now(timezone.utc).isoformat(),
)
self._registry[module_name] = existing
self._save_registry()
logger.info(
"Re-registered existing image '%s' for module '%s'",
image_tag, module_name,
)
return OOTInstallResult( return OOTInstallResult(
success=True, success=True,
image=existing, image=existing,
@ -307,30 +235,13 @@ class OOTInstallerMiddleware:
except Exception: except Exception:
return False return False
@staticmethod
def _build_context(dockerfile: str) -> io.BytesIO:
"""Create a tar archive build context with Dockerfile and helper scripts."""
buf = io.BytesIO()
with tarfile.open(fileobj=buf, mode="w") as tar:
for name, content in [
("Dockerfile", dockerfile),
("fix_binding_hashes.py", FIX_BINDING_HASHES_SCRIPT),
]:
data = content.encode("utf-8")
info = tarfile.TarInfo(name=name)
info.size = len(data)
tar.addfile(info, io.BytesIO(data))
buf.seek(0)
return buf
def _docker_build(self, dockerfile: str, tag: str) -> list[str]: def _docker_build(self, dockerfile: str, tag: str) -> list[str]:
"""Build a Docker image from a Dockerfile string. Returns log lines.""" """Build a Docker image from a Dockerfile string. Returns log lines."""
context = self._build_context(dockerfile) f = io.BytesIO(dockerfile.encode("utf-8"))
log_lines: list[str] = [] log_lines: list[str] = []
try: try:
_image, build_log = self._client.images.build( _image, build_log = self._client.images.build(
fileobj=context, fileobj=f,
custom_context=True,
tag=tag, tag=tag,
rm=True, rm=True,
forcerm=True, forcerm=True,
@ -346,25 +257,18 @@ class OOTInstallerMiddleware:
return log_lines return log_lines
def _load_registry(self) -> dict[str, OOTImageInfo]: def _load_registry(self) -> dict[str, OOTImageInfo]:
"""Load the OOT image registry from disk. """Load the OOT image registry from disk."""
Validates entries individually so one corrupted entry
doesn't discard the entire registry.
"""
if not self._registry_path.exists(): if not self._registry_path.exists():
return {} return {}
try: try:
data = json.loads(self._registry_path.read_text()) data = json.loads(self._registry_path.read_text())
return {
k: OOTImageInfo(**v)
for k, v in data.items()
}
except Exception as e: except Exception as e:
logger.warning("Failed to parse OOT registry JSON: %s", e) logger.warning("Failed to load OOT registry: %s", e)
return {} return {}
registry: dict[str, OOTImageInfo] = {}
for k, v in data.items():
try:
registry[k] = OOTImageInfo(**v)
except Exception as e:
logger.warning("Skipping corrupt registry entry '%s': %s", k, e)
return registry
def _save_registry(self) -> None: def _save_registry(self) -> None:
"""Persist the OOT image registry to disk.""" """Persist the OOT image registry to disk."""
@ -374,389 +278,3 @@ class OOTInstallerMiddleware:
for k, v in self._registry.items() for k, v in self._registry.items()
} }
self._registry_path.write_text(json.dumps(data, indent=2)) self._registry_path.write_text(json.dumps(data, indent=2))
# ──────────────────────────────────────────
# Combo Image (Multi-OOT) Support
# ──────────────────────────────────────────
@staticmethod
def _combo_key(module_names: list[str]) -> str:
"""Deterministic key from module names: sorted, deduped, joined."""
names = sorted(set(module_names))
return "combo:" + "+".join(names)
@staticmethod
def _combo_image_tag(module_names: list[str]) -> str:
"""Deterministic image tag for a combo of modules."""
names = sorted(set(module_names))
return f"gr-combo-{'-'.join(names)}:latest"
def generate_combo_dockerfile(self, module_names: list[str]) -> str:
"""Generate multi-stage Dockerfile that COPYs from existing single-OOT images."""
names = sorted(set(module_names))
stages: list[str] = []
copies: list[str] = []
for name in names:
info = self._registry.get(name)
if info is None:
raise ValueError(
f"Module '{name}' not found in OOT registry. "
f"Build it first with build_module()."
)
stage_alias = f"stage_{name}"
stages.append(f"FROM {info.image_tag} AS {stage_alias}")
for path in ["/usr/lib/", "/usr/include/", "/usr/share/gnuradio/"]:
copies.append(f"COPY --from={stage_alias} {path} {path}")
return "\n".join([
*stages,
"",
f"FROM {self._base_image}",
"",
*copies,
"",
"RUN ldconfig",
"WORKDIR /flowgraphs",
'ENV PYTHONPATH="/usr/lib/python3.11/site-packages:${PYTHONPATH}"',
"",
])
def build_combo_image(
self,
module_names: list[str],
force: bool = False,
) -> ComboImageResult:
"""Build a combined Docker image with multiple OOT modules.
Modules already in the registry are used as-is. Modules found
in the OOT catalog but not yet built are auto-built first.
"""
from gnuradio_mcp.oot_catalog import CATALOG
names = sorted(set(module_names))
if len(names) < 2:
return ComboImageResult(
success=False,
error="At least 2 distinct modules required for a combo image.",
)
combo_key = self._combo_key(names)
image_tag = self._combo_image_tag(names)
try:
# Idempotent: skip if combo already exists
if not force and self._image_exists(image_tag):
existing = self._combo_registry.get(combo_key)
if existing is not None:
return ComboImageResult(
success=True,
image=existing,
skipped=True,
)
# Auto-build missing modules from catalog
modules_built: list[str] = []
for name in names:
if name in self._registry:
continue
entry = CATALOG.get(name)
if entry is None:
return ComboImageResult(
success=False,
error=(
f"Module '{name}' is not in the OOT registry and "
f"not found in the catalog. Build it manually first "
f"with build_module()."
),
)
# Auto-build from catalog
logger.info("Auto-building '%s' from catalog for combo image", name)
result = self.build_module(
git_url=entry.git_url,
branch=entry.branch,
build_deps=entry.build_deps or None,
cmake_args=entry.cmake_args or None,
)
if not result.success:
return ComboImageResult(
success=False,
error=f"Auto-build of '{name}' failed: {result.error}",
modules_built=modules_built,
)
modules_built.append(name)
# Generate and build combo
dockerfile = self.generate_combo_dockerfile(names)
log_lines = self._docker_build(dockerfile, image_tag)
build_log_tail = "\n".join(log_lines[-30:])
# Collect module infos for the combo record
module_infos = [self._registry[n] for n in names]
info = ComboImageInfo(
combo_key=combo_key,
image_tag=image_tag,
modules=module_infos,
built_at=datetime.now(timezone.utc).isoformat(),
)
self._combo_registry[combo_key] = info
self._save_combo_registry()
return ComboImageResult(
success=True,
image=info,
build_log_tail=build_log_tail,
modules_built=modules_built,
)
except Exception as e:
logger.exception("Combo image build failed")
return ComboImageResult(
success=False,
error=str(e),
)
def list_combo_images(self) -> list[ComboImageInfo]:
"""List all combined multi-OOT images."""
return list(self._combo_registry.values())
def remove_combo_image(self, combo_key: str) -> bool:
"""Remove a combo image by its key (e.g., 'combo:adsb+lora_sdr')."""
info = self._combo_registry.pop(combo_key, None)
if info is None:
return False
try:
self._client.images.remove(info.image_tag, force=True)
except Exception as e:
logger.warning(
"Failed to remove combo Docker image %s: %s", info.image_tag, e
)
self._save_combo_registry()
return True
# ──────────────────────────────────────────
# Combo Registry Persistence
# ──────────────────────────────────────────
def _load_combo_registry(self) -> dict[str, ComboImageInfo]:
"""Load the combo image registry from disk."""
if not self._combo_registry_path.exists():
return {}
try:
data = json.loads(self._combo_registry_path.read_text())
return {k: ComboImageInfo(**v) for k, v in data.items()}
except Exception as e:
logger.warning("Failed to load combo registry: %s", e)
return {}
def _save_combo_registry(self) -> None:
"""Persist the combo image registry to disk."""
self._combo_registry_path.parent.mkdir(parents=True, exist_ok=True)
data = {k: v.model_dump() for k, v in self._combo_registry.items()}
self._combo_registry_path.write_text(json.dumps(data, indent=2))
# ──────────────────────────────────────────
# OOT Module Detection
# ──────────────────────────────────────────
# Core GNU Radio module prefixes to ignore during detection
_CORE_BLOCK_PREFIXES = frozenset([
"blocks_", "analog_", "digital_", "filter_", "qtgui_", "fft_",
"audio_", "channels_", "trellis_", "vocoder_", "video_sdl_",
"dtv_", "fec_", "network_", "pdu_", "soapy_", "uhd_", "zeromq_",
"variable_", # All variable_* blocks (qtgui controls, function probe, etc.)
])
_CORE_BLOCK_EXACT = frozenset([
# Special blocks without prefixes
"variable", # The basic variable block (no underscore)
"import", "snippet", "options", "parameter",
"pad_source", "pad_sink", "virtual_source", "virtual_sink",
"note", "epy_block", "epy_module",
# Core filter blocks (not prefixed with filter_)
"low_pass_filter", "high_pass_filter", "band_pass_filter",
"band_reject_filter", "root_raised_cosine_filter",
# XML-RPC (part of core GR)
"xmlrpc_server", "xmlrpc_client",
])
# Explicit block ID → module mappings for blocks that don't follow
# the standard `module_prefix_` naming convention
_BLOCK_TO_MODULE: dict[str, str] = {
# gr-lora_sdr: main blocks use short names
"lora_rx": "lora_sdr",
"lora_tx": "lora_sdr",
# gr-adsb: decoder blocks
"adsb_decoder": "adsb",
"adsb_framer": "adsb",
# gr-iridium: main blocks
"iridium_extractor": "iridium",
"iridium_frame_sorter": "iridium",
"iridium_qpsk_demod": "iridium",
# gr-rds: decoder/encoder
"rds_decoder": "rds",
"rds_encoder": "rds",
"rds_parser": "rds",
"rds_panel": "rds",
# gr-satellites: common blocks
"satellites_ax25_deframer": "satellites",
"satellites_hdlc_deframer": "satellites",
"satellites_nrzi_decode": "satellites",
# gr-gsm: receiver blocks
"gsm_receiver": "gsm",
"gsm_input": "gsm",
"gsm_clock_offset_control": "gsm",
"gsm_controlled_rotator_cc": "gsm",
}
def detect_required_modules(self, flowgraph_path: str) -> OOTDetectionResult:
"""Detect OOT modules required by a flowgraph.
For .py files: parse Python imports
For .grc files: heuristic prefix matching against catalog
Args:
flowgraph_path: Path to a .py or .grc flowgraph file
Returns:
OOTDetectionResult with detected modules and recommended image
"""
path = Path(flowgraph_path)
if not path.exists():
raise FileNotFoundError(f"Flowgraph not found: {flowgraph_path}")
if path.suffix == ".py":
return self._detect_from_python(path)
elif path.suffix == ".grc":
return self._detect_from_grc(path)
else:
raise ValueError(
f"Unsupported file type: {path.suffix}. "
f"Expected .py or .grc"
)
def _detect_from_python(self, path: Path) -> OOTDetectionResult:
"""Parse Python imports to find OOT modules."""
from gnuradio_mcp.oot_catalog import CATALOG
content = path.read_text()
modules: set[str] = set()
# Pattern: import gnuradio.MODULE or from gnuradio import MODULE
for match in re.finditer(
r"(?:import gnuradio\.(\w+)|from gnuradio import (\w+))", content
):
module = match.group(1) or match.group(2)
if module in CATALOG:
modules.add(module)
# Pattern: import MODULE (top-level OOT like osmosdr)
for match in re.finditer(r"^import (\w+)\s*$", content, re.MULTILINE):
module = match.group(1)
if module in CATALOG:
modules.add(module)
# Pattern: from MODULE import ... (top-level OOT)
for match in re.finditer(r"^from (\w+) import", content, re.MULTILINE):
module = match.group(1)
if module in CATALOG:
modules.add(module)
sorted_modules = sorted(modules)
return OOTDetectionResult(
flowgraph_path=str(path),
detected_modules=sorted_modules,
detection_method="python_imports",
recommended_image=self._recommend_image(sorted_modules),
)
def _detect_from_grc(self, path: Path) -> OOTDetectionResult:
"""Heuristic: match block IDs against catalog module prefixes."""
import yaml
from gnuradio_mcp.oot_catalog import CATALOG
data = yaml.safe_load(path.read_text())
blocks = data.get("blocks", [])
block_ids = [b.get("id", "") for b in blocks if isinstance(b, dict)]
modules: set[str] = set()
unknown: list[str] = []
for block_id in block_ids:
# Skip empty IDs
if not block_id:
continue
# Skip core GNU Radio blocks
if block_id in self._CORE_BLOCK_EXACT:
continue
if any(block_id.startswith(prefix) for prefix in self._CORE_BLOCK_PREFIXES):
continue
# Phase 1: Check explicit block-to-module mapping (handles edge cases)
if block_id in self._BLOCK_TO_MODULE:
module = self._BLOCK_TO_MODULE[block_id]
if module in CATALOG:
modules.add(module)
continue
# Phase 2: Match against catalog module names as prefixes
matched = False
for module_name in CATALOG:
# Check for prefix match (e.g., "lora_sdr_gray_demap" -> "lora_sdr")
# Also handle cases like "osmosdr_source" -> "osmosdr"
if block_id.startswith(f"{module_name}_") or block_id == module_name:
modules.add(module_name)
matched = True
break
if not matched and "_" in block_id:
# Looks like an OOT block but not in catalog
# Don't flag all unknown blocks, just those with OOT-like patterns
prefix = block_id.split("_")[0]
# Check it's not a known core prefix that might have been missed
if not any(p.startswith(prefix) for p in self._CORE_BLOCK_PREFIXES):
unknown.append(block_id)
sorted_modules = sorted(modules)
return OOTDetectionResult(
flowgraph_path=str(path),
detected_modules=sorted_modules,
unknown_blocks=unknown,
detection_method="grc_prefix_heuristic",
recommended_image=self._recommend_image(sorted_modules),
)
def _recommend_image(self, modules: list[str]) -> str | None:
"""Recommend Docker image for detected modules.
Args:
modules: Sorted list of module names
Returns:
- Base runtime image if no OOT modules
- Single OOT image tag if one module (and built)
- Combo image tag if multiple modules
"""
if not modules:
return self._base_image
if len(modules) == 1:
# Single module - check if already built
info = self._registry.get(modules[0])
if info:
return info.image_tag
# Not built yet - return what the tag would be
return None
# Multiple modules - return combo tag (may or may not exist yet)
combo_key = self._combo_key(modules)
combo = self._combo_registry.get(combo_key)
if combo:
return combo.image_tag
# Return what the tag would be
return self._combo_image_tag(modules)

34
src/gnuradio_mcp/models.py
View File

@ -391,37 +391,3 @@ class OOTInstallResult(BaseModel):
build_log_tail: str = "" # Last ~30 lines of build output build_log_tail: str = "" # Last ~30 lines of build output
error: str | None = None error: str | None = None
skipped: bool = False # True if image already existed skipped: bool = False # True if image already existed
class ComboImageInfo(BaseModel):
"""Metadata for a combined multi-OOT image."""
combo_key: str # "combo:adsb+lora_sdr" (sorted, deduped)
image_tag: str # "gr-combo-adsb-lora_sdr:latest"
modules: list[OOTImageInfo] # source images used
built_at: str # ISO-8601
class ComboImageResult(BaseModel):
"""Result of combo image build."""
success: bool
image: ComboImageInfo | None = None
build_log_tail: str = ""
error: str | None = None
skipped: bool = False # True if combo image already existed
modules_built: list[str] = [] # modules auto-built from catalog first
class OOTDetectionResult(BaseModel):
"""Result of OOT module detection from a flowgraph.
Analyzes .py or .grc files to identify which OOT modules are required,
enabling automatic Docker image selection for launch_flowgraph().
"""
flowgraph_path: str
detected_modules: list[str] # OOT modules found (catalog matches)
unknown_blocks: list[str] = [] # Blocks that look OOT but aren't in catalog
detection_method: str # "python_imports" | "grc_prefix_heuristic"
recommended_image: str | None = None # Image tag to use (if modules found)

68
src/gnuradio_mcp/oot_catalog.py
View File

@ -139,7 +139,7 @@ CATALOG: dict[str, OOTModuleEntry] = {
description="FM RDS/RBDS (Radio Data System) decoder", description="FM RDS/RBDS (Radio Data System) decoder",
category="Broadcast", category="Broadcast",
git_url="https://github.com/bastibl/gr-rds", git_url="https://github.com/bastibl/gr-rds",
branch="maint-3.10", branch="main",
homepage="https://github.com/bastibl/gr-rds", homepage="https://github.com/bastibl/gr-rds",
preinstalled=True, preinstalled=True,
), ),
@ -217,33 +217,6 @@ CATALOG: dict[str, OOTModuleEntry] = {
preinstalled=True, preinstalled=True,
), ),
# ── Installable via install_oot_module ── # ── Installable via install_oot_module ──
_entry(
name="foo",
description="Wireshark PCAP connector, burst tagger, periodic msg source",
category="Utility",
git_url="https://github.com/bastibl/gr-foo",
branch="maint-3.10",
build_deps=["castxml"],
homepage="https://github.com/bastibl/gr-foo",
),
_entry(
name="owc",
description="Optical Wireless Communication channel simulation and modulation",
category="Optical",
git_url="https://github.com/UCaNLabUMB/gr-owc",
branch="main",
homepage="https://github.com/UCaNLabUMB/gr-owc",
),
_entry(
name="dab",
description="DAB/DAB+ digital audio broadcast receiver",
category="Broadcast",
git_url="https://github.com/hboeglen/gr-dab",
branch="maint-3.10",
build_deps=["autoconf", "automake", "libtool", "libfaad-dev"],
cmake_args=["-DENABLE_DOXYGEN=OFF"],
homepage="https://github.com/hboeglen/gr-dab",
),
_entry( _entry(
name="lora_sdr", name="lora_sdr",
description="LoRa PHY transceiver (CSS modulation/demodulation)", description="LoRa PHY transceiver (CSS modulation/demodulation)",
@ -257,8 +230,7 @@ CATALOG: dict[str, OOTModuleEntry] = {
description="IEEE 802.11a/g/p OFDM transceiver", description="IEEE 802.11a/g/p OFDM transceiver",
category="WiFi", category="WiFi",
git_url="https://github.com/bastibl/gr-ieee802-11", git_url="https://github.com/bastibl/gr-ieee802-11",
branch="maint-3.10", branch="main",
build_deps=["castxml"],
homepage="https://github.com/bastibl/gr-ieee802-11", homepage="https://github.com/bastibl/gr-ieee802-11",
), ),
_entry( _entry(
@ -266,8 +238,7 @@ CATALOG: dict[str, OOTModuleEntry] = {
description="IEEE 802.15.4 (Zigbee) O-QPSK transceiver", description="IEEE 802.15.4 (Zigbee) O-QPSK transceiver",
category="IoT", category="IoT",
git_url="https://github.com/bastibl/gr-ieee802-15-4", git_url="https://github.com/bastibl/gr-ieee802-15-4",
branch="maint-3.10", branch="main",
build_deps=["castxml"],
homepage="https://github.com/bastibl/gr-ieee802-15-4", homepage="https://github.com/bastibl/gr-ieee802-15-4",
), ),
_entry( _entry(
@ -275,7 +246,7 @@ CATALOG: dict[str, OOTModuleEntry] = {
description="ADS-B (1090 MHz) aircraft transponder decoder", description="ADS-B (1090 MHz) aircraft transponder decoder",
category="Aviation", category="Aviation",
git_url="https://github.com/mhostetter/gr-adsb", git_url="https://github.com/mhostetter/gr-adsb",
branch="maint-3.10", branch="main",
homepage="https://github.com/mhostetter/gr-adsb", homepage="https://github.com/mhostetter/gr-adsb",
), ),
_entry( _entry(
@ -283,24 +254,8 @@ CATALOG: dict[str, OOTModuleEntry] = {
description="Iridium satellite burst detector and demodulator", description="Iridium satellite burst detector and demodulator",
category="Satellite", category="Satellite",
git_url="https://github.com/muccc/gr-iridium", git_url="https://github.com/muccc/gr-iridium",
branch="master",
homepage="https://github.com/muccc/gr-iridium",
),
_entry(
name="leo",
description="LEO satellite channel simulator (Doppler, path loss, atmosphere)",
category="Satellite",
git_url="https://gitlab.com/librespacefoundation/gr-leo",
branch="gnuradio-3.10",
homepage="https://gitlab.com/librespacefoundation/gr-leo",
),
_entry(
name="dl5eu",
description="DVB-T OFDM synchronization and TPS decoder",
category="Broadcast",
git_url="https://github.com/dl5eu/gr-dl5eu",
branch="main", branch="main",
homepage="https://github.com/dl5eu/gr-dl5eu", homepage="https://github.com/muccc/gr-iridium",
), ),
_entry( _entry(
name="inspector", name="inspector",
@ -308,19 +263,24 @@ CATALOG: dict[str, OOTModuleEntry] = {
category="Analysis", category="Analysis",
git_url="https://github.com/gnuradio/gr-inspector", git_url="https://github.com/gnuradio/gr-inspector",
branch="master", branch="master",
build_deps=["qtbase5-dev", "libqwt-qt5-dev"],
homepage="https://github.com/gnuradio/gr-inspector", homepage="https://github.com/gnuradio/gr-inspector",
gr_versions="3.9 (master branch has API compat issues with 3.10)",
), ),
_entry( _entry(
name="nrsc5", name="nrsc5",
description="HD Radio (NRSC-5) digital broadcast decoder", description="HD Radio (NRSC-5) digital broadcast decoder",
category="Broadcast", category="Broadcast",
git_url="https://github.com/argilo/gr-nrsc5", git_url="https://github.com/argilo/gr-nrsc5",
branch="master", branch="main",
build_deps=["autoconf", "automake", "libtool"],
homepage="https://github.com/argilo/gr-nrsc5", homepage="https://github.com/argilo/gr-nrsc5",
), ),
_entry(
name="packet_radio",
description="Amateur packet radio (AFSK, AX.25) modem",
category="Amateur",
git_url="https://github.com/duggabe/gr-packet-radio",
branch="main",
homepage="https://github.com/duggabe/gr-packet-radio",
),
] ]
} }

382
src/gnuradio_mcp/providers/mcp_runtime.py
View File

@ -1,10 +1,8 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import Any, Callable
from fastmcp import Context, FastMCP from fastmcp import FastMCP
from pydantic import BaseModel
from gnuradio_mcp.middlewares.docker import DockerMiddleware from gnuradio_mcp.middlewares.docker import DockerMiddleware
from gnuradio_mcp.middlewares.oot import OOTInstallerMiddleware from gnuradio_mcp.middlewares.oot import OOTInstallerMiddleware
@ -13,361 +11,81 @@ from gnuradio_mcp.providers.runtime import RuntimeProvider
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class RuntimeModeStatus(BaseModel):
"""Status of runtime mode and available capabilities."""
enabled: bool
tools_registered: list[str]
docker_available: bool
oot_available: bool
class RootsCapability(BaseModel):
"""Client roots capability - expose workspace directories to servers."""
supported: bool = False
list_changed: bool | None = None # Client emits notifications when roots change
class SamplingCapability(BaseModel):
"""Client sampling capability - let servers request LLM completions.
Enables recursive agent patterns where servers can invoke the client's LLM.
"""
supported: bool = False
tools: bool = False # Allow tool use during sampling
context: bool = False # Include context from other servers (deprecated)
class ElicitationCapability(BaseModel):
"""Client elicitation capability - servers can prompt users for input.
Form mode: collect structured data via forms
URL mode: redirect to URLs for OAuth/payment/sensitive data
"""
supported: bool = False
form: bool = False # In-band structured data collection
url: bool = False # Out-of-band URL navigation for sensitive flows
class ClientCapabilities(BaseModel):
"""MCP client capability information from initialize handshake.
Based on MCP spec 2025-11-25. Clients advertise which features they support:
- roots: Expose workspace/project directories
- sampling: Let servers request LLM completions
- elicitation: Let servers prompt users for input
"""
client_name: str | None = None
client_version: str | None = None
protocol_version: str | None = None
# Structured capability objects
roots: RootsCapability = RootsCapability()
sampling: SamplingCapability = SamplingCapability()
elicitation: ElicitationCapability = ElicitationCapability()
# Raw capability dict for any unknown/future capabilities
raw_capabilities: dict[str, Any] = {}
experimental: dict[str, Any] = {}
class ClientRoot(BaseModel):
"""A root directory advertised by the MCP client."""
uri: str
name: str | None = None
class McpRuntimeProvider: class McpRuntimeProvider:
"""Registers runtime control tools with FastMCP. """Registers runtime control tools with FastMCP.
Uses dynamic tool registration to minimize context usage: Docker is optional: if unavailable, container lifecycle and visual
- At startup: only mode control tools are registered feedback tools are skipped, but XML-RPC connection/control tools
- When runtime mode is enabled: all runtime tools are registered are still registered (for connecting to externally-managed flowgraphs).
- When disabled: runtime tools are removed
This keeps the tool list small when only doing flowgraph design,
and expands it when connecting to SDR hardware or running flowgraphs.
""" """
def __init__(self, mcp_instance: FastMCP, runtime_provider: RuntimeProvider): def __init__(self, mcp_instance: FastMCP, runtime_provider: RuntimeProvider):
self._mcp = mcp_instance self._mcp = mcp_instance
self._provider = runtime_provider self._provider = runtime_provider
self._runtime_tools: dict[str, Callable] = {} self.__init_tools()
self._runtime_enabled = False
self.__init_mode_tools()
self.__init_resources() self.__init_resources()
def __init_mode_tools(self): def __init_tools(self):
"""Register only the mode control tools at startup."""
@self._mcp.tool
def get_runtime_mode() -> RuntimeModeStatus:
"""Check if runtime mode is enabled and what capabilities are available.
Runtime mode provides tools for:
- Connecting to running flowgraphs (XML-RPC, ControlPort)
- Launching flowgraphs in Docker containers
- Installing OOT modules
- Controlling SDR hardware
Call enable_runtime_mode() to register these tools.
"""
return RuntimeModeStatus(
enabled=self._runtime_enabled,
tools_registered=list(self._runtime_tools.keys()),
docker_available=self._provider._has_docker,
oot_available=self._provider._has_oot,
)
@self._mcp.tool
def enable_runtime_mode() -> RuntimeModeStatus:
"""Enable runtime mode, registering all runtime control tools.
This adds tools for:
- XML-RPC connection and variable control
- ControlPort/Thrift for performance monitoring
- Docker container lifecycle (if Docker available)
- OOT module installation (if Docker available)
Use this when you need to:
- Connect to a running flowgraph
- Launch flowgraphs in containers
- Control SDR hardware
- Monitor performance
"""
if self._runtime_enabled:
return RuntimeModeStatus(
enabled=True,
tools_registered=list(self._runtime_tools.keys()),
docker_available=self._provider._has_docker,
oot_available=self._provider._has_oot,
)
self._register_runtime_tools()
self._runtime_enabled = True
logger.info(
"Runtime mode enabled: registered %d tools",
len(self._runtime_tools),
)
return RuntimeModeStatus(
enabled=True,
tools_registered=list(self._runtime_tools.keys()),
docker_available=self._provider._has_docker,
oot_available=self._provider._has_oot,
)
@self._mcp.tool
def disable_runtime_mode() -> RuntimeModeStatus:
"""Disable runtime mode, removing runtime tools to reduce context.
Use this when you're done with runtime operations and want to
reduce the tool list for flowgraph design work.
"""
if not self._runtime_enabled:
return RuntimeModeStatus(
enabled=False,
tools_registered=[],
docker_available=self._provider._has_docker,
oot_available=self._provider._has_oot,
)
self._unregister_runtime_tools()
self._runtime_enabled = False
logger.info("Runtime mode disabled: removed runtime tools")
return RuntimeModeStatus(
enabled=False,
tools_registered=[],
docker_available=self._provider._has_docker,
oot_available=self._provider._has_oot,
)
# Debug tools for MCP client inspection
@self._mcp.tool
async def get_client_capabilities(ctx: Context) -> ClientCapabilities:
"""Get the connected MCP client's capabilities.
Returns information about the client including:
- Client name and version (e.g., "claude-code" v2.1.15)
- MCP protocol version
- Supported capabilities (roots, sampling, etc.)
- Experimental features
Useful for debugging MCP connections and understanding
what features the client supports.
"""
session = ctx.session
client_params = session.client_params if session else None
if client_params is None:
return ClientCapabilities()
client_info = getattr(client_params, "clientInfo", None)
caps = getattr(client_params, "capabilities", None)
result = ClientCapabilities(
client_name=client_info.name if client_info else None,
client_version=client_info.version if client_info else None,
protocol_version=getattr(client_params, "protocolVersion", None),
)
if caps:
# Roots capability - workspace directory exposure
if hasattr(caps, "roots") and caps.roots is not None:
result.roots = RootsCapability(
supported=True,
list_changed=getattr(caps.roots, "listChanged", None),
)
result.raw_capabilities["roots"] = {
"listChanged": getattr(caps.roots, "listChanged", None)
}
# Sampling capability - server-initiated LLM requests
if hasattr(caps, "sampling") and caps.sampling is not None:
sampling_obj = caps.sampling
result.sampling = SamplingCapability(
supported=True,
tools=hasattr(sampling_obj, "tools")
and sampling_obj.tools is not None,
context=hasattr(sampling_obj, "context")
and sampling_obj.context is not None,
)
result.raw_capabilities["sampling"] = {
"tools": result.sampling.tools,
"context": result.sampling.context,
}
# Elicitation capability - server-prompted user input
if hasattr(caps, "elicitation") and caps.elicitation is not None:
elicit_obj = caps.elicitation
# Empty object means form-only (backwards compat)
has_form = hasattr(elicit_obj, "form") and elicit_obj.form is not None
has_url = hasattr(elicit_obj, "url") and elicit_obj.url is not None
# If elicitation exists but no sub-caps, default to form
if not has_form and not has_url:
has_form = True
result.elicitation = ElicitationCapability(
supported=True,
form=has_form,
url=has_url,
)
result.raw_capabilities["elicitation"] = {
"form": has_form,
"url": has_url,
}
# Experimental features
if hasattr(caps, "experimental"):
result.experimental = caps.experimental or {}
return result
@self._mcp.tool
async def list_client_roots(ctx: Context) -> list[ClientRoot]:
"""List the root directories advertised by the MCP client.
Roots represent project directories or workspaces the client
wants the server to be aware of. Typically includes the
current working directory.
Returns empty list if roots capability is not supported.
"""
try:
roots = await ctx.list_roots()
return [
ClientRoot(uri=str(root.uri), name=root.name)
for root in roots
]
except Exception as e:
logger.warning("Failed to list client roots: %s", e)
return []
logger.info(
"Registered 5 mode control tools (runtime mode disabled by default)"
)
def _register_runtime_tools(self):
"""Dynamically register all runtime tools."""
p = self._provider p = self._provider
# Connection management # Connection management (always available)
self._add_tool("connect", p.connect) self._mcp.tool(p.connect)
self._add_tool("disconnect", p.disconnect) self._mcp.tool(p.disconnect)
self._add_tool("get_status", p.get_status) self._mcp.tool(p.get_status)
# Variable control # Variable control (always available)
self._add_tool("list_variables", p.list_variables) self._mcp.tool(p.list_variables)
self._add_tool("get_variable", p.get_variable) self._mcp.tool(p.get_variable)
self._add_tool("set_variable", p.set_variable) self._mcp.tool(p.set_variable)
# Flowgraph execution # Flowgraph execution (always available)
self._add_tool("start", p.start) self._mcp.tool(p.start)
self._add_tool("stop", p.stop) self._mcp.tool(p.stop)
self._add_tool("lock", p.lock) self._mcp.tool(p.lock)
self._add_tool("unlock", p.unlock) self._mcp.tool(p.unlock)
# ControlPort/Thrift tools # ControlPort/Thrift tools (always available - Phase 2)
self._add_tool("connect_controlport", p.connect_controlport) self._mcp.tool(p.connect_controlport)
self._add_tool("disconnect_controlport", p.disconnect_controlport) self._mcp.tool(p.disconnect_controlport)
self._add_tool("get_knobs", p.get_knobs) self._mcp.tool(p.get_knobs)
self._add_tool("set_knobs", p.set_knobs) self._mcp.tool(p.set_knobs)
self._add_tool("get_knob_properties", p.get_knob_properties) self._mcp.tool(p.get_knob_properties)
self._add_tool("get_performance_counters", p.get_performance_counters) self._mcp.tool(p.get_performance_counters)
self._add_tool("post_message", p.post_message) self._mcp.tool(p.post_message)
# Docker-dependent tools # Docker-dependent tools
if p._has_docker: if p._has_docker:
# Container lifecycle # Container lifecycle
self._add_tool("launch_flowgraph", p.launch_flowgraph) self._mcp.tool(p.launch_flowgraph)
self._add_tool("list_containers", p.list_containers) self._mcp.tool(p.list_containers)
self._add_tool("stop_flowgraph", p.stop_flowgraph) self._mcp.tool(p.stop_flowgraph)
self._add_tool("remove_flowgraph", p.remove_flowgraph) self._mcp.tool(p.remove_flowgraph)
self._add_tool("connect_to_container", p.connect_to_container) self._mcp.tool(p.connect_to_container)
self._add_tool( self._mcp.tool(p.connect_to_container_controlport) # Phase 2
"connect_to_container_controlport", p.connect_to_container_controlport
)
# Visual feedback # Visual feedback
self._add_tool("capture_screenshot", p.capture_screenshot) self._mcp.tool(p.capture_screenshot)
self._add_tool("get_container_logs", p.get_container_logs) self._mcp.tool(p.get_container_logs)
# Coverage collection # Coverage collection
self._add_tool("collect_coverage", p.collect_coverage) self._mcp.tool(p.collect_coverage)
self._add_tool("generate_coverage_report", p.generate_coverage_report) self._mcp.tool(p.generate_coverage_report)
self._add_tool("combine_coverage", p.combine_coverage) self._mcp.tool(p.combine_coverage)
self._add_tool("delete_coverage", p.delete_coverage) self._mcp.tool(p.delete_coverage)
# OOT module installation # OOT module installation
if p._has_oot: if p._has_oot:
self._add_tool("detect_oot_modules", p.detect_oot_modules) self._mcp.tool(p.install_oot_module)
self._add_tool("install_oot_module", p.install_oot_module) self._mcp.tool(p.list_oot_images)
self._add_tool("list_oot_images", p.list_oot_images) self._mcp.tool(p.remove_oot_image)
self._add_tool("remove_oot_image", p.remove_oot_image) logger.info("Registered 32 runtime tools (Docker + OOT available)")
self._add_tool("build_multi_oot_image", p.build_multi_oot_image) else:
self._add_tool("list_combo_images", p.list_combo_images) logger.info("Registered 29 runtime tools (Docker available)")
self._add_tool("remove_combo_image", p.remove_combo_image) else:
logger.info(
def _unregister_runtime_tools(self): "Registered 17 runtime tools (Docker unavailable, "
"""Remove all dynamically registered runtime tools.""" "container tools skipped)"
for name in list(self._runtime_tools.keys()): )
try:
self._mcp.remove_tool(name)
except Exception as e:
logger.warning("Failed to remove tool %s: %s", name, e)
self._runtime_tools.clear()
def _add_tool(self, name: str, func: Callable):
"""Add a tool and track it for later removal."""
self._mcp.add_tool(func)
self._runtime_tools[name] = func
def __init_resources(self): def __init_resources(self):
from gnuradio_mcp.oot_catalog import ( from gnuradio_mcp.oot_catalog import (

117
src/gnuradio_mcp/providers/runtime.py
View File

@ -12,15 +12,12 @@ from gnuradio_mcp.middlewares.oot import OOTInstallerMiddleware
from gnuradio_mcp.middlewares.thrift import ThriftMiddleware from gnuradio_mcp.middlewares.thrift import ThriftMiddleware
from gnuradio_mcp.middlewares.xmlrpc import XmlRpcMiddleware from gnuradio_mcp.middlewares.xmlrpc import XmlRpcMiddleware
from gnuradio_mcp.models import ( from gnuradio_mcp.models import (
ComboImageInfo,
ComboImageResult,
ConnectionInfoModel, ConnectionInfoModel,
ContainerModel, ContainerModel,
CoverageDataModel, CoverageDataModel,
CoverageReportModel, CoverageReportModel,
KnobModel, KnobModel,
KnobPropertiesModel, KnobPropertiesModel,
OOTDetectionResult,
OOTImageInfo, OOTImageInfo,
OOTInstallResult, OOTInstallResult,
PerfCounterModel, PerfCounterModel,
@ -110,7 +107,6 @@ class RuntimeProvider:
enable_perf_counters: bool = True, enable_perf_counters: bool = True,
device_paths: list[str] | None = None, device_paths: list[str] | None = None,
image: str | None = None, image: str | None = None,
auto_image: bool = False,
) -> ContainerModel: ) -> ContainerModel:
"""Launch a flowgraph in a Docker container with Xvfb. """Launch a flowgraph in a Docker container with Xvfb.
@ -125,17 +121,8 @@ class RuntimeProvider:
enable_perf_counters: Enable performance counters (requires controlport) enable_perf_counters: Enable performance counters (requires controlport)
device_paths: Host device paths to pass through device_paths: Host device paths to pass through
image: Docker image to use (e.g., 'gnuradio-lora-runtime:latest') image: Docker image to use (e.g., 'gnuradio-lora-runtime:latest')
auto_image: Automatically detect required OOT modules and build
appropriate Docker image. If True and image is not specified,
analyzes the flowgraph to determine OOT dependencies and
builds a single-OOT or combo image as needed.
""" """
docker = self._require_docker() docker = self._require_docker()
# Auto-detect and build image if requested
if auto_image and image is None and self._has_oot:
image = self._auto_select_image(flowgraph_path)
if name is None: if name is None:
name = f"gr-{Path(flowgraph_path).stem}" name = f"gr-{Path(flowgraph_path).stem}"
return docker.launch( return docker.launch(
@ -151,50 +138,6 @@ class RuntimeProvider:
image=image, image=image,
) )
def _auto_select_image(self, flowgraph_path: str) -> str | None:
"""Detect OOT modules and build/select appropriate image.
Auto-builds missing modules from catalog when needed.
"""
from gnuradio_mcp.oot_catalog import CATALOG
oot = self._require_oot()
detection = oot.detect_required_modules(flowgraph_path)
if not detection.detected_modules:
logger.info("No OOT modules detected, using base runtime image")
return detection.recommended_image
modules = detection.detected_modules
logger.info("Detected OOT modules: %s", modules)
if len(modules) == 1:
# Single module - ensure it's built
module = modules[0]
if module not in oot._registry:
entry = CATALOG.get(module)
if entry:
logger.info("Auto-building module '%s' from catalog", module)
result = oot.build_module(
git_url=entry.git_url,
branch=entry.branch,
build_deps=entry.build_deps or None,
cmake_args=entry.cmake_args or None,
)
if not result.success:
logger.error("Auto-build of '%s' failed: %s", module, result.error)
return None
info = oot._registry.get(module)
return info.image_tag if info else None
else:
# Multiple modules - build combo
logger.info("Building combo image for modules: %s", modules)
result = oot.build_combo_image(modules)
if result.success and result.image:
return result.image.image_tag
logger.error("Combo image build failed: %s", result.error)
return None
def list_containers(self) -> list[ContainerModel]: def list_containers(self) -> list[ContainerModel]:
"""List all gr-mcp managed containers.""" """List all gr-mcp managed containers."""
docker = self._require_docker() docker = self._require_docker()
@ -738,37 +681,9 @@ class RuntimeProvider:
return deleted return deleted
# ────────────────────────────────────────── # ──────────────────────────────────────────
# OOT Module Detection & Installation # OOT Module Installation
# ────────────────────────────────────────── # ──────────────────────────────────────────
def detect_oot_modules(self, flowgraph_path: str) -> OOTDetectionResult:
"""Detect which OOT modules a flowgraph requires.
Analyzes .py or .grc files to find OOT module dependencies.
Returns recommended Docker image to use with launch_flowgraph().
For .py files: parses Python imports (most accurate)
For .grc files: uses heuristic prefix matching against the
OOT catalog (fast, no Docker required)
Args:
flowgraph_path: Path to a .py or .grc flowgraph file
Returns:
OOTDetectionResult with detected modules, unknown blocks,
and recommended image tag.
Example:
result = detect_oot_modules("lora_rx.grc")
# -> detected_modules=["lora_sdr", "osmosdr"]
# -> recommended_image="gr-combo-lora_sdr-osmosdr:latest"
# Then launch with auto-built image:
launch_flowgraph("lora_rx.py", auto_image=True)
"""
oot = self._require_oot()
return oot.detect_required_modules(flowgraph_path)
def install_oot_module( def install_oot_module(
self, self,
git_url: str, git_url: str,
@ -803,33 +718,3 @@ class RuntimeProvider:
"""Remove an OOT module image and its registry entry.""" """Remove an OOT module image and its registry entry."""
oot = self._require_oot() oot = self._require_oot()
return oot.remove_image(module_name) return oot.remove_image(module_name)
# ──────────────────────────────────────────
# Multi-OOT Combo Images
# ──────────────────────────────────────────
def build_multi_oot_image(
self,
module_names: list[str],
force: bool = False,
) -> ComboImageResult:
"""Combine multiple OOT modules into a single Docker image.
Modules are merged using multi-stage Docker builds from existing
single-OOT images. Missing modules that exist in the catalog
are auto-built first.
Use the returned image_tag with launch_flowgraph().
"""
oot = self._require_oot()
return oot.build_combo_image(module_names, force)
def list_combo_images(self) -> list[ComboImageInfo]:
"""List all combined multi-OOT images."""
oot = self._require_oot()
return oot.list_combo_images()
def remove_combo_image(self, combo_key: str) -> bool:
"""Remove a combined image by its combo key (e.g., 'combo:adsb+lora_sdr')."""
oot = self._require_oot()
return oot.remove_combo_image(combo_key)

101
tests/integration/test_mcp_runtime.py
View File

@ -184,13 +184,8 @@ def runtime_mcp_app() -> FastMCP:
@pytest.fixture @pytest.fixture
async def runtime_client(runtime_mcp_app: FastMCP): async def runtime_client(runtime_mcp_app: FastMCP):
"""Create FastMCP client for runtime tools. """Create FastMCP client for runtime tools."""
Automatically enables runtime mode so runtime tools are available.
"""
async with Client(runtime_mcp_app) as client: async with Client(runtime_mcp_app) as client:
# Enable runtime mode to register runtime tools dynamically
await client.call_tool(name="enable_runtime_mode")
yield client yield client
@ -415,97 +410,3 @@ class TestRuntimeMcpToolsFullWorkflow:
assert extract_raw_value(result) == freq assert extract_raw_value(result) == freq
await runtime_client.call_tool(name="disconnect") await runtime_client.call_tool(name="disconnect")
class TestDynamicRuntimeMode:
"""Test dynamic tool registration via runtime mode toggle."""
async def test_runtime_mode_starts_disabled(self, runtime_mcp_app: FastMCP):
"""Runtime mode should be disabled by default."""
async with Client(runtime_mcp_app) as client:
result = await client.call_tool(name="get_runtime_mode")
assert result.data.enabled is False
assert result.data.tools_registered == []
async def test_enable_runtime_mode_registers_tools(self, runtime_mcp_app: FastMCP):
"""Enabling runtime mode should register runtime tools."""
async with Client(runtime_mcp_app) as client:
# Check tools before enabling
tools_before = await client.list_tools()
result = await client.call_tool(name="enable_runtime_mode")
assert result.data.enabled is True
assert len(result.data.tools_registered) > 0
assert "connect" in result.data.tools_registered
assert "list_variables" in result.data.tools_registered
# Verify tools are actually callable
tools_after = await client.list_tools()
assert len(tools_after) > len(tools_before)
async def test_disable_runtime_mode_removes_tools(self, runtime_mcp_app: FastMCP):
"""Disabling runtime mode should remove runtime tools."""
async with Client(runtime_mcp_app) as client:
# Enable first
await client.call_tool(name="enable_runtime_mode")
tools_enabled = await client.list_tools()
# Now disable
result = await client.call_tool(name="disable_runtime_mode")
assert result.data.enabled is False
assert result.data.tools_registered == []
# Verify tools are actually removed
tools_disabled = await client.list_tools()
assert len(tools_disabled) < len(tools_enabled)
async def test_enable_runtime_mode_idempotent(self, runtime_mcp_app: FastMCP):
"""Enabling runtime mode twice should be safe."""
async with Client(runtime_mcp_app) as client:
result1 = await client.call_tool(name="enable_runtime_mode")
result2 = await client.call_tool(name="enable_runtime_mode")
assert result1.data.enabled is True
assert result2.data.enabled is True
assert result1.data.tools_registered == result2.data.tools_registered
async def test_disable_runtime_mode_idempotent(self, runtime_mcp_app: FastMCP):
"""Disabling runtime mode twice should be safe."""
async with Client(runtime_mcp_app) as client:
result1 = await client.call_tool(name="disable_runtime_mode")
result2 = await client.call_tool(name="disable_runtime_mode")
assert result1.data.enabled is False
assert result2.data.enabled is False
class TestClientCapabilities:
"""Test MCP client capability inspection tools."""
async def test_get_client_capabilities_returns_structured_data(
self, runtime_mcp_app: FastMCP
):
"""get_client_capabilities should return structured capability info."""
async with Client(runtime_mcp_app) as client:
result = await client.call_tool(name="get_client_capabilities")
# Should have structured capability objects
assert result.data is not None
assert hasattr(result.data, "roots")
assert hasattr(result.data, "sampling")
assert hasattr(result.data, "elicitation")
# Each capability should have 'supported' field
assert hasattr(result.data.roots, "supported")
assert hasattr(result.data.sampling, "supported")
assert hasattr(result.data.elicitation, "supported")
async def test_list_client_roots_returns_list(self, runtime_mcp_app: FastMCP):
"""list_client_roots should return a list (may be empty in test)."""
async with Client(runtime_mcp_app) as client:
result = await client.call_tool(name="list_client_roots")
# Should return a list (FastMCP test client may not advertise roots)
assert isinstance(result.data, list) or result.data is None

602
tests/unit/test_oot_installer.py
View File

@ -1,7 +1,7 @@
"""Unit tests for OOT module installer middleware. """Unit tests for OOT module installer middleware.
Tests Dockerfile generation, module name extraction, registry persistence, Tests Dockerfile generation, module name extraction, registry persistence,
image naming, and OOT module detection — all without requiring Docker. and image naming — all without requiring Docker.
""" """
import json import json
@ -11,7 +11,7 @@ from unittest.mock import MagicMock
import pytest import pytest
from gnuradio_mcp.middlewares.oot import OOTInstallerMiddleware from gnuradio_mcp.middlewares.oot import OOTInstallerMiddleware
from gnuradio_mcp.models import ComboImageInfo, OOTDetectionResult, OOTImageInfo from gnuradio_mcp.models import OOTImageInfo
@pytest.fixture @pytest.fixture
@ -22,11 +22,9 @@ def mock_docker_client():
@pytest.fixture @pytest.fixture
def oot(mock_docker_client, tmp_path): def oot(mock_docker_client, tmp_path):
mw = OOTInstallerMiddleware(mock_docker_client) mw = OOTInstallerMiddleware(mock_docker_client)
# Override registry paths to use tmp_path # Override registry path to use tmp_path
mw._registry_path = tmp_path / "oot-registry.json" mw._registry_path = tmp_path / "oot-registry.json"
mw._registry = {} mw._registry = {}
mw._combo_registry_path = tmp_path / "oot-combo-registry.json"
mw._combo_registry = {}
return mw return mw
@ -96,9 +94,7 @@ class TestDockerfileGeneration:
assert "FROM gnuradio-runtime:latest" in dockerfile assert "FROM gnuradio-runtime:latest" in dockerfile
assert "git clone --depth 1 --branch master" in dockerfile assert "git clone --depth 1 --branch master" in dockerfile
assert "https://github.com/tapparelj/gr-lora_sdr.git" in dockerfile assert "https://github.com/tapparelj/gr-lora_sdr.git" in dockerfile
assert "cd gr-lora_sdr" in dockerfile assert "cd gr-lora_sdr && mkdir build" in dockerfile
assert "fix_binding_hashes.py" in dockerfile
assert "mkdir build" in dockerfile
assert "cmake -DCMAKE_INSTALL_PREFIX=/usr" in dockerfile assert "cmake -DCMAKE_INSTALL_PREFIX=/usr" in dockerfile
assert "make -j$(nproc)" in dockerfile assert "make -j$(nproc)" in dockerfile
assert "ldconfig" in dockerfile assert "ldconfig" in dockerfile
@ -187,44 +183,6 @@ class TestRegistry:
mw._save_registry() mw._save_registry()
assert mw._registry_path.exists() assert mw._registry_path.exists()
def test_load_skips_corrupt_entries(self, oot):
"""Per-entry validation: one corrupt entry doesn't nuke valid ones."""
data = {
"good_module": {
"module_name": "good_module",
"image_tag": "gr-oot-good:main-abc1234",
"git_url": "https://example.com/gr-good",
"branch": "main",
"git_commit": "abc1234",
"base_image": "gnuradio-runtime:latest",
"built_at": "2025-01-01T00:00:00+00:00",
},
"bad_module": {
"image_id": "sha256:deadbeef",
"build_deps": ["libfoo-dev"],
},
}
oot._registry_path.write_text(json.dumps(data))
loaded = oot._load_registry()
assert "good_module" in loaded
assert "bad_module" not in loaded
def test_build_module_reregisters_orphaned_image(self, oot):
"""build_module re-registers when image exists but registry empty."""
# Mock: _get_remote_commit returns a commit, _image_exists says yes
oot._get_remote_commit = MagicMock(return_value="abc1234")
oot._image_exists = MagicMock(return_value=True)
result = oot.build_module(
git_url="https://github.com/example/gr-test",
branch="main",
)
assert result.success is True
assert result.skipped is True
assert "test" in oot._registry
assert oot._registry["test"].image_tag == "gr-oot-test:main-abc1234"
# ────────────────────────────────────────── # ──────────────────────────────────────────
# Image Naming # Image Naming
@ -293,555 +251,3 @@ class TestRemoveImage:
result = oot.remove_image("lora_sdr") result = oot.remove_image("lora_sdr")
assert result is True assert result is True
assert "lora_sdr" not in oot._registry assert "lora_sdr" not in oot._registry
# ──────────────────────────────────────────
# Combo Key Generation
# ──────────────────────────────────────────
class TestComboKeyGeneration:
def test_sorted_and_deduped(self):
key = OOTInstallerMiddleware._combo_key(["lora_sdr", "adsb", "lora_sdr"])
assert key == "combo:adsb+lora_sdr"
def test_alphabetical_order(self):
key = OOTInstallerMiddleware._combo_key(["osmosdr", "adsb", "lora_sdr"])
assert key == "combo:adsb+lora_sdr+osmosdr"
def test_single_module(self):
key = OOTInstallerMiddleware._combo_key(["adsb"])
assert key == "combo:adsb"
def test_two_modules(self):
key = OOTInstallerMiddleware._combo_key(["lora_sdr", "adsb"])
assert key == "combo:adsb+lora_sdr"
# ──────────────────────────────────────────
# Combo Image Tag
# ──────────────────────────────────────────
class TestComboImageTag:
def test_format(self):
tag = OOTInstallerMiddleware._combo_image_tag(["lora_sdr", "adsb"])
assert tag == "gr-combo-adsb-lora_sdr:latest"
def test_sorted_and_deduped(self):
tag = OOTInstallerMiddleware._combo_image_tag(
["osmosdr", "adsb", "osmosdr"]
)
assert tag == "gr-combo-adsb-osmosdr:latest"
def test_three_modules(self):
tag = OOTInstallerMiddleware._combo_image_tag(
["lora_sdr", "adsb", "osmosdr"]
)
assert tag == "gr-combo-adsb-lora_sdr-osmosdr:latest"
# ──────────────────────────────────────────
# Combo Dockerfile Generation
# ──────────────────────────────────────────
def _make_oot_info(name: str, tag: str) -> OOTImageInfo:
"""Helper to create a minimal OOTImageInfo for testing."""
return OOTImageInfo(
module_name=name,
image_tag=tag,
git_url=f"https://example.com/gr-{name}.git",
branch="main",
git_commit="abc1234",
base_image="gnuradio-runtime:latest",
built_at="2025-01-01T00:00:00+00:00",
)
class TestComboDockerfileGeneration:
def test_multi_stage_structure(self, oot):
oot._registry["adsb"] = _make_oot_info("adsb", "gr-oot-adsb:main-abc1234")
oot._registry["lora_sdr"] = _make_oot_info(
"lora_sdr", "gr-oot-lora_sdr:master-def5678"
)
dockerfile = oot.generate_combo_dockerfile(["lora_sdr", "adsb"])
# Stage aliases (sorted order: adsb first)
assert "FROM gr-oot-adsb:main-abc1234 AS stage_adsb" in dockerfile
assert "FROM gr-oot-lora_sdr:master-def5678 AS stage_lora_sdr" in dockerfile
# Final base image
assert "FROM gnuradio-runtime:latest" in dockerfile
# COPY directives for both modules
assert "COPY --from=stage_adsb /usr/lib/ /usr/lib/" in dockerfile
assert "COPY --from=stage_adsb /usr/include/ /usr/include/" in dockerfile
assert "COPY --from=stage_adsb /usr/share/gnuradio/ /usr/share/gnuradio/" in dockerfile
assert "COPY --from=stage_lora_sdr /usr/lib/ /usr/lib/" in dockerfile
assert "COPY --from=stage_lora_sdr /usr/include/ /usr/include/" in dockerfile
# Runtime setup
assert "RUN ldconfig" in dockerfile
assert "WORKDIR /flowgraphs" in dockerfile
assert "PYTHONPATH" in dockerfile
def test_missing_module_raises(self, oot):
oot._registry["adsb"] = _make_oot_info("adsb", "gr-oot-adsb:main-abc1234")
with pytest.raises(ValueError, match="lora_sdr"):
oot.generate_combo_dockerfile(["adsb", "lora_sdr"])
def test_uses_configured_base_image(self, mock_docker_client, tmp_path):
mw = OOTInstallerMiddleware(mock_docker_client, base_image="my-custom:v2")
mw._registry_path = tmp_path / "oot-registry.json"
mw._registry = {
"adsb": _make_oot_info("adsb", "gr-oot-adsb:main-abc1234"),
"lora_sdr": _make_oot_info("lora_sdr", "gr-oot-lora_sdr:main-def5678"),
}
mw._combo_registry_path = tmp_path / "oot-combo-registry.json"
mw._combo_registry = {}
dockerfile = mw.generate_combo_dockerfile(["adsb", "lora_sdr"])
assert "FROM my-custom:v2" in dockerfile
# ──────────────────────────────────────────
# Combo Registry Persistence
# ──────────────────────────────────────────
class TestComboRegistry:
def test_separate_file(self, oot):
"""Combo registry uses a different file from single-OOT registry."""
assert oot._combo_registry_path != oot._registry_path
assert "combo" in str(oot._combo_registry_path)
def test_empty_on_fresh_start(self, oot):
assert oot._combo_registry == {}
assert oot.list_combo_images() == []
def test_save_and_load_roundtrip(self, oot):
info = ComboImageInfo(
combo_key="combo:adsb+lora_sdr",
image_tag="gr-combo-adsb-lora_sdr:latest",
modules=[
_make_oot_info("adsb", "gr-oot-adsb:main-abc1234"),
_make_oot_info("lora_sdr", "gr-oot-lora_sdr:master-def5678"),
],
built_at="2025-01-01T00:00:00+00:00",
)
oot._combo_registry["combo:adsb+lora_sdr"] = info
oot._save_combo_registry()
loaded = oot._load_combo_registry()
assert "combo:adsb+lora_sdr" in loaded
assert loaded["combo:adsb+lora_sdr"].image_tag == "gr-combo-adsb-lora_sdr:latest"
assert len(loaded["combo:adsb+lora_sdr"].modules) == 2
def test_load_missing_file_returns_empty(self, oot):
oot._combo_registry_path = oot._combo_registry_path.parent / "nope" / "r.json"
result = oot._load_combo_registry()
assert result == {}
def test_load_corrupt_file_returns_empty(self, oot):
oot._combo_registry_path.write_text("broken{{{")
result = oot._load_combo_registry()
assert result == {}
def test_list_returns_values(self, oot):
info = ComboImageInfo(
combo_key="combo:adsb+lora_sdr",
image_tag="gr-combo-adsb-lora_sdr:latest",
modules=[],
built_at="2025-01-01T00:00:00+00:00",
)
oot._combo_registry["combo:adsb+lora_sdr"] = info
result = oot.list_combo_images()
assert len(result) == 1
assert result[0].combo_key == "combo:adsb+lora_sdr"
def test_remove_existing(self, oot, mock_docker_client):
info = ComboImageInfo(
combo_key="combo:adsb+lora_sdr",
image_tag="gr-combo-adsb-lora_sdr:latest",
modules=[],
built_at="2025-01-01T00:00:00+00:00",
)
oot._combo_registry["combo:adsb+lora_sdr"] = info
result = oot.remove_combo_image("combo:adsb+lora_sdr")
assert result is True
assert "combo:adsb+lora_sdr" not in oot._combo_registry
mock_docker_client.images.remove.assert_called_once_with(
"gr-combo-adsb-lora_sdr:latest", force=True
)
def test_remove_nonexistent(self, oot):
result = oot.remove_combo_image("combo:nope+nada")
assert result is False
def test_remove_survives_docker_error(self, oot, mock_docker_client):
info = ComboImageInfo(
combo_key="combo:adsb+lora_sdr",
image_tag="gr-combo-adsb-lora_sdr:latest",
modules=[],
built_at="2025-01-01T00:00:00+00:00",
)
oot._combo_registry["combo:adsb+lora_sdr"] = info
mock_docker_client.images.remove.side_effect = Exception("gone")
result = oot.remove_combo_image("combo:adsb+lora_sdr")
assert result is True
assert "combo:adsb+lora_sdr" not in oot._combo_registry
# ──────────────────────────────────────────
# Build Combo Image
# ──────────────────────────────────────────
class TestBuildComboImage:
def test_requires_at_least_two_modules(self, oot):
result = oot.build_combo_image(["adsb"])
assert result.success is False
assert "2 distinct" in result.error
def test_rejects_duplicate_as_single(self, oot):
result = oot.build_combo_image(["adsb", "adsb"])
assert result.success is False
assert "2 distinct" in result.error
def test_idempotent_skip(self, oot, mock_docker_client):
"""Skips build if combo image already exists."""
oot._registry["adsb"] = _make_oot_info("adsb", "gr-oot-adsb:main-abc")
oot._registry["lora_sdr"] = _make_oot_info("lora_sdr", "gr-oot-lora:m-def")
existing = ComboImageInfo(
combo_key="combo:adsb+lora_sdr",
image_tag="gr-combo-adsb-lora_sdr:latest",
modules=[],
built_at="2025-01-01T00:00:00+00:00",
)
oot._combo_registry["combo:adsb+lora_sdr"] = existing
# Docker image exists
mock_docker_client.images.get.return_value = MagicMock()
result = oot.build_combo_image(["lora_sdr", "adsb"])
assert result.success is True
assert result.skipped is True
def test_happy_path(self, oot, mock_docker_client):
"""Builds combo from pre-existing single-OOT images."""
oot._registry["adsb"] = _make_oot_info("adsb", "gr-oot-adsb:main-abc1234")
oot._registry["lora_sdr"] = _make_oot_info(
"lora_sdr", "gr-oot-lora_sdr:master-def5678"
)
# Docker image does not exist yet
mock_docker_client.images.get.side_effect = Exception("not found")
# Mock successful build
mock_docker_client.images.build.return_value = (
MagicMock(),
[{"stream": "Step 1/5 : FROM ...\n"}],
)
result = oot.build_combo_image(["adsb", "lora_sdr"])
assert result.success is True
assert result.skipped is False
assert result.image is not None
assert result.image.combo_key == "combo:adsb+lora_sdr"
assert result.image.image_tag == "gr-combo-adsb-lora_sdr:latest"
assert len(result.image.modules) == 2
assert result.modules_built == []
# Verify persisted to combo registry
assert "combo:adsb+lora_sdr" in oot._combo_registry
def test_unknown_module_not_in_catalog(self, oot):
"""Fails if module not in registry and not in catalog."""
oot._registry["adsb"] = _make_oot_info("adsb", "gr-oot-adsb:main-abc1234")
result = oot.build_combo_image(["adsb", "totally_fake_module"])
assert result.success is False
assert "totally_fake_module" in result.error
assert "not found in the catalog" in result.error
def test_force_rebuilds(self, oot, mock_docker_client):
"""force=True bypasses idempotency check."""
oot._registry["adsb"] = _make_oot_info("adsb", "gr-oot-adsb:main-abc")
oot._registry["lora_sdr"] = _make_oot_info("lora_sdr", "gr-oot-lora:m-def")
existing = ComboImageInfo(
combo_key="combo:adsb+lora_sdr",
image_tag="gr-combo-adsb-lora_sdr:latest",
modules=[],
built_at="2025-01-01T00:00:00+00:00",
)
oot._combo_registry["combo:adsb+lora_sdr"] = existing
# Docker image exists
mock_docker_client.images.get.return_value = MagicMock()
mock_docker_client.images.build.return_value = (
MagicMock(),
[{"stream": "rebuilt\n"}],
)
result = oot.build_combo_image(["adsb", "lora_sdr"], force=True)
assert result.success is True
assert result.skipped is False
# ──────────────────────────────────────────
# OOT Detection from Python Files
# ──────────────────────────────────────────
class TestDetectFromPython:
def test_detects_gnuradio_import(self, oot, tmp_path):
"""Detects 'import gnuradio.MODULE' pattern."""
py_file = tmp_path / "test.py"
py_file.write_text(
"import gnuradio.lora_sdr as lora_sdr\n"
"from gnuradio import blocks\n"
)
result = oot.detect_required_modules(str(py_file))
assert result.detected_modules == ["lora_sdr"]
assert result.detection_method == "python_imports"
def test_detects_from_gnuradio_import(self, oot, tmp_path):
"""Detects 'from gnuradio import MODULE' pattern."""
py_file = tmp_path / "test.py"
py_file.write_text("from gnuradio import adsb, blocks, analog\n")
result = oot.detect_required_modules(str(py_file))
assert result.detected_modules == ["adsb"]
def test_detects_multiple_modules(self, oot, tmp_path):
"""Detects multiple OOT modules in one file."""
py_file = tmp_path / "test.py"
py_file.write_text(
"import gnuradio.lora_sdr as lora_sdr\n"
"from gnuradio import adsb\n"
)
result = oot.detect_required_modules(str(py_file))
assert result.detected_modules == ["adsb", "lora_sdr"]
def test_detects_toplevel_osmosdr(self, oot, tmp_path):
"""Detects top-level 'import osmosdr' (not in gnuradio namespace)."""
py_file = tmp_path / "test.py"
py_file.write_text("import osmosdr\n")
result = oot.detect_required_modules(str(py_file))
assert "osmosdr" in result.detected_modules
def test_detects_from_toplevel_import(self, oot, tmp_path):
"""Detects 'from MODULE import ...' for top-level OOT."""
py_file = tmp_path / "test.py"
py_file.write_text("from osmosdr import source\n")
result = oot.detect_required_modules(str(py_file))
assert "osmosdr" in result.detected_modules
def test_ignores_core_modules(self, oot, tmp_path):
"""Ignores core GNU Radio modules not in catalog."""
py_file = tmp_path / "test.py"
py_file.write_text(
"from gnuradio import blocks, analog, gr, digital, filter\n"
"import numpy\n"
)
result = oot.detect_required_modules(str(py_file))
assert result.detected_modules == []
def test_no_oot_returns_empty(self, oot, tmp_path):
"""Returns empty list when no OOT imports found."""
py_file = tmp_path / "test.py"
py_file.write_text("from gnuradio import blocks\nprint('hello')\n")
result = oot.detect_required_modules(str(py_file))
assert result.detected_modules == []
# ──────────────────────────────────────────
# OOT Detection from GRC Files
# ──────────────────────────────────────────
class TestDetectFromGrc:
def test_detects_prefixed_blocks(self, oot, tmp_path):
"""Detects blocks with OOT module prefix."""
grc_file = tmp_path / "test.grc"
grc_file.write_text(
"""
blocks:
- id: lora_sdr_gray_demap
name: gray0
- id: adsb_demod
name: demod0
- id: blocks_null_sink
name: null0
"""
)
result = oot.detect_required_modules(str(grc_file))
assert result.detected_modules == ["adsb", "lora_sdr"]
assert result.detection_method == "grc_prefix_heuristic"
def test_detects_exact_match_blocks(self, oot, tmp_path):
"""Detects blocks that exactly match module name."""
grc_file = tmp_path / "test.grc"
grc_file.write_text(
"""
blocks:
- id: osmosdr_source
name: src0
"""
)
result = oot.detect_required_modules(str(grc_file))
assert "osmosdr" in result.detected_modules
def test_reports_unknown_blocks(self, oot, tmp_path):
"""Reports blocks that look OOT but aren't in catalog."""
grc_file = tmp_path / "test.grc"
grc_file.write_text(
"""
blocks:
- id: unknown_module_block
name: blk0
- id: another_weird_thing
name: blk1
"""
)
result = oot.detect_required_modules(str(grc_file))
assert "unknown_module_block" in result.unknown_blocks
assert "another_weird_thing" in result.unknown_blocks
def test_ignores_core_blocks(self, oot, tmp_path):
"""Ignores core GNU Radio blocks."""
grc_file = tmp_path / "test.grc"
grc_file.write_text(
"""
blocks:
- id: blocks_null_sink
name: null0
- id: analog_sig_source_x
name: src0
- id: digital_constellation_decoder_cb
name: dec0
- id: qtgui_time_sink_x
name: time0
- id: filter_fir_filter_xxx
name: fir0
"""
)
result = oot.detect_required_modules(str(grc_file))
assert result.detected_modules == []
assert result.unknown_blocks == []
def test_ignores_special_blocks(self, oot, tmp_path):
"""Ignores variables, imports, and other special blocks."""
grc_file = tmp_path / "test.grc"
grc_file.write_text(
"""
blocks:
- id: variable
name: samp_rate
- id: variable_qtgui_range
name: freq
- id: import
name: import_0
- id: options
name: opts
- id: pad_source
name: in0
- id: virtual_sink
name: vsink0
"""
)
result = oot.detect_required_modules(str(grc_file))
assert result.detected_modules == []
assert result.unknown_blocks == []
def test_handles_empty_blocks(self, oot, tmp_path):
"""Handles GRC with no blocks."""
grc_file = tmp_path / "test.grc"
grc_file.write_text("blocks: []\n")
result = oot.detect_required_modules(str(grc_file))
assert result.detected_modules == []
def test_handles_missing_blocks_key(self, oot, tmp_path):
"""Handles GRC without blocks key."""
grc_file = tmp_path / "test.grc"
grc_file.write_text("options:\n id: test\n")
result = oot.detect_required_modules(str(grc_file))
assert result.detected_modules == []
# ──────────────────────────────────────────
# Image Recommendation Logic
# ──────────────────────────────────────────
class TestRecommendImage:
def test_recommends_base_for_no_modules(self, oot):
"""Returns base runtime image when no OOT modules."""
result = oot._recommend_image([])
assert result == "gnuradio-runtime:latest"
def test_recommends_single_oot_image_if_built(self, oot):
"""Returns single OOT image tag if already built."""
oot._registry["lora_sdr"] = _make_oot_info(
"lora_sdr", "gr-oot-lora_sdr:master-abc1234"
)
result = oot._recommend_image(["lora_sdr"])
assert result == "gr-oot-lora_sdr:master-abc1234"
def test_returns_none_if_single_not_built(self, oot):
"""Returns None if single module not yet built."""
result = oot._recommend_image(["lora_sdr"])
assert result is None
def test_recommends_combo_tag_if_exists(self, oot):
"""Returns existing combo image tag."""
combo_info = ComboImageInfo(
combo_key="combo:adsb+lora_sdr",
image_tag="gr-combo-adsb-lora_sdr:latest",
modules=[],
built_at="2025-01-01T00:00:00+00:00",
)
oot._combo_registry["combo:adsb+lora_sdr"] = combo_info
result = oot._recommend_image(["adsb", "lora_sdr"])
assert result == "gr-combo-adsb-lora_sdr:latest"
def test_recommends_combo_tag_format_if_not_built(self, oot):
"""Returns what combo tag would be if not yet built."""
result = oot._recommend_image(["adsb", "lora_sdr"])
assert result == "gr-combo-adsb-lora_sdr:latest"
def test_combo_tag_sorted_alphabetically(self, oot):
"""Combo tag always uses sorted module names."""
result = oot._recommend_image(["lora_sdr", "adsb"])
assert result == "gr-combo-adsb-lora_sdr:latest"
# ──────────────────────────────────────────
# Edge Cases and Error Handling
# ──────────────────────────────────────────
class TestDetectionEdgeCases:
def test_file_not_found(self, oot, tmp_path):
"""Raises FileNotFoundError for missing file."""
with pytest.raises(FileNotFoundError, match="Flowgraph not found"):
oot.detect_required_modules(str(tmp_path / "does_not_exist.py"))
def test_unsupported_extension(self, oot, tmp_path):
"""Raises ValueError for unsupported file type."""
txt_file = tmp_path / "test.txt"
txt_file.write_text("some content")
with pytest.raises(ValueError, match="Unsupported file type"):
oot.detect_required_modules(str(txt_file))
def test_result_includes_flowgraph_path(self, oot, tmp_path):
"""Result includes the analyzed path."""
py_file = tmp_path / "my_flowgraph.py"
py_file.write_text("from gnuradio import blocks\n")
result = oot.detect_required_modules(str(py_file))
assert str(py_file) in result.flowgraph_path