diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index f212c7e..7d27340 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -12,6 +12,7 @@ Added
- Publish event ``kytos/topology.current`` for topology reconciliation
- Subscribed to event ``kytos/topology.get`` to publish the current topology
- Added ``notified_up_at`` internal reserved metadata
+- Enabling/disabling a switch or an interface will send ``link_up`` and ``link_down`` notifications
Changed
=======
diff --git a/README.rst b/README.rst
index aa16dce..6e7843d 100644
--- a/README.rst
+++ b/README.rst
@@ -63,6 +63,7 @@ Subscribed
- ``kytos/.*.liveness.(up|down)``
- ``kytos/.*.liveness.disabled``
- ``kytos/topology.get``
+- ``kytos/topology.notify_link_up_if_status``
Published
@@ -196,3 +197,18 @@ Content:
:target: https://github.com/kytos-ng/topology
.. |Tag| image:: https://img.shields.io/github/tag/kytos-ng/topology.svg
:target: https://github.com/kytos-ng/topology/tags
+
+
+kytos/topology.notify_link_up_if_status
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+Event reporting that the link was changed to 'down'. It contains the link instance.
+
+Content:
+
+.. code-block:: python3
+
+ {
+ 'reason': 'link_enabled'
+ 'link':
+ }
diff --git a/main.py b/main.py
index f70cdcc..39573b6 100644
--- a/main.py
+++ b/main.py
@@ -269,6 +269,7 @@ def enable_switch(self, dpid):
self.notify_switch_enabled(dpid)
self.notify_topology_update()
+ self.notify_switch_links_status(switch, "link enabled")
return jsonify("Operation successful"), 201
@rest('v3/switches//disable', methods=['POST'])
@@ -283,6 +284,7 @@ def disable_switch(self, dpid):
self.notify_switch_disabled(dpid)
self.notify_topology_update()
+ self.notify_switch_links_status(switch, "link disabled")
return jsonify("Operation successful"), 201
@rest('v3/switches//metadata')
@@ -357,12 +359,14 @@ def enable_interface(self, interface_enable_id=None, dpid=None):
interface = switch.interfaces[interface_number]
self.topo_controller.enable_interface(interface.id)
interface.enable()
+ self.notify_interface_link_status(interface, "link enabled")
except KeyError:
msg = f"Switch {dpid} interface {interface_number} not found"
return jsonify(msg), 404
else:
for interface in switch.interfaces.copy().values():
interface.enable()
+ self.notify_interface_link_status(interface, "link enabled")
self.topo_controller.upsert_switch(switch.id, switch.as_dict())
self.notify_topology_update()
return jsonify("Operation successful"), 200
@@ -385,12 +389,14 @@ def disable_interface(self, interface_disable_id=None, dpid=None):
interface = switch.interfaces[interface_number]
self.topo_controller.disable_interface(interface.id)
interface.disable()
+ self.notify_interface_link_status(interface, "link disabled")
except KeyError:
msg = f"Switch {dpid} interface {interface_number} not found"
return jsonify(msg), 404
else:
for interface in switch.interfaces.copy().values():
interface.disable()
+ self.notify_interface_link_status(interface, "link disabled")
self.topo_controller.upsert_switch(switch.id, switch.as_dict())
self.notify_topology_update()
return jsonify("Operation successful"), 200
@@ -764,7 +770,7 @@ def link_status_hook_link_up_timer(self, link) -> Optional[EntityStatus]:
return EntityStatus.DOWN
return None
- def notify_link_up_if_status(self, link) -> None:
+ def notify_link_up_if_status(self, link, reason="link up") -> None:
"""Tries to notify link up and topology changes based on its status
Currently, it needs to wait up to a timer."""
@@ -783,7 +789,7 @@ def notify_link_up_if_status(self, link) -> None:
link.update_metadata(key, now())
self.topo_controller.add_link_metadata(link.id, {key: notified_at})
self.notify_topology_update()
- self.notify_link_status_change(link, reason="link up")
+ self.notify_link_status_change(link, reason)
def handle_link_up(self, interface):
"""Handle link up for an interface."""
@@ -806,7 +812,7 @@ def handle_link_up(self, interface):
link.extend_metadata(metadata)
link.activate()
self.topo_controller.activate_link(link.id, **metadata)
- self.notify_link_up_if_status(link)
+ self.notify_link_up_if_status(link, "link up")
@listen_to('.*.switch.interface.link_down')
def on_interface_link_down(self, event):
@@ -910,7 +916,7 @@ def add_links(self, event):
}
link.extend_metadata(metadata)
self.topo_controller.upsert_link(link.id, link.as_dict())
- self.notify_link_up_if_status(link)
+ self.notify_link_up_if_status(link, "link up")
@listen_to('.*.of_lldp.network_status.updated')
def on_lldp_status_updated(self, event):
@@ -948,6 +954,19 @@ def notify_switch_enabled(self, dpid):
event = KytosEvent(name=name, content={'dpid': dpid})
self.controller.buffers.app.put(event)
+ def notify_switch_links_status(self, switch, reason):
+ """Send an event to notify the status of a link in a switch"""
+ with self._links_lock:
+ for link in self.links.values():
+ if switch in (link.endpoint_a.switch, link.endpoint_b.switch):
+ if reason == "link enabled":
+ name = 'kytos/topology.notify_link_up_if_status'
+ content = {'reason': reason, "link": link}
+ event = KytosEvent(name=name, content=content)
+ self.controller.buffers.app.put(event)
+ else:
+ self.notify_link_status_change(link, reason)
+
def notify_switch_disabled(self, dpid):
"""Send an event to notify that a switch is disabled."""
name = 'kytos/topology.switch.disabled'
@@ -961,6 +980,19 @@ def notify_topology_update(self):
self._get_topology()})
self.controller.buffers.app.put(event)
+ def notify_interface_link_status(self, interface, reason):
+ """Send an event to notify the status of a link from
+ an interface."""
+ link = self._get_link_from_interface(interface)
+ if link:
+ if reason == "link enabled":
+ name = 'kytos/topology.notify_link_up_if_status'
+ content = {'reason': reason, "link": link}
+ event = KytosEvent(name=name, content=content)
+ self.controller.buffers.app.put(event)
+ else:
+ self.notify_link_status_change(link, reason)
+
def notify_link_status_change(self, link, reason='not given'):
"""Send an event to notify about a status change on a link."""
name = 'kytos/topology.'
@@ -998,6 +1030,13 @@ def notify_metadata_changes(self, obj, action):
self.controller.buffers.app.put(event)
log.debug(f'Metadata from {obj.id} was {action}.')
+ @listen_to('kytos/topology.notify_link_up_if_status')
+ def on_notify_link_up_if_status(self, event):
+ """Tries to notify link up and topology changes"""
+ link = event.content["link"]
+ reason = event.content["reason"]
+ self.notify_link_up_if_status(link, reason)
+
@listen_to('.*.switch.port.created')
def on_notify_port_created(self, event):
"""Notify when a port is created."""
diff --git a/tests/integration/test_main.py b/tests/integration/test_main.py
index 5eba8f0..8305cf1 100644
--- a/tests/integration/test_main.py
+++ b/tests/integration/test_main.py
@@ -154,7 +154,8 @@ def test_get_event_listeners(self):
'.*.switch.interface.link_down',
'.*.switch.interface.link_up',
'.*.switch.(new|reconnected)',
- '.*.switch.port.created']
+ '.*.switch.port.created',
+ 'kytos/topology.notify_link_up_if_status']
self.assertCountEqual(expected_events, actual_events)
def test_verify_api_urls(self):
diff --git a/tests/unit/test_main.py b/tests/unit/test_main.py
index 8101b05..e281edc 100644
--- a/tests/unit/test_main.py
+++ b/tests/unit/test_main.py
@@ -87,7 +87,8 @@ def test_get_event_listeners(self):
'kytos/.*.liveness.(up|down)',
'kytos/.*.liveness.disabled',
'kytos/topology.get',
- '.*.switch.port.created']
+ '.*.switch.port.created',
+ 'kytos/topology.notify_link_up_if_status']
actual_events = self.napp.listeners()
self.assertCountEqual(expected_events, actual_events)
@@ -623,8 +624,9 @@ def test_fail_load_link(self, get_link_or_create_mock):
with self.assertRaises(RestoreError):
self.napp._load_link(link_attrs_fail)
+ @patch('napps.kytos.topology.main.Main.notify_switch_links_status')
@patch('napps.kytos.topology.main.Main.notify_topology_update')
- def test_enable_switch(self, mock_notify_topo):
+ def test_enable_switch(self, mock_notify_topo, mock_sw_l_status):
"""Test enable_switch."""
dpid = "00:00:00:00:00:00:00:01"
mock_switch = get_switch_mock(dpid)
@@ -637,6 +639,7 @@ def test_enable_switch(self, mock_notify_topo):
self.assertEqual(mock_switch.enable.call_count, 1)
self.napp.topo_controller.enable_switch.assert_called_once_with(dpid)
mock_notify_topo.assert_called()
+ mock_sw_l_status.assert_called()
# fail case
mock_switch.enable.call_count = 0
@@ -646,8 +649,9 @@ def test_enable_switch(self, mock_notify_topo):
self.assertEqual(response.status_code, 404, response.data)
self.assertEqual(mock_switch.enable.call_count, 0)
+ @patch('napps.kytos.topology.main.Main.notify_switch_links_status')
@patch('napps.kytos.topology.main.Main.notify_topology_update')
- def test_disable_switch(self, mock_notify_topo):
+ def test_disable_switch(self, mock_notify_topo, mock_sw_l_status):
"""Test disable_switch."""
dpid = "00:00:00:00:00:00:00:01"
mock_switch = get_switch_mock(dpid)
@@ -660,6 +664,7 @@ def test_disable_switch(self, mock_notify_topo):
self.assertEqual(mock_switch.disable.call_count, 1)
self.napp.topo_controller.disable_switch.assert_called_once_with(dpid)
mock_notify_topo.assert_called()
+ mock_sw_l_status.assert_called()
# fail case
mock_switch.disable.call_count = 0
@@ -1567,17 +1572,65 @@ def test_notify_link_up_if_status(
link = MagicMock(status=EntityStatus.UP)
link.get_metadata.return_value = now()
- assert not self.napp.notify_link_up_if_status(link)
+ assert not self.napp.notify_link_up_if_status(link, "link up")
link.update_metadata.assert_not_called()
mock_notify_topo.assert_not_called()
mock_notify_link.assert_not_called()
link = MagicMock(status=EntityStatus.UP)
link.get_metadata.return_value = now() - timedelta(seconds=60)
- assert not self.napp.notify_link_up_if_status(link)
+ assert not self.napp.notify_link_up_if_status(link, "link up")
link.update_metadata.assert_called()
self.napp.topo_controller.add_link_metadata.assert_called()
mock_notify_topo.assert_called()
mock_notify_link.assert_called()
assert mock_sleep.call_count == 2
+
+ @patch('napps.kytos.topology.main.Main.notify_link_status_change')
+ def test_notify_switch_links_status(self, mock_notify_link_status_change):
+ """Test switch links notification when switch status change"""
+ buffers_app_mock = MagicMock()
+ self.napp.controller.buffers.app = buffers_app_mock
+ dpid = "00:00:00:00:00:00:00:01"
+ mock_switch = get_switch_mock(dpid)
+ link1 = MagicMock()
+ link1.endpoint_a.switch = mock_switch
+ self.napp.links = {1: link1}
+
+ self.napp.notify_switch_links_status(mock_switch, "link enabled")
+ assert self.napp.controller.buffers.app.put.call_count == 1
+
+ self.napp.notify_switch_links_status(mock_switch, "link disabled")
+ assert self.napp.controller.buffers.app.put.call_count == 1
+ assert mock_notify_link_status_change.call_count == 1
+
+ # Without notification
+ link1.endpoint_a.switch = None
+ self.napp.notify_switch_links_status(mock_switch, "link enabled")
+ assert self.napp.controller.buffers.app.put.call_count == 1
+
+ @patch('napps.kytos.topology.main.Main.notify_link_status_change')
+ @patch('napps.kytos.topology.main.Main._get_link_from_interface')
+ def test_notify_interface_link_status(self, *args):
+ """Test interface links notification when enable"""
+ (mock_get_link_from_interface,
+ mock_notify_link_status_change) = args
+ buffers_app_mock = MagicMock()
+ self.napp.controller.buffers.app = buffers_app_mock
+ mock_link = MagicMock()
+ mock_get_link_from_interface.return_value = mock_link
+ self.napp.notify_interface_link_status(MagicMock(), "link enabled")
+ assert mock_get_link_from_interface.call_count == 1
+ assert self.napp.controller.buffers.app.put.call_count == 1
+
+ self.napp.notify_interface_link_status(MagicMock(), "link disabled")
+ assert mock_get_link_from_interface.call_count == 2
+ assert mock_notify_link_status_change.call_count == 1
+ assert self.napp.controller.buffers.app.put.call_count == 1
+
+ # Without notification
+ mock_get_link_from_interface.return_value = None
+ self.napp.notify_interface_link_status(MagicMock(), "link enabled")
+ assert mock_get_link_from_interface.call_count == 3
+ assert self.napp.controller.buffers.app.put.call_count == 1