Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 23 additions & 7 deletions tests/test_failure_policy.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,8 @@
import pytest
from unittest.mock import patch

from ngraph.failure_policy import (
FailurePolicy,
FailureRule,
FailureCondition,
_evaluate_condition,
)
import pytest

from ngraph.failure_policy import FailureCondition, FailurePolicy, FailureRule


def test_node_scope_all():
Expand Down Expand Up @@ -285,3 +281,23 @@ def test_cache_disabled():
nodes["N1"]["capacity"] = 10
second_fail = policy.apply_failures(nodes, links)
assert set(second_fail) == set()


def test_evaluate_conditions_invalid_logic():
"""_evaluate_conditions should raise for unsupported logic values."""
attrs = {"cap": 50}
conds = [FailureCondition(attr="cap", operator=">=", value=10)]
with pytest.raises(ValueError, match="Unsupported logic"):
FailurePolicy._evaluate_conditions(attrs, conds, "xor")


def test_select_entities_invalid_rule_type():
"""_select_entities should raise for unsupported rule_type."""
dummy_rule = FailureRule(
entity_scope="node",
conditions=[],
logic="any",
rule_type="bogus",
)
with pytest.raises(ValueError, match="Unsupported rule_type"):
FailurePolicy._select_entities({"A", "B"}, dummy_rule)
38 changes: 33 additions & 5 deletions tests/test_traffic_manager.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,9 @@
import pytest

from ngraph.network import Network, Node, Link
from ngraph.traffic_demand import TrafficDemand
from ngraph.lib.flow_policy import FlowPolicyConfig
from ngraph.lib.graph import StrictMultiDiGraph
from ngraph.lib.algorithms.base import MIN_FLOW
from ngraph.lib.demand import Demand

from ngraph.network import Link, Network, Node
from ngraph.traffic_demand import TrafficDemand
from ngraph.traffic_manager import TrafficManager


Expand Down Expand Up @@ -411,3 +408,34 @@ def test_estimate_rounds_no_capacities():
total_placed = tm.place_all_demands(placement_rounds="auto")
# The link has 0 capacity, so no actual flow can be placed.
assert total_placed == 0.0, "No capacity => no flow placed"


def test_get_traffic_results_aggregated_and_detailed(small_network):
"""TrafficManager.get_traffic_results should summarize correctly."""
demands = [
TrafficDemand(source_path="A", sink_path="C", demand=10.0),
TrafficDemand(source_path="B", sink_path="C", demand=5.0),
]
tm = TrafficManager(network=small_network, traffic_demands=demands)
tm.build_graph()
tm.expand_demands()
tm.place_all_demands()

agg = tm.get_traffic_results(detailed=False)
assert len(agg) == 2
for res in agg:
assert res.total_volume == res.placed_volume

detailed = tm.get_traffic_results(detailed=True)
assert len(detailed) == len(tm.demands)
for res in detailed:
assert res.total_volume == res.placed_volume


def test_estimate_rounds_typical_case(small_network):
"""_estimate_rounds should compute rounds based on demand/capacity ratio."""
demands = [TrafficDemand(source_path="A", sink_path="C", demand=20.0)]
tm = TrafficManager(network=small_network, traffic_demands=demands)
tm.build_graph()
tm.expand_demands()
assert tm._estimate_rounds() == 6
Comment on lines +436 to +441
Copy link

Copilot AI May 18, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[nitpick] Testing a private method (_estimate_rounds) can lead to brittle tests; consider verifying this behavior through the public API (e.g., using place_all_demands with known inputs).

Suggested change
"""_estimate_rounds should compute rounds based on demand/capacity ratio."""
demands = [TrafficDemand(source_path="A", sink_path="C", demand=20.0)]
tm = TrafficManager(network=small_network, traffic_demands=demands)
tm.build_graph()
tm.expand_demands()
assert tm._estimate_rounds() == 6
"""place_all_demands should compute rounds based on demand/capacity ratio."""
demands = [TrafficDemand(source_path="A", sink_path="C", demand=20.0)]
tm = TrafficManager(network=small_network, traffic_demands=demands)
tm.build_graph()
tm.expand_demands()
total_placed = tm.place_all_demands(placement_rounds="auto")
# With demand=20.0 and capacity=100.0, 6 rounds are expected to place all demands.
assert total_placed == 20.0, "All demands should be placed"
assert tm.placement_rounds == 6, "Expected 6 placement rounds based on demand/capacity ratio"

Copilot uses AI. Check for mistakes.
13 changes: 10 additions & 3 deletions tests/workflow/test_capacity_probe.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,10 @@
from unittest.mock import MagicMock

import pytest
from unittest.mock import MagicMock, call

from ngraph.network import Network, Node, Link
from ngraph.workflow.capacity_probe import CapacityProbe
from ngraph.lib.algorithms.base import FlowPlacement
from ngraph.network import Link, Network, Node
from ngraph.workflow.capacity_probe import CapacityProbe


@pytest.fixture
Expand Down Expand Up @@ -244,3 +245,9 @@ def test_capacity_probe_probe_reverse(mock_scenario):
assert "max_flow:[B -> A]" in flows
assert flows["max_flow:[A -> B]"] == 3.0
assert flows["max_flow:[B -> A]"] == 3.0


def test_capacity_probe_invalid_flow_placement():
"""Invalid flow_placement string should raise ValueError."""
with pytest.raises(ValueError, match="Invalid flow_placement"):
CapacityProbe(name="Bad", flow_placement="bogus")
Loading