Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Raise an error when unreserve_vlan() can't find the request ID #222

Merged
merged 7 commits into from
Aug 30, 2024
Merged
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
12 changes: 11 additions & 1 deletion src/sdx_pce/topology/temanager.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
VlanTaggedPort,
)
from sdx_pce.topology.manager import TopologyManager
from sdx_pce.utils.exceptions import ValidationError
from sdx_pce.utils.exceptions import UnknownRequestError, ValidationError

UNUSED_VLAN = None

Expand Down Expand Up @@ -841,11 +841,21 @@ def unreserve_vlan(self, request_id: str):
"""
Return previously reserved VLANs back to the pool.
"""
found_assignment = False

for domain, port_table in self._vlan_tags_table.items():
for port, vlan_table in port_table.items():
for vlan, assignment in vlan_table.items():
if assignment == request_id:
vlan_table[vlan] = UNUSED_VLAN
found_assignment = True

# We should let the invoker know that we could not find the
# request ID.
if not found_assignment:
raise UnknownRequestError(
"Unknown connection request", request_id=request_id
)

# to be called by delete_connection()
def _unreserve_vlan_breakdown(self, break_down):
Expand Down
16 changes: 15 additions & 1 deletion src/sdx_pce/utils/exceptions.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,21 @@
class ValidationError(Exception):
"""
A custom exception class to indicate errors.
A custom exception to represent validation errors.
"""

def __init__(self, message):
super().__init__(message)


class UnknownRequestError(Exception):
"""
A custom exception to represent unknown requests.
"""

def __init__(self, message: str, request_id: str):
"""
:param message: a string containing the error message.
:param request_id: a string containing request ID.
"""
super().__init__(f"{message} (ID: {request_id})")
self.request_id = request_id
12 changes: 12 additions & 0 deletions tests/test_te_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from sdx_pce.load_balancing.te_solver import TESolver
from sdx_pce.models import ConnectionRequest, ConnectionSolution, TrafficMatrix
from sdx_pce.topology.temanager import TEManager
from sdx_pce.utils.exceptions import UnknownRequestError

from . import TestData

Expand Down Expand Up @@ -1250,3 +1251,14 @@ def test_connection_amlight_to_zaoxi_user_port_any(self):
self.assertIsInstance(segment.get("uni_z").get("tag").get("value"), int)
self.assertIsInstance(segment.get("uni_z").get("tag").get("tag_type"), int)
self.assertIsInstance(segment.get("uni_z").get("port_id"), str)

def test_unreserve_unknown_connection_requests(self):
"""
TEManager should raise an error if we attempt to unreserve a
non-existent connection request.
"""
request_id = "non-existent-request-id"

with self.assertRaises(UnknownRequestError) as e:
TEManager(topology_data=None).unreserve_vlan(request_id)
self.assertEqual(e.request_id, request_id)