test coverage

Author: networmix

docs/reference/api-full.md

@@ -12,7 +12,7 @@ Quick links:
 - [CLI Reference](cli.md)
 - [DSL Reference](dsl.md)
 
-Generated from source code on: August 15, 2025 at 04:00 UTC
+Generated from source code on: August 15, 2025 at 11:31 UTC
 
 Modules auto-discovered: 72
 
@@ -2451,22 +2451,22 @@ convenience to run both.
 
 ## ngraph.workflow.analysis.capacity_matrix
 
-Capacity matrix analysis for MaxFlow results.
+Capacity matrix analysis.
 
-Consumes ``flow_results`` from a MaxFlow step and builds a node->node capacity
-matrix using the maximum placed value observed per pair across iterations
-(capacity ceiling under the tested failure set). Provides statistics and a
-heatmap for quick visual inspection.
+Consumes `flow_results` (from MaxFlow step). Builds node→node capacity matrix
+using the *maximum placed value observed* per pair across iterations (i.e., the
+capacity ceiling under the tested failure set). Provides stats and a heatmap.
 
 ### CapacityMatrixAnalyzer
 
-Analyze max-flow capacities into matrices, statistics, and plots.
+Analyze max-flow capacities into matrices/statistics/plots.
 
 **Methods:**
 
-- `analyze(self, results: 'dict[str, Any]', **kwargs) -> 'dict[str, Any]'` - Compute capacity matrix for a MaxFlow step.
+- `analyze(self, results: 'Dict[str, Any]', **kwargs) -> 'Dict[str, Any]'` - Return analysis outputs for a given results document.
 - `analyze_and_display(self, results: 'dict[str, Any]', **kwargs) -> 'None'` - Analyze results and render them in notebook format.
-- `display_analysis(self, analysis: 'dict[str, Any]', **kwargs) -> 'None'` - Render capacity matrix statistics and heatmap.
+- `analyze_and_display_step(self, results: 'Dict[str, Any]', **kwargs) -> 'None'` - Analyze and render capacity matrix for a single workflow step.
+- `display_analysis(self, analysis: 'Dict[str, Any]', **kwargs) -> 'None'` - Render analysis outputs in notebook format.
 - `get_description(self) -> 'str'` - Return a concise description of the analyzer purpose.
 
 ---
@@ -2628,8 +2628,8 @@ Includes bindings for ``NetworkStats``, ``MaximumSupportedDemand``,
 High-level summary analyzer for results documents.
 
 Provides quick counts of steps and basic categorisation by presence of
-``flow_results`` in the new schema. Also contains convenience helpers for
-``NetworkStats`` sections.
+``flow_results`` in the new schema. Also contains a small helper for
+``NetworkStats`` sections aimed at notebook usage.
 
 ### SummaryAnalyzer
 

ngraph/demand/manager/manager.py

@@ -20,9 +20,9 @@
 from ngraph.graph.strict_multidigraph import StrictMultiDiGraph
 from ngraph.model.network import Network
 
-if TYPE_CHECKING:
-    from ngraph.demand.matrix import TrafficMatrixSet
-    from ngraph.model.view import NetworkView
+if TYPE_CHECKING:  # pragma: no cover - typing-only imports
+    from ngraph.demand.matrix import TrafficMatrixSet  # pragma: no cover
+    from ngraph.model.view import NetworkView  # pragma: no cover
 
 
 def _new_td_map() -> Dict[str, List[Demand]]:

ngraph/workflow/analysis/__init__.py

@@ -11,7 +11,7 @@
 
 import itables.options as itables_opt
 import matplotlib.pyplot as plt
-from itables import show
+from itables import show  # pragma: no cover - display-only binding
 
 from .bac import BACAnalyzer
 from .base import AnalysisContext, NotebookAnalyzer

ngraph/workflow/analysis/bac.py

@@ -122,20 +122,22 @@ def availability_curve(series: pd.Series):
             return xs, avail
 
         x, a = availability_curve(s)
-        plt.figure(figsize=(9, 5.5))
-        sns.lineplot(x=x, y=a, drawstyle="steps-post", label=mode.capitalize())
+        plt.figure(figsize=(9, 5.5))  # pragma: no cover - display-only
+        sns.lineplot(
+            x=x, y=a, drawstyle="steps-post", label=mode.capitalize()
+        )  # pragma: no cover - display-only
         if overlay is not None and len(overlay) == len(s):
             xo, ao = availability_curve(overlay)
             sns.lineplot(
                 x=xo, y=ao, drawstyle="steps-post", label=overlay_label or "overlay"
-            )
+            )  # pragma: no cover - display-only
 
         plt.xlabel("Delivered bandwidth (Gbps)")
         plt.ylabel("Availability (≥x)")
         plt.title(f"Bandwidth-Availability Curve - {name}")
-        plt.grid(True, linestyle=":", linewidth=0.5)
-        plt.tight_layout()
-        plt.show()
+        plt.grid(True, linestyle=":", linewidth=0.5)  # pragma: no cover - display-only
+        plt.tight_layout()  # pragma: no cover - display-only
+        plt.show()  # pragma: no cover - display-only
 
     # ---------- helpers ----------
 

ngraph/workflow/analysis/capacity_matrix.py

@@ -118,13 +118,13 @@ def display_analysis(self, analysis: Dict[str, Any], **kwargs) -> None:
 
         # Heatmap
         # Scale figure size with matrix dimensions; bias toward readability
-        plt.figure(
+        plt.figure(  # pragma: no cover - display-only
             figsize=(
                 min(16, 2 + 0.35 * max(3, matrix.shape[1])),
                 min(12, 2 + 0.35 * max(3, matrix.shape[0])),
             )
         )
-        sns.heatmap(
+        sns.heatmap(  # pragma: no cover - display-only
             matrix.replace(0.0, np.nan),
             annot=False,
             fmt=".0f",
@@ -135,8 +135,8 @@ def display_analysis(self, analysis: Dict[str, Any], **kwargs) -> None:
         plt.title(f"Node→Node Capacity (Max over iterations) — {step}")
         plt.xlabel("Destination")
         plt.ylabel("Source")
-        plt.tight_layout()
-        plt.show()
+        plt.tight_layout()  # pragma: no cover - display-only
+        plt.show()  # pragma: no cover - display-only
 
     # ---------- helpers ----------
 

ngraph/workflow/analysis/latency.py

@@ -123,14 +123,16 @@ def display_analysis(self, analysis: dict[str, Any], **kwargs) -> None:
 
         print(f"✅ Latency/Stretch for {name} — iterations={len(df)}")
 
-        fig, ax = plt.subplots(figsize=(9, 5.5))
-        sns.scatterplot(data=df, x="mean_km_per_gbps", y="stretch", s=60)
+        fig, ax = plt.subplots(figsize=(9, 5.5))  # pragma: no cover - display-only
+        sns.scatterplot(
+            data=df, x="mean_km_per_gbps", y="stretch", s=60
+        )  # pragma: no cover - display-only
         ax.set_xlabel("Mean distance per Gbps (km/Gbps)")
         ax.set_ylabel("Latency stretch (≈avg path cost / baseline LB)")
         ax.set_title(f"Distance & Stretch by Failure Iteration - {name}")
         ax.grid(True, linestyle=":", linewidth=0.5)
-        plt.tight_layout()
-        plt.show()
+        plt.tight_layout()  # pragma: no cover - display-only
+        plt.show()  # pragma: no cover - display-only
 
         print("  Summary:")
         print(

ngraph/workflow/analysis/msd.py

@@ -60,25 +60,31 @@ def display_analysis(self, analysis: dict[str, Any], **kwargs) -> None:
             print("  No probe trace available.")
             return
 
-        plt.figure(figsize=(8, 5))
+        plt.figure(figsize=(8, 5))  # pragma: no cover - display-only
         sns.lineplot(
             data=trace,
             x="alpha",
             y="min_placement_ratio",
             marker="o",
             label="min placement ratio",
-        )
+        )  # pragma: no cover - display-only
         sns.scatterplot(
             data=trace,
             x="alpha",
             y="min_placement_ratio",
             hue=trace["feasible"].map({True: "feasible", False: "infeasible"}),
             legend=True,
-        )
-        plt.axvline(alpha_star, linestyle="--", linewidth=1.0, label="alpha*")
-        plt.xlabel("Alpha")
-        plt.ylabel("Min placement ratio across pairs")
-        plt.title(f"MSD bracketing/bisection trace — {name}")
-        plt.grid(True, linestyle=":", linewidth=0.5)
-        plt.tight_layout()
-        plt.show()
+        )  # pragma: no cover - display-only
+        plt.axvline(
+            alpha_star, linestyle="--", linewidth=1.0, label="alpha*"
+        )  # pragma: no cover - display-only
+        plt.xlabel("Alpha")  # pragma: no cover - display-only
+        plt.ylabel(
+            "Min placement ratio across pairs"
+        )  # pragma: no cover - display-only
+        plt.title(
+            f"MSD bracketing/bisection trace — {name}"
+        )  # pragma: no cover - display-only
+        plt.grid(True, linestyle=":", linewidth=0.5)  # pragma: no cover - display-only
+        plt.tight_layout()  # pragma: no cover - display-only
+        plt.show()  # pragma: no cover - display-only

ngraph/workflow/analysis/placement_matrix.py

@@ -180,7 +180,7 @@ def fmt(x: float) -> str:
             print("No placement data available")
             return
 
-        for prio in sorted(matrices.keys()):
+        for prio in sorted(matrices.keys()):  # pragma: no cover - display-only
             print(f"\nPriority {prio}")
             stats = stats_by_prio.get(prio, {"has_data": False})
             if not stats.get("has_data"):
@@ -195,7 +195,7 @@ def fmt(x: float) -> str:
             matrix_display = matrices[prio].copy()
             matrix_display.index.name = "Source"
             matrix_display.columns.name = "Destination"
-            if not matrix_display.empty:
+            if not matrix_display.empty:  # pragma: no cover - display-only
                 md = matrix_display.applymap(fmt)
                 show(
                     md,

tests/cli/test_cli_logging_smoke.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+from ngraph import cli
+
+
+def test_cli_verbose_and_quiet_switch_levels(
+    caplog, tmp_path: Path, monkeypatch
+) -> None:
+    # Minimal scenario
+    scenario = tmp_path / "t.yaml"
+    scenario.write_text(
+        """
+network:
+  nodes:
+    A: {}
+workflow:
+  - step_type: BuildGraph
+"""
+    )
+
+    # run in temp directory to avoid polluting repo
+    monkeypatch.chdir(tmp_path)
+
+    # verbose enables debug
+    with caplog.at_level(logging.DEBUG, logger="ngraph"):
+        cli.main(
+            ["--verbose", "run", str(scenario), "--no-results"]
+        )  # avoid writing results
+    assert any("Debug logging enabled" in r.message for r in caplog.records)
+
+    # quiet suppresses info
+    caplog.clear()
+    with caplog.at_level(logging.INFO, logger="ngraph"):
+        cli.main(
+            ["--quiet", "run", str(scenario), "--no-results"]
+        )  # avoid writing results
+    assert not any(r.levelno == logging.INFO for r in caplog.records)

tests/demand/manager/test_builder_unit.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import pytest
+
+from ngraph.demand.manager.builder import (
+    _coerce_flow_policy_config,
+    build_traffic_matrix_set,
+)
+from ngraph.demand.matrix import TrafficMatrixSet
+from ngraph.flows.policy import FlowPolicyConfig
+
+
+def test_coerce_flow_policy_config_variants() -> None:
+    assert _coerce_flow_policy_config(None) is None
+    assert (
+        _coerce_flow_policy_config(FlowPolicyConfig.SHORTEST_PATHS_ECMP)
+        == FlowPolicyConfig.SHORTEST_PATHS_ECMP
+    )
+    assert (
+        _coerce_flow_policy_config(int(FlowPolicyConfig.SHORTEST_PATHS_ECMP))
+        == FlowPolicyConfig.SHORTEST_PATHS_ECMP
+    )
+    assert (
+        _coerce_flow_policy_config(str(int(FlowPolicyConfig.SHORTEST_PATHS_ECMP)))
+        == FlowPolicyConfig.SHORTEST_PATHS_ECMP
+    )
+    assert (
+        _coerce_flow_policy_config("shortest_paths_ecmp")
+        == FlowPolicyConfig.SHORTEST_PATHS_ECMP
+    )
+    with pytest.raises(ValueError):
+        _coerce_flow_policy_config("not-an-enum")
+
+
+def test_build_traffic_matrix_set_happy_and_errors() -> None:
+    raw = {
+        "default": [
+            {
+                "source_path": "A",
+                "sink_path": "B",
+                "demand": 10.0,
+                "priority": 0,
+                "flow_policy_config": "shortest_paths_ecmp",
+            }
+        ]
+    }
+    tms = build_traffic_matrix_set(raw)
+    assert isinstance(tms, TrafficMatrixSet)
+    m = tms.get_default_matrix()
+    assert m[0].flow_policy_config == FlowPolicyConfig.SHORTEST_PATHS_ECMP
+
+    with pytest.raises(ValueError):
+        build_traffic_matrix_set([1, 2, 3])  # type: ignore[arg-type]
+    with pytest.raises(ValueError):
+        build_traffic_matrix_set({"x": 1})  # type: ignore[arg-type]
+    with pytest.raises(ValueError):
+        build_traffic_matrix_set({"x": [1]})  # type: ignore[arg-type]