Merge pull request #175 from kytos-ng/epic/vlan_range

Moved `_get_tag_ranges()` to kytos/core

main.py

@@ -12,15 +12,15 @@
 
 from kytos.core import KytosEvent, KytosNApp, log, rest
 from kytos.core.common import EntityStatus
-from kytos.core.exceptions import (KytosLinkCreationError,
-                                   KytosSetTagRangeError,
-                                   KytosTagtypeNotSupported)
+from kytos.core.exceptions import (KytosInvalidTagRanges,
+                                   KytosLinkCreationError, KytosTagError)
 from kytos.core.helpers import listen_to, load_spec, now, validate_openapi
 from kytos.core.interface import Interface
 from kytos.core.link import Link
 from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
                                  content_type_json_or_415, get_json_or_400)
 from kytos.core.switch import Switch
+from kytos.core.tag_ranges import get_tag_ranges
 from napps.kytos.topology import settings
 
 from .controllers import TopoController
@@ -486,75 +486,26 @@ def delete_interface_metadata(self, request: Request) -> JSONResponse:
         self.notify_metadata_changes(interface, 'removed')
         return JSONResponse("Operation successful")
 
-    @staticmethod
-    def map_singular_values(tag_range):
-        """Change integer or singular interger list to
-        list[int, int] when necessary"""
-        if isinstance(tag_range, int):
-            tag_range = [tag_range] * 2
-        elif len(tag_range) == 1:
-            tag_range = [tag_range[0]] * 2
-        return tag_range
-
-    def _get_tag_ranges(self, content: dict):
-        """Get tag_ranges and check validity:
-        - It should be ordered
-        - Not unnecessary partition (eg. [[10,20],[20,30]])
-        - Singular intergers are changed to ranges (eg. [10] to [[10, 10]])
-        The ranges are understood as [inclusive, inclusive]"""
-        ranges = content["tag_ranges"]
-        if len(ranges) < 1:
-            detail = "tag_ranges is empty"
-            raise HTTPException(400, detail=detail)
-        last_tag = 0
-        ranges_n = len(ranges)
-        for i in range(0, ranges_n):
-            ranges[i] = self.map_singular_values(ranges[i])
-            if ranges[i][0] > ranges[i][1]:
-                detail = f"The range {ranges[i]} is not ordered"
-                raise HTTPException(400, detail=detail)
-            if last_tag and last_tag > ranges[i][0]:
-                detail = f"tag_ranges is not ordered. {last_tag}"\
-                         f" is higher than {ranges[i][0]}"
-                raise HTTPException(400, detail=detail)
-            if last_tag and last_tag == ranges[i][0] - 1:
-                detail = f"tag_ranges has an unnecessary partition. "\
-                         f"{last_tag} is before to {ranges[i][0]}"
-                raise HTTPException(400, detail=detail)
-            if last_tag and last_tag == ranges[i][0]:
-                detail = f"tag_ranges has repetition. {ranges[i-1]}"\
-                         f" have same values as {ranges[i]}"
-                raise HTTPException(400, detail=detail)
-            last_tag = ranges[i][1]
-        if ranges[-1][1] > 4095:
-            detail = "Maximum value for tag_ranges is 4095"
-            raise HTTPException(400, detail=detail)
-        if ranges[0][0] < 1:
-            detail = "Minimum value for tag_ranges is 1"
-            raise HTTPException(400, detail=detail)
-        return ranges
-
     @rest('v3/interfaces/{interface_id}/tag_ranges', methods=['POST'])
     @validate_openapi(spec)
     def set_tag_range(self, request: Request) -> JSONResponse:
         """Set tag range"""
         content_type_json_or_415(request)
         content = get_json_or_400(request, self.controller.loop)
         tag_type = content.get("tag_type")
-        ranges = self._get_tag_ranges(content)
+        try:
+            ranges = get_tag_ranges(content["tag_ranges"])
+        except KytosInvalidTagRanges as err:
+            raise HTTPException(400, detail=str(err))
         interface_id = request.path_params["interface_id"]
         interface = self.controller.get_interface_by_id(interface_id)
         if not interface:
             raise HTTPException(404, detail="Interface not found")
         try:
             interface.set_tag_ranges(ranges, tag_type)
             self.handle_on_interface_tags(interface)
-        except KytosSetTagRangeError as err:
-            detail = f"The new tag_ranges cannot be applied {err}"
-            raise HTTPException(400, detail=detail)
-        except KytosTagtypeNotSupported as err:
-            detail = f"Error with tag_type. {err}"
-            raise HTTPException(400, detail=detail)
+        except KytosTagError as err:
+            raise HTTPException(400, detail=str(err))
         return JSONResponse("Operation Successful", status_code=200)
 
     @rest('v3/interfaces/{interface_id}/tag_ranges', methods=['DELETE'])
@@ -570,9 +521,8 @@ def delete_tag_range(self, request: Request) -> JSONResponse:
         try:
             interface.remove_tag_ranges(tag_type)
             self.handle_on_interface_tags(interface)
-        except KytosTagtypeNotSupported as err:
-            detail = f"Error with tag_type. {err}"
-            raise HTTPException(400, detail=detail)
+        except KytosTagError as err:
+            raise HTTPException(400, detail=str(err))
         return JSONResponse("Operation Successful", status_code=200)
 
     @rest('v3/interfaces/tag_ranges', methods=['GET'])

tests/unit/test_main.py

@@ -13,7 +13,6 @@
                                    KytosTagtypeNotSupported)
 from kytos.core.interface import Interface
 from kytos.core.link import Link
-from kytos.core.rest_api import HTTPException
 from kytos.core.switch import Switch
 from kytos.lib.helpers import (get_interface_mock, get_link_mock,
                                get_controller_mock, get_switch_mock,
@@ -1630,57 +1629,6 @@ def test_interruption_end(
         assert mock_notify_link_status_change.call_count == 2
         mock_notify_topology_update.assert_called_once()
 
-    def test_map_singular_values(self):
-        """Test map_singular_values"""
-        mock_tag = 201
-        result = self.napp.map_singular_values(mock_tag)
-        assert result == [201, 201]
-
-        mock_tag = [201]
-        result = self.napp.map_singular_values(mock_tag)
-        assert result == [201, 201]
-
-    def test_get_tag_ranges(self):
-        """Test _get_tag_ranges"""
-        mock_content = {'tag_ranges': [100, [150], [200, 3000]]}
-        result = self.napp._get_tag_ranges(mock_content)
-        assert result == [[100, 100], [150, 150], [200, 3000]]
-
-        # Empty
-        mock_content = {'tag_ranges': []}
-        with pytest.raises(HTTPException):
-            self.napp._get_tag_ranges(mock_content)
-
-        # Range not ordered
-        mock_content = {'tag_ranges': [[20, 19]]}
-        with pytest.raises(HTTPException):
-            self.napp._get_tag_ranges(mock_content)
-
-        # Ranges not ordered
-        mock_content = {'tag_ranges': [[20, 50], [30, 3000]]}
-        with pytest.raises(HTTPException):
-            self.napp._get_tag_ranges(mock_content)
-
-        # Unnecessary partition
-        mock_content = {'tag_ranges': [[20, 50], [51, 3000]]}
-        with pytest.raises(HTTPException):
-            self.napp._get_tag_ranges(mock_content)
-
-        # Repeated tag
-        mock_content = {'tag_ranges': [[20, 50], [50, 3000]]}
-        with pytest.raises(HTTPException):
-            self.napp._get_tag_ranges(mock_content)
-
-        # Over 4095
-        mock_content = {'tag_ranges': [[20, 50], [50, 4096]]}
-        with pytest.raises(HTTPException):
-            self.napp._get_tag_ranges(mock_content)
-
-        # Under 1
-        mock_content = {'tag_ranges': [[0, 50], [50, 3000]]}
-        with pytest.raises(HTTPException):
-            self.napp._get_tag_ranges(mock_content)
-
     async def test_set_tag_range(self, event_loop):
         """Test set_tag_range"""
         self.napp.controller.loop = event_loop
@@ -1727,7 +1675,7 @@ async def test_set_tag_range_tag_error(self, event_loop):
         mock_switch = get_switch_mock(dpid)
         mock_interface = get_interface_mock('s1-eth1', 1, mock_switch)
         mock_interface.set_tag_ranges = MagicMock()
-        mock_interface.set_tag_ranges.side_effect = KytosSetTagRangeError()
+        mock_interface.set_tag_ranges.side_effect = KytosSetTagRangeError("")
         mock_interface.notify_interface_tags = MagicMock()
         self.napp.controller.get_interface_by_id = MagicMock()
         self.napp.controller.get_interface_by_id.return_value = mock_interface
@@ -1748,7 +1696,9 @@ async def test_set_tag_range_type_error(self, event_loop):
         mock_switch = get_switch_mock(dpid)
         mock_interface = get_interface_mock('s1-eth1', 1, mock_switch)
         mock_interface.set_tag_ranges = MagicMock()
-        mock_interface.set_tag_ranges.side_effect = KytosTagtypeNotSupported()
+        mock_interface.set_tag_ranges.side_effect = KytosTagtypeNotSupported(
+            ""
+        )
         self.napp.handle_on_interface_tags = MagicMock()
         self.napp.controller.get_interface_by_id = MagicMock()
         self.napp.controller.get_interface_by_id.return_value = mock_interface
@@ -1801,7 +1751,7 @@ async def test_delete_tag_range_type_error(self, event_loop):
         mock_interface = get_interface_mock('s1-eth1', 1, mock_switch)
         mock_interface.remove_tag_ranges = MagicMock()
         remove_tag = mock_interface.remove_tag_ranges
-        remove_tag.side_effect = KytosTagtypeNotSupported()
+        remove_tag.side_effect = KytosTagtypeNotSupported("")
         self.napp.controller.get_interface_by_id = MagicMock()
         self.napp.controller.get_interface_by_id.return_value = mock_interface
         url = f"{self.base_endpoint}/interfaces/{interface_id}/tag_ranges"