diff --git a/tests/test_failure_policy.py b/tests/test_failure_policy.py index 2437a78..b77785f 100644 --- a/tests/test_failure_policy.py +++ b/tests/test_failure_policy.py @@ -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(): @@ -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) diff --git a/tests/test_traffic_manager.py b/tests/test_traffic_manager.py index d742087..c501391 100644 --- a/tests/test_traffic_manager.py +++ b/tests/test_traffic_manager.py @@ -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 @@ -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 diff --git a/tests/workflow/test_capacity_probe.py b/tests/workflow/test_capacity_probe.py index 663eafd..d606871 100644 --- a/tests/workflow/test_capacity_probe.py +++ b/tests/workflow/test_capacity_probe.py @@ -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 @@ -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")