Using events now

CHANGELOG.rst

@@ -6,6 +6,13 @@ All notable changes to the coloring NApp will be documented in this file.
 [UNRELEASED] - Under development
 ********************************
 
+[2024.1.2] - 2024-08-30
+***********************
+
+Changed
+=======
+- Flow installation is done through events now.
+
 [2024.1.1] - 2024-08-23
 ***********************
 

main.py

@@ -9,7 +9,7 @@
 import struct
 from threading import Lock
 from collections import defaultdict
-import httpx
+
 from kytos.core import KytosNApp, log, rest
 from kytos.core.common import EntityStatus
 from kytos.core.helpers import listen_to, alisten_to
@@ -125,7 +125,7 @@ def update_colors(self, links):
                         switch_dict['flows'][neighbor] = flow_dict
                         dpid_flows[dpid].append(flow_dict)
 
-        self._send_flow_mods(dpid_flows)
+        self._send_flow_mods(dpid_flows, "install")
 
     def handle_link_disabled(self, link):
         """Handle link deletion. Deletes only flows from the proper switches.
@@ -152,7 +152,7 @@ def handle_link_disabled(self, link):
             })
             self.switches[switch_a_id]['flows'].pop(switch_b_id)
             self.switches[switch_b_id]['flows'].pop(switch_a_id)
-        self._remove_flow_mods(flow_mods)
+        self._send_flow_mods(flow_mods, "delete")
 
     def handle_switch_disabled(self, dpid):
         """Handle switch deletion. Links are expected to be disabled first
@@ -214,29 +214,12 @@ def _switch_colors(self) -> dict:
                                 )}
             return colors
 
-    # pylint: disable=missing-timeout
-    def _send_flow_mods(self, dpid_flows: dict) -> None:
-        """Send FlowMods."""
-        with httpx.Client() as client:
-            for dpid, flows in dpid_flows.items():
-                try:
-                    res = client.post(
-                        self._flow_manager_url % dpid,
-                        json={'flows': flows, 'force': True},
-                        timeout=10
-                    )
-                except httpx.TimeoutException as err:
-                    log.error(f"Failed to post flows, error: {err}")
-                    continue
-                if res.status_code // 100 != 2:
-                    log.error(f'Flow manager returned an error inserting '
-                              f'flows {flows}. Status code {res.status_code} '
-                              f'on dpid {dpid}')
-
-    def _remove_flow_mods(self, flows: dict, force: bool = True) -> None:
+    def _send_flow_mods(
+        self, flows: dict, action: str, force: bool = True
+    ) -> None:
         """Remove FlowMods"""
         for dpid, mod_flows in flows.items():
-            name = "kytos.flow_manager.flows.delete"
+            name = f"kytos.flow_manager.flows.{action}"
             content = {
                 'dpid': dpid,
                 'flow_dict': {'flows': mod_flows},

requirements/run.in

@@ -1 +0,0 @@
-httpx==0.27.0

requirements/run.txt

@@ -3,24 +3,4 @@
 # by the following command:
 #
 #    pip-compile --output-file=requirements/run.txt requirements/run.in
-#
-anyio==4.3.0
-    # via httpx
-certifi==2024.2.4
-    # via
-    #   httpcore
-    #   httpx
-h11==0.14.0
-    # via httpcore
-httpcore==1.0.4
-    # via httpx
-httpx==0.27.0
-    # via -r requirements/run.in
-idna==3.6
-    # via
-    #   anyio
-    #   httpx
-sniffio==1.3.1
-    # via
-    #   anyio
-    #   httpx
+#

tests/unit/test_main.py

@@ -1,6 +1,5 @@
 """Test the Main class."""
 from unittest.mock import AsyncMock, Mock, patch, MagicMock
-from httpx import TimeoutException
 from kytos.lib.helpers import get_controller_mock, get_test_client
 
 from kytos.core.common import EntityStatus
@@ -113,11 +112,9 @@ async def test_rest_settings(self):
         topology_url = json_response['topology_url']
         assert topology_url.endswith("/api/kytos/topology/v3/links")
 
-    # pylint: disable=too-many-statements, unnecessary-dunder-call
-    @patch('httpx.Client')
-    def test_update_colors(self, client_mock):
+    # pylint: disable=too-many-statements
+    def test_update_colors(self):
         """Test method update_colors."""
-        post_mock = client_mock().__enter__().post
         switch1 = Mock()
         switch1.dpid = '00:00:00:00:00:00:00:01'
         switch1.ofp_version = '0x04'
@@ -186,8 +183,9 @@ def switch_by_dpid(dpid):
         cookie = self.napp.get_cookie(switch2.dpid)
         assert sw2['flows'][dpid1]['cookie'] == cookie
 
-        # Tests that the FLOW_MANAGER_URL was called twice to insert flow.
-        assert post_mock.call_count == 2
+        put_mock = self.napp.controller.buffers.app.put
+        # Tests that the FLOW_MANAGER_URL was called twice to insert flow
+        assert put_mock.call_count == 2
 
         # Verify switches with no neighbors, flows cleanup is performed
         # by handle_link_disabled()
@@ -215,20 +213,9 @@ def switch_by_dpid(dpid):
             }
         ]
 
-        post_mock.reset_mock()
+        put_mock.reset_mock()
         self.napp.update_colors(links2)
-        post_mock.assert_not_called()
-
-    # pylint: disable=protected-access, unnecessary-dunder-call
-    @patch('napps.amlight.coloring.main.log')
-    @patch('httpx.Client')
-    def test_send_flow_mods_error(self, client_mock, mock_log):
-        """Test _send_flow_mods with Timeout exception"""
-        post_mock = client_mock().__enter__().post
-        flows = {'00:01': ['mock_flow']}
-        post_mock.side_effect = TimeoutException('Timeout')
-        self.napp._send_flow_mods(flows)
-        assert mock_log.error.call_count == 1
+        put_mock.assert_not_called()
 
     def test_update_colors_without_links(self):
         """Test method update_colors without links."""
@@ -329,8 +316,8 @@ def test_handle_switch_disabled(self, mock_log):
         self.napp.handle_switch_disabled("mock_switch")
         assert mock_log.error.call_count == 2
 
-    @patch('napps.amlight.coloring.main.Main._remove_flow_mods')
-    def test_handle_link_disabled(self, mock_remove):
+    @patch('napps.amlight.coloring.main.Main._send_flow_mods')
+    def test_handle_link_disabled(self, mock_send_flow):
         """Test handle_link_disabled"""
         self.napp.switches = {
             '00:00:00:00:00:00:00:01': {
@@ -360,27 +347,33 @@ def test_handle_link_disabled(self, mock_remove):
         self.napp.handle_link_disabled(link)
         assert not self.napp.switches['00:00:00:00:00:00:00:01']['flows']
         assert not self.napp.switches['00:00:00:00:00:00:00:02']['flows']
-        assert mock_remove.call_count == 1
+        assert mock_send_flow.call_count == 1
 
         link.endpoint_b.switch.dpid = '00:00:00:00:00:00:00:01'
         self.napp.handle_link_disabled(link)
-        assert mock_remove.call_count == 1
+        assert mock_send_flow.call_count == 1
 
     # pylint: disable=protected-access
-    def test_remove_flow_mods(self):
-        """Test _remove_flow_mods"""
+    def test_send_flow_mods(self):
+        """Test _send_flow_mods"""
         flows = {
             "00:01": [{
                 'match': {'dl_src': 'ee:ee:ee:ee:ee:02'},
                 'table_id': 0
             }]
         }
-        self.napp._remove_flow_mods(flows)
+        self.napp._send_flow_mods(flows, "delete")
         args = self.napp.controller.buffers.app.put.call_args[0][0]
         assert args.name == "kytos.flow_manager.flows.delete"
         assert args.content['flow_dict']['flows'] == flows['00:01']
         assert self.napp.controller.buffers.app.put.call_count == 1
 
+        self.napp._send_flow_mods(flows, "install")
+        args = self.napp.controller.buffers.app.put.call_args[0][0]
+        assert args.name == "kytos.flow_manager.flows.install"
+        assert args.content['flow_dict']['flows'] == flows['00:01']
+        assert self.napp.controller.buffers.app.put.call_count == 2
+
     def test_update_switches_table(self):
         """Test update_switches_table"""
         sw1 = '00:00:00:00:00:00:00:01'