improved test coverage

Author: networmix

tests/cli/test_cli_helpers.py

@@ -0,0 +1,70 @@
+from typing import Any
+
+from ngraph import cli as cli_mod
+
+
+class DummyStep:
+    def __init__(self) -> None:
+        self.name = "x"
+        self.src_path = "A/*"
+        self.dst_regex = "B.*"
+        self.empty = ""
+        self._private = "ignore"
+        self.count = 3
+
+
+class DummyNet:
+    def __init__(self) -> None:
+        self.calls: list[str] = []
+
+    def select_node_groups_by_path(self, pattern: str) -> dict[str, list[Any]]:
+        self.calls.append(pattern)
+        if pattern == "ERR":
+            raise RuntimeError("boom")
+
+        # Return two labeled groups with simple objects exposing .disabled
+        class N:
+            def __init__(self, disabled: bool) -> None:
+                self.disabled = disabled
+
+        return {"G1": [N(False), N(True)], "G2": [N(False)]}
+
+
+def test_format_table_and_plural() -> None:
+    table = cli_mod._format_table(
+        ["H1", "H2"], [["abc", "1"], ["defghi", "2"]], max_col_width=5
+    )
+    assert "H1" in table and "H2" in table
+    # Ensure clipping with ASCII ellipsis (max_col_width=5 -> keep 2 chars + '...')
+    assert "de..." in table
+
+    assert cli_mod._plural(1, "node") == "node"
+    assert cli_mod._plural(2, "node") == "nodes"
+    assert cli_mod._plural(2, "node", "vertices") == "vertices"
+
+
+def test_collect_and_summarize_node_matches() -> None:
+    step = DummyStep()
+    net = DummyNet()
+    summary = cli_mod._summarize_node_matches(step, net)
+    # Only *_path and *_regex fields considered
+    assert set(summary.keys()) == {"src_path", "dst_regex"}
+    # Each entry should include expected keys
+    for v in summary.values():
+        assert set(v.keys()) >= {
+            "pattern",
+            "groups",
+            "nodes",
+            "enabled_nodes",
+            "labels",
+        }
+        assert v["groups"] == 2
+        assert v["nodes"] == 3
+        assert v["enabled_nodes"] == 2
+
+
+def test_summarize_pattern_error_path() -> None:
+    net = DummyNet()
+    out = cli_mod._summarize_pattern("ERR", net)
+    assert out["pattern"] == "ERR"
+    assert "error" in out

tests/failure/manager/test_helpers_smoke.py

@@ -0,0 +1,8 @@
+def test_import_failure_manager_helper_modules() -> None:
+    import ngraph.failure.manager.aggregate as agg
+    import ngraph.failure.manager.enumerate as enum
+    import ngraph.failure.manager.simulate as sim
+
+    assert isinstance(agg.__doc__, str)
+    assert isinstance(enum.__doc__, str)
+    assert isinstance(sim.__doc__, str)

tests/failure/test_conditions_unit.py

@@ -0,0 +1,81 @@
+from typing import Any
+
+import pytest
+
+from ngraph.failure.conditions import (
+    FailureCondition,
+    evaluate_condition,
+    evaluate_conditions,
+)
+
+
+class TestEvaluateCondition:
+    def test_equality_and_inequality(self) -> None:
+        attrs = {"x": 5, "y": "abc"}
+        assert evaluate_condition(attrs, FailureCondition("x", "==", 5)) is True
+        assert evaluate_condition(attrs, FailureCondition("x", "!=", 6)) is True
+        assert evaluate_condition(attrs, FailureCondition("y", "==", "abc")) is True
+
+    def test_ordering_with_none_guard(self) -> None:
+        attrs = {"a": 3, "b": None}
+        assert evaluate_condition(attrs, FailureCondition("a", ">", 2)) is True
+        assert evaluate_condition(attrs, FailureCondition("a", ">=", 3)) is True
+        assert evaluate_condition(attrs, FailureCondition("a", "<", 10)) is True
+        assert evaluate_condition(attrs, FailureCondition("a", "<=", 3)) is True
+        # None comparisons must return False rather than raising
+        assert evaluate_condition(attrs, FailureCondition("b", ">", 0)) is False
+        assert evaluate_condition(attrs, FailureCondition("missing", "<", 0)) is False
+
+    def test_contains_and_not_contains(self) -> None:
+        attrs = {"s": "hello", "l": [1, 2, 3], "n": None, "i": 123}
+        assert (
+            evaluate_condition(attrs, FailureCondition("s", "contains", "ell")) is True
+        )
+        assert evaluate_condition(attrs, FailureCondition("l", "contains", 2)) is True
+        assert (
+            evaluate_condition(attrs, FailureCondition("s", "not_contains", "xyz"))
+            is True
+        )
+        # None yields False for contains and True for not_contains
+        assert evaluate_condition(attrs, FailureCondition("n", "contains", 1)) is False
+        assert (
+            evaluate_condition(attrs, FailureCondition("n", "not_contains", 1)) is True
+        )
+        # Non-iterable must not raise
+        assert evaluate_condition(attrs, FailureCondition("i", "contains", 1)) is False
+        assert (
+            evaluate_condition(attrs, FailureCondition("i", "not_contains", 1)) is True
+        )
+
+    def test_any_value_and_no_value(self) -> None:
+        attrs: dict[str, Any] = {"p": 0, "q": None}
+        assert evaluate_condition(attrs, FailureCondition("p", "any_value")) is True
+        assert evaluate_condition(attrs, FailureCondition("q", "any_value")) is True
+        assert (
+            evaluate_condition(attrs, FailureCondition("missing", "any_value")) is False
+        )
+        assert (
+            evaluate_condition(attrs, FailureCondition("missing", "no_value")) is True
+        )
+        assert evaluate_condition(attrs, FailureCondition("q", "no_value")) is True
+        assert evaluate_condition(attrs, FailureCondition("p", "no_value")) is False
+
+    def test_unsupported_operator_raises(self) -> None:
+        with pytest.raises(ValueError, match="Unsupported operator"):
+            evaluate_condition({}, FailureCondition("x", "bad"))
+
+
+class TestEvaluateConditions:
+    def test_and_or_logic(self) -> None:
+        attrs = {"x": 10, "y": "abc"}
+        conds = [FailureCondition("x", ">", 5), FailureCondition("y", "==", "abc")]
+        assert evaluate_conditions(attrs, conds, "and") is True
+        assert evaluate_conditions(attrs, conds, "or") is True
+
+        conds2 = [FailureCondition("x", "<", 5), FailureCondition("y", "!=", "abc")]
+        assert evaluate_conditions(attrs, conds2, "and") is False
+        assert evaluate_conditions(attrs, conds2, "or") is False
+
+    def test_unsupported_logic(self) -> None:
+        with pytest.raises(ValueError, match="Unsupported logic"):
+            evaluate_conditions({}, [], "xor")

tests/profiling/test_reporter_smoke.py

@@ -0,0 +1,6 @@
+def test_import_profiling_reporter_module() -> None:
+    # Module is a placeholder; ensure it imports
+    import ngraph.profiling.reporter as reporter
+
+    assert hasattr(reporter, "__doc__")
+    # Some environments may drop docstrings when -OO; only assert attribute exists

tests/solver/test_helpers_smoke.py

@@ -0,0 +1,4 @@
+def test_import_solver_helpers_module() -> None:
+    import ngraph.solver.helpers as helpers
+
+    assert hasattr(helpers, "__doc__")

tests/workflow/analysis/test_placement_matrix.py

@@ -0,0 +1,81 @@
+from typing import Any, Dict
+
+import pandas as pd
+import pytest
+
+from ngraph.workflow.analysis.placement_matrix import PlacementMatrixAnalyzer
+
+
+class TestPlacementMatrixAnalyzer:
+    def test_analyze_requires_step_name(self) -> None:
+        analyzer = PlacementMatrixAnalyzer()
+        with pytest.raises(ValueError, match="step_name required"):
+            analyzer.analyze({}, step_name=None)  # type: ignore[arg-type]
+
+    def test_analyze_no_envelopes(self) -> None:
+        analyzer = PlacementMatrixAnalyzer()
+        results: Dict[str, Dict[str, Any]] = {"step": {"placement_envelopes": {}}}
+        with pytest.raises(ValueError, match="No placement envelope data"):
+            analyzer.analyze(results, step_name="step")
+
+    def test_extract_filter_and_analyze_happy_path(self) -> None:
+        analyzer = PlacementMatrixAnalyzer()
+        # Mixed-quality input: some entries missing fields must be ignored
+        envs = {
+            "A->B|prio=0": {"src": "A", "dst": "B", "priority": 0, "mean": 0.8},
+            "A->C|prio=1": {"source": "A", "sink": "C", "priority": 1, "mean": 0.5},
+            # Invalid entries below should be skipped by _extract_matrix_data
+            "bad1": {"src": "A", "dst": None, "mean": 0.1},
+            "bad2": {"src": None, "dst": "B", "mean": 0.2},
+            "bad3": {"src": "X", "dst": "Y", "priority": 2},
+            "bad4": 42,
+        }
+        results = {"pm": {"placement_envelopes": envs}}
+
+        out = analyzer.analyze(results, step_name="pm")
+        assert out["status"] == "success"
+        assert out["step_name"] == "pm"
+
+        matrix_data = out["matrix_data"]
+        # Only two valid rows should remain
+        assert isinstance(matrix_data, list)
+        assert len(matrix_data) == 2
+        # Ensure schema
+        row0 = matrix_data[0]
+        for key in ("source", "destination", "ratio", "flow_path", "priority"):
+            assert key in row0
+
+        # Combined matrix should have sources as index and destinations as columns
+        pmatrix: pd.DataFrame = out["placement_matrix"]
+        assert set(pmatrix.index) == {"A"}
+        assert set(pmatrix.columns) == {"B", "C"}
+        assert pytest.approx(pmatrix.loc["A", "B"], rel=1e-9) == 0.8
+        assert pytest.approx(pmatrix.loc["A", "C"], rel=1e-9) == 0.5
+
+        # Per-priority matrices present for priorities 0 and 1
+        by_prio: Dict[int, pd.DataFrame] = out["placement_matrices"]
+        assert set(by_prio.keys()) == {0, 1}
+        assert pytest.approx(by_prio[0].loc["A", "B"], rel=1e-9) == 0.8
+
+        # Statistics computed with non-zero enforcement
+        stats: Dict[str, Any] = out["statistics"]
+        assert stats["has_data"] is True
+        assert stats["ratio_min"] <= stats["ratio_mean"] <= stats["ratio_max"]
+        assert stats["num_sources"] == 1
+        assert stats["num_destinations"] == 2
+
+    def test_calculate_statistics_empty_returns_flag(self) -> None:
+        analyzer = PlacementMatrixAnalyzer()
+        # Empty matrix yields has_data=False
+        empty_matrix = pd.DataFrame()
+        assert analyzer._calculate_statistics(empty_matrix) == {"has_data": False}
+
+    def test_analyze_and_display_step_raises_and_prints_on_error(
+        self, capsys: Any
+    ) -> None:
+        analyzer = PlacementMatrixAnalyzer()
+        with pytest.raises(ValueError):
+            analyzer.analyze_and_display_step({}, step_name="missing")
+        captured = capsys.readouterr()
+        # Should include the error banner
+        assert "❌ Placement matrix analysis failed" in captured.out