Not allowing s_vlan from DB for some cases

kytos-ng · Nov 4, 2024 · d460074 · d460074
1 parent dcc4d8b
commit d460074
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 5 deletions.
diff --git a/main.py b/main.py
@@ -1057,15 +1057,15 @@ def _evc_dict_with_instances(self, evc_dict):
             #     backup_links_cache
             if "links" in attribute:
                 data[attribute] = [
-                    self._link_from_dict(link) for link in value
+                    self._link_from_dict(link, attribute) for link in value
                 ]
 
             # Ex: current_path,
             #     primary_path,
             #     backup_path
             if "path" in attribute and attribute != "dynamic_backup_path":
                 data[attribute] = Path(
-                    [self._link_from_dict(link) for link in value]
+                    [self._link_from_dict(link, attribute) for link in value]
                 )
 
         return data
@@ -1105,7 +1105,7 @@ def _uni_from_dict(self, uni_dict):
         uni = UNI(interface, tag)
         return uni
 
-    def _link_from_dict(self, link_dict) -> Link:
+    def _link_from_dict(self, link_dict: dict, attribute: str) -> Link:
         """Return a Link object from python dict."""
         id_a = link_dict.get("endpoint_a").get("id")
         id_b = link_dict.get("endpoint_b").get("id")
@@ -1120,7 +1120,8 @@ def _link_from_dict(self, link_dict) -> Link:
             raise ValueError(error_msg)
 
         link = Link(endpoint_a, endpoint_b)
-        if "metadata" in link_dict:
+        allowed_paths = {"current_path", "failover_path"}
+        if "metadata" in link_dict and attribute in allowed_paths:
             link.extend_metadata(link_dict.get("metadata"))
 
         s_vlan = link.get_metadata("s_vlan")

diff --git a/tests/unit/test_main.py b/tests/unit/test_main.py
@@ -1669,7 +1669,27 @@ def test_link_from_dict_non_existent_intf(self):
             "endpoint_b": {"id": "b"}
         }
         with pytest.raises(ValueError):
-            self.napp._link_from_dict(link_dict)
+            self.napp._link_from_dict(link_dict, "current_path")
+
+    def test_link_from_dict_vlan_metadata(self):
+        """Test that link_from_dict only accepts vlans for current_path
+         and failover_path."""
+        intf = MagicMock(id="01:1")
+        self.napp.controller.get_interface_by_id = MagicMock(return_value=intf)
+        link_dict = {
+            'id': 'mock_link',
+            'endpoint_a': {'id': '00:00:00:00:00:00:00:01:4'},
+            'endpoint_b': {'id': '00:00:00:00:00:00:00:05:2'},
+            'metadata': {'s_vlan': {'tag_type': 'vlan', 'value': 1}}
+        }
+        link = self.napp._link_from_dict(link_dict, "current_path")
+        assert link.metadata.get('s_vlan', None)
+
+        link = self.napp._link_from_dict(link_dict, "failover_path")
+        assert link.metadata.get('s_vlan', None)
+
+        link = self.napp._link_from_dict(link_dict, "primary_path")
+        assert link.metadata.get('s_vlan', None) is None
 
     def test_uni_from_dict_non_existent_intf(self):
         """Test _link_from_dict non existent intf."""