diff --git a/.github/workflows/solver-tests.yml b/.github/workflows/solver-tests.yml new file mode 100644 index 0000000..dbfb0c7 --- /dev/null +++ b/.github/workflows/solver-tests.yml @@ -0,0 +1,29 @@ +name: Solver regression tests + +on: + push: + branches: [main] + pull_request: + branches: [main] + +permissions: + contents: read + +jobs: + tests: + name: Python ${{ matrix.python-version }} + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ['3.11', '3.12'] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + - name: Install dependencies + run: python -m pip install -r requirements.txt + - name: Run solver regressions + run: python -m unittest discover -s tests -v diff --git a/tests/test_solver.py b/tests/test_solver.py new file mode 100644 index 0000000..26ef199 --- /dev/null +++ b/tests/test_solver.py @@ -0,0 +1,109 @@ +"""Regression tests for the signal-flow graph solver. + +Run from the repository root with: python -m unittest discover -s tests -v +""" +import unittest + +import sympy as sp + +from app import Connection, DiagramData, _delta_from_loops, solve_diagram, validate_diagram + + +def diagram(edges, *, nodes=None, input_node="S", output_node="T"): + return DiagramData( + nodes=nodes if nodes is not None else ["S", "T"], + input_node=input_node, + output_node=output_node, + connections=[ + Connection(id=f"edge-{index}", from_node=start, to_node=end, gain=gain) + for index, (start, end, gain) in enumerate(edges) + ], + ) + + +def assert_equivalent(testcase, actual, expected): + testcase.assertEqual(sp.simplify(sp.sympify(actual) - sp.sympify(expected)), 0) + + +class SolverRegressionTests(unittest.TestCase): + def test_series_edges_multiply(self): + result = solve_diagram( + diagram( + [("S", "A", "G1"), ("A", "T", "G2")], + nodes=["S", "A", "T"], + ) + ) + self.assertEqual(result["status"], "success") + self.assertFalse(result["mason_mismatch"]) + assert_equivalent(self, result["transfer_function_expr"], "G1*G2") + self.assertEqual(len(result["paths"]), 1) + + def test_parallel_edges_add_instead_of_overwriting(self): + result = solve_diagram( + diagram([("S", "T", "G1"), ("S", "T", "G2")]) + ) + self.assertEqual(result["status"], "success") + self.assertFalse(result["mason_mismatch"]) + assert_equivalent(self, result["transfer_function_expr"], "G1 + G2") + self.assertEqual(len(result["paths"]), 2) + + def test_single_feedback_loop(self): + result = solve_diagram( + diagram( + [("S", "A", "G"), ("A", "T", "H"), ("A", "A", "K")], + nodes=["S", "A", "T"], + ) + ) + self.assertEqual(result["status"], "success") + self.assertFalse(result["mason_mismatch"]) + assert_equivalent(self, result["transfer_function_expr"], "G*H/(1-K)") + self.assertEqual(len(result["loops"]), 1) + + def test_disjoint_loops_produce_second_order_delta_term(self): + loops = [ + {"nodes": {"A"}, "ordered_nodes": ["A"], "gain": sp.Symbol("L1"), "edge_ids": ["a"]}, + {"nodes": {"B"}, "ordered_nodes": ["B"], "gain": sp.Symbol("L2"), "edge_ids": ["b"]}, + ] + delta, non_touching = _delta_from_loops(loops) + assert_equivalent(self, str(delta), "(1-L1)*(1-L2)") + self.assertEqual(len(non_touching[2]), 1) + self.assertCountEqual(non_touching[2][0]["combo_edge_ids"], ["a", "b"]) + + def test_touching_loops_have_no_second_order_term(self): + loops = [ + {"nodes": {"A", "B"}, "ordered_nodes": ["A", "B"], "gain": sp.Symbol("L1")}, + {"nodes": {"B", "C"}, "ordered_nodes": ["B", "C"], "gain": sp.Symbol("L2")}, + ] + delta, non_touching = _delta_from_loops(loops) + assert_equivalent(self, str(delta), "1-L1-L2") + self.assertEqual(non_touching, {}) + + def test_validate_rejects_duplicate_node_names(self): + result = validate_diagram(diagram([], nodes=["S", "T", "S"])) + self.assertEqual(result["status"], "error") + self.assertIn("duplicate_node_names", {e["type"] for e in result["errors"]}) + + def test_validate_rejects_empty_gain(self): + result = validate_diagram(diagram([("S", "T", "")])) + self.assertEqual(result["status"], "error") + self.assertIn("missing_gain", {e["type"] for e in result["errors"]}) + + def test_validate_rejects_missing_output_node(self): + result = validate_diagram( + diagram([("S", "A", "G")], nodes=["S", "A"], output_node="T") + ) + self.assertEqual(result["status"], "error") + self.assertIn("missing_output_node", {e["type"] for e in result["errors"]}) + + def test_validate_reports_off_path_and_isolated_nodes(self): + result = validate_diagram( + diagram([("S", "T", "G")], nodes=["S", "T", "Z"]) + ) + self.assertEqual(result["status"], "success") + warning_types = {w["type"] for w in result["warnings"]} + self.assertIn("off_path_nodes", warning_types) + self.assertIn("isolated_nodes", warning_types) + + +if __name__ == "__main__": + unittest.main()