diff --git a/src/gnuradio_mcp/middlewares/flowgraph.py b/src/gnuradio_mcp/middlewares/flowgraph.py index 9a045bf..b9a15f7 100644 --- a/src/gnuradio_mcp/middlewares/flowgraph.py +++ b/src/gnuradio_mcp/middlewares/flowgraph.py @@ -1,6 +1,5 @@ from __future__ import annotations -from functools import lru_cache from typing import TYPE_CHECKING, Optional from gnuradio.grc.core.blocks.block import Block @@ -45,11 +44,18 @@ class FlowGraphMiddleware(ElementMiddleware): block_middleware = self.get_block(block_name) self._flowgraph.remove_element(block_middleware._block) - @lru_cache(maxsize=None) def get_block(self, block_name: str) -> BlockMiddleware: - return BlockMiddleware( - next(block for block in self._flowgraph.blocks if block.name == block_name) + """Look up a block by name from the live flowgraph. + + Always queries the actual flowgraph state — no caching — so that + block renames, removals, and re-creations are immediately visible. + """ + block = next( + (b for b in self._flowgraph.blocks if b.name == block_name), None ) + if block is None: + raise KeyError(f"Block {block_name!r} not found in flowgraph") + return BlockMiddleware(block) def connect_blocks( self, src_port_model: PortModel, dst_port_model: PortModel diff --git a/tests/unit/test_flowgraph.py b/tests/unit/test_flowgraph.py index 159fb39..2f656f6 100644 --- a/tests/unit/test_flowgraph.py +++ b/tests/unit/test_flowgraph.py @@ -119,6 +119,48 @@ def test_block_disconnection(flowgraph_middleware: FlowGraphMiddleware, block_ke assert len(flowgraph_middleware.get_connections()) == 0 +def test_get_block_after_removal_raises(flowgraph_middleware: FlowGraphMiddleware): + """Removed blocks must not be accessible — prevents stale references.""" + model = flowgraph_middleware.add_block("analog_sig_source_x") + flowgraph_middleware.remove_block(model.name) + with pytest.raises(KeyError): + flowgraph_middleware.get_block(model.name) + + +def test_recreated_block_gets_fresh_state(flowgraph_middleware: FlowGraphMiddleware): + """Re-creating a block after removal must return the new instance.""" + model = flowgraph_middleware.add_block("analog_sig_source_x") + block_mw = flowgraph_middleware.get_block(model.name) + block_mw.set_params({"freq": "99999"}) + + flowgraph_middleware.remove_block(model.name) + model2 = flowgraph_middleware.add_block("analog_sig_source_x", model.name) + + block_mw2 = flowgraph_middleware.get_block(model2.name) + freq_param = next(p for p in block_mw2.params if p.key == "freq") + # New block should have default value, not the one we set before removal + assert freq_param.value != "99999" + + +def test_renamed_block_accessible_by_new_name( + flowgraph_middleware: FlowGraphMiddleware, +): + """After renaming via set_params(id=...), the new name must resolve.""" + model = flowgraph_middleware.add_block("variable") + old_name = model.name + block_mw = flowgraph_middleware.get_block(old_name) + block_mw.set_params({"id": "my_var", "value": "42"}) + + # New name works + new_block = flowgraph_middleware.get_block("my_var") + val = next(p for p in new_block.params if p.key == "value") + assert val.value == "42" + + # Old name is gone + with pytest.raises(KeyError): + flowgraph_middleware.get_block(old_name) + + def test_default_flowgraph_errors(flowgraph_middleware: FlowGraphMiddleware): for error in flowgraph_middleware.get_all_errors(): assert isinstance(error, ErrorModel)