"""Tests for the flowgraph layout organizer."""
import pytest
from organize_flowgraph import (
LayoutConfig,
classify_block,
build_connection_graph,
compute_block_depths,
)
class TestLayoutConfig:
"""Tests for LayoutConfig defaults."""
def test_default_values(self):
config = LayoutConfig()
assert config.margin_x == 16
assert config.margin_y == 16
assert config.options_x == 16
assert config.options_y == 16
assert config.setup_max_per_row == 8
def test_custom_values(self):
config = LayoutConfig(margin_x=32, signal_start_y=300)
assert config.margin_x == 32
assert config.signal_start_y == 300
class TestClassifyBlock:
"""Tests for block classification."""
def test_options_block(self):
block = {"id": "options", "parameters": {"id": "options"}}
assert classify_block(block) == "options"
def test_variable_block(self):
block = {"id": "variable", "parameters": {"id": "samp_rate"}}
assert classify_block(block) == "setup"
def test_gui_sink_qtgui(self):
block = {"id": "qtgui_freq_sink_x", "parameters": {"id": "qtgui_freq_sink_x_0"}}
assert classify_block(block) == "gui_sink"
def test_gui_sink_wxgui(self):
block = {"id": "wxgui_fftsink2", "parameters": {"id": "wxgui_fftsink2_0"}}
assert classify_block(block) == "gui_sink"
def test_signal_processing_block(self):
block = {"id": "analog_sig_source_x", "parameters": {"id": "analog_sig_source_x_0"}}
assert classify_block(block) == "signal"
def test_import_block(self):
block = {"id": "import", "parameters": {"id": "import_0"}}
assert classify_block(block) == "setup"
def test_gui_range_block(self):
block = {"id": "variable_qtgui_range", "parameters": {"id": "variable_qtgui_range_0"}}
result = classify_block(block)
assert result == "setup"
class TestBuildConnectionGraph:
"""Tests for connection graph construction."""
def test_empty_connections(self):
forward, reverse, sources, sinks = build_connection_graph([])
assert forward == {}
assert reverse == {}
assert sources == set()
assert sinks == set()
def test_simple_chain(self):
connections = [
["source_0", "0", "throttle_0", "0"],
["throttle_0", "0", "sink_0", "0"],
]
forward, reverse, sources, sinks = build_connection_graph(connections)
assert "throttle_0" in forward.get("source_0", set())
assert "sink_0" in forward.get("throttle_0", set())
assert "source_0" in sources
assert "sink_0" in sinks
def test_branching_connections(self):
connections = [
["source_0", "0", "filter_0", "0"],
["source_0", "0", "sink_0", "0"],
]
forward, reverse, sources, sinks = build_connection_graph(connections)
assert len(forward["source_0"]) == 2
class TestComputeBlockDepths:
"""Tests for depth computation in the signal flow graph."""
def test_linear_chain(self):
forward = {
"a": {"b"},
"b": {"c"},
}
reverse = {
"b": {"a"},
"c": {"b"},
}
sources = {"a"}
depths = compute_block_depths(forward, reverse, sources)
assert depths["a"] == 0
assert depths["b"] == 1
assert depths["c"] == 2
def test_parallel_paths(self):
forward = {
"a": {"b", "c"},
"b": {"d"},
"c": {"d"},
}
reverse = {
"b": {"a"},
"c": {"a"},
"d": {"b", "c"},
}
sources = {"a"}
depths = compute_block_depths(forward, reverse, sources)
assert depths["a"] == 0
assert depths["b"] == 1
assert depths["c"] == 1
assert depths["d"] == 2
def test_single_node(self):
forward = {}
reverse = {}
sources = {"a"}
depths = compute_block_depths(forward, reverse, sources)
assert depths["a"] == 0
