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.

Sign up for GitHub

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

Jump to bottom

Adapted to exception for vlan usage #99

Merged
merged 3 commits into from
Nov 22, 2023
Merged

Adapted to exception for vlan usage #99

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d5d6e7c
Adapted to exception for vlan usage
Alopalao Oct 30, 2023
14fa744
Upgraded exceptions
Alopalao Nov 9, 2023
ededd6a
Reverted dev.in
Alopalao Nov 20, 2023
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
24 changes: 14 additions & 10 deletions main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
wait_combine, wait_fixed, wait_random)

from kytos.core import KytosEvent, KytosNApp, log, rest
from kytos.core.exceptions import KytosTagError
from kytos.core.helpers import alisten_to, listen_to
from kytos.core.link import Link
from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
Expand Down Expand Up @@ -293,23 +294,26 @@ def use_vlan(self, switch: Switch) -> None:
return
for interface_id in switch.interfaces:
interface = switch.interfaces[interface_id]
added = interface.use_tags(self.controller, self.vlan_id)
if not added:
log.error(f"TAG {self.vlan_id} is not available in"
f" {switch.id}:{interface_id}.")
try:
interface.use_tags(self.controller, self.vlan_id)
except KytosTagError as err:
log.error(err)

def make_vlan_available(self, switch: Switch) -> None:
"""Makes vlan from interface available"""
if self.vlan_id is None:
return
for interface_id in switch.interfaces:
interface = switch.interfaces[interface_id]
added = interface.make_tags_available(
self.controller, self.vlan_id
)
if not added:
log.warning(f"Tag {self.vlan_id} was already available"
f" in {switch.id}:{interface_id}")
try:
conflict = interface.make_tags_available(
self.controller, self.vlan_id
)
if conflict:
log.warning(f"Tags {conflict} was already available"
f" in {switch.id}:{interface_id}")
except KytosTagError as err:
log.error(err)

@retry(
stop=stop_after_attempt(3),
Expand Down
19 changes: 17 additions & 2 deletions tests/unit/test_main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@
import pytest
from httpx import RequestError
from kytos.core.events import KytosEvent
from kytos.core.exceptions import (KytosTagsNotInTagRanges,
KytosTagsAreNotAvailable)
from kytos.lib.helpers import (get_controller_mock, get_interface_mock,
get_kytos_event_mock, get_switch_mock,
get_test_client)
Expand Down Expand Up @@ -542,7 +544,7 @@ def test_use_vlan(self, mock_log):
assert interface_a.use_tags.call_count == 1
assert interface_b.use_tags.call_count == 1

interface_a.use_tags.return_value = False
interface_a.use_tags.side_effect = KytosTagsAreNotAvailable([], "1")
self.napp.use_vlan(switch)
assert interface_a.use_tags.call_count == 2
assert interface_b.use_tags.call_count == 2
Expand All @@ -559,14 +561,18 @@ def test_make_vlan_available(self, mock_log):
switch = get_switch_mock("00:00:00:00:00:00:00:01", 0x04)
interface_a = get_interface_mock("mock_a", 1, switch)
interface_a.make_tags_available = MagicMock()
a_make_ava = interface_a.make_tags_available
a_make_ava.return_value = []
interface_b = get_interface_mock("mock_b", 2, switch)
interface_b.make_tags_available = MagicMock()
b_make_ava = interface_b.make_tags_available
b_make_ava.return_value = []
switch.interfaces = {1: interface_a, 2: interface_b}
self.napp.make_vlan_available(switch)
assert interface_a.make_tags_available.call_count == 1
assert interface_b.make_tags_available.call_count == 1

interface_a.make_tags_available.return_value = False
a_make_ava.return_value = [[3799, 3799]]
self.napp.make_vlan_available(switch)
assert interface_a.make_tags_available.call_count == 2
assert interface_b.make_tags_available.call_count == 2
Expand All @@ -577,6 +583,15 @@ def test_make_vlan_available(self, mock_log):
assert interface_a.make_tags_available.call_count == 2
assert interface_b.make_tags_available.call_count == 2

self.napp.vlan_id = 3799
b_make_ava.side_effect = KytosTagsNotInTagRanges(
[[3799, 3799]], "01:2"
)
self.napp.make_vlan_available(switch)
assert interface_a.make_tags_available.call_count == 3
assert interface_b.make_tags_available.call_count == 3
assert mock_log.error.call_count == 1

@patch('httpx.get')
def test_get_flows_by_switch_retry(self, mock_get):
"""Test get_flows_by_switch retry"""
Expand Down