Feature.addpatchposition

requirements.txt

@@ -41,3 +41,4 @@ PyYAML==5.3.1
 pytz==2023.3.post1
 types-pytz==2024.2.0.20241003
 types-requests==2.32.0.20241016
+junos-eznc==2.7.1

src/cnaas_nms/api/interface.py

@@ -30,6 +30,10 @@
         "redundant_link": fields.Boolean(required=False, example=True),
         "tags": fields.List(fields.String(), required=False, description="List of tags", example=["tag1", "tag2"]),
         "cli_append_str": fields.String(required=False),
+        "patch_position": fields.String(
+            required=False,
+            description="Patch Position",
+        ),
     },
 )
 
@@ -94,13 +98,16 @@ def put(self, hostname):
         json_data = request.get_json()
         data = {}
         errors = []
+
         device_settings = None
         with sqla_session() as session:  # type: ignore
             dev: Optional[Device] = session.query(Device).filter(Device.hostname == hostname).one_or_none()
             if not dev:
                 return empty_result("error", "Device not found"), 404
-
             updated = False
+            intfs = session.query(Interface).filter(Interface.device == dev).all()
+            patch_positions = [intf.data["patch_position"] for intf in intfs if "patch_position" in intf.data]
+            print(patch_positions)
             if "interfaces" in json_data and isinstance(json_data["interfaces"], dict):
                 for if_name, if_dict in json_data["interfaces"].items():
                     if not isinstance(if_dict, dict):
@@ -270,6 +277,21 @@ def put(self, hostname):
                                 errors.append(
                                     "cli_append_str must be a string, got: {}".format(if_dict["data"]["cli_append_str"])
                                 )
+                        if "patch_position" in if_dict["data"]:
+                            if isinstance(if_dict["data"]["patch_position"], str):
+                                if if_dict["data"]["patch_position"] not in patch_positions:
+                                    intfdata["patch_position"] = if_dict["data"]["patch_position"]
+                                    patch_positions.append(if_dict["data"]["patch_position"])
+                                else:
+                                    errors.append(
+                                        "patch_position must be unique for the interfaces: {}".format(
+                                            if_dict["data"]["patch_position"]
+                                        )
+                                    )
+                            else:
+                                errors.append(
+                                    "patch_position must be a string, got: {}".format(if_dict["data"]["patch_position"])
+                                )
                     elif "data" in if_dict and not if_dict["data"]:
                         intfdata: None = None  # type: ignore [no-redef]
 

src/cnaas_nms/api/tests/test_interface.py

@@ -0,0 +1,141 @@
+import json
+import os
+import unittest
+from ipaddress import IPv4Address
+
+import pkg_resources
+import pytest
+import yaml
+
+from cnaas_nms.api import app
+from cnaas_nms.api.tests.app_wrapper import TestAppWrapper
+from cnaas_nms.db.device import Device, DeviceState, DeviceType
+from cnaas_nms.db.interface import Interface, InterfaceConfigType
+from cnaas_nms.db.session import sqla_session
+
+
+@pytest.mark.integration
+class InterfaceTests(unittest.TestCase):
+    @pytest.fixture(autouse=True)
+    def requirements(self, postgresql, settings_directory):
+        """Ensures the required pytest fixtures are loaded implicitly for all these tests"""
+        pass
+
+    def cleandb(self):
+        with sqla_session() as session:  # type: ignore
+            for interface_name in ["custom", "downlink"]:
+                interface = session.query(Interface).filter(Interface.name == interface_name).one_or_none()
+                if interface:
+                    session.delete(interface)
+                    session.commit()
+            for hostname in ["testdevice"]:
+                device = session.query(Device).filter(Device.hostname == hostname).one_or_none()
+                if device:
+                    session.delete(device)
+                    session.commit()
+
+    def setUp(self):
+        self.jwt_auth_token = None
+        data_dir = pkg_resources.resource_filename(__name__, "data")
+        with open(os.path.join(data_dir, "testdata.yml"), "r") as f_testdata:
+            self.testdata = yaml.safe_load(f_testdata)
+            if "jwt_auth_token" in self.testdata:
+                self.jwt_auth_token = self.testdata["jwt_auth_token"]
+        self.app = app.app
+        self.app.wsgi_app = TestAppWrapper(self.app.wsgi_app, self.jwt_auth_token)
+        self.client = self.app.test_client()
+        self.cleandb()
+        device_id, hostname = self.add_device()
+        self.device_id = device_id
+        self.device_hostname = hostname
+        self.add_interfaces(device_id)
+
+    def tearDown(self):
+        self.cleandb()
+
+    def add_device(self):
+        with sqla_session() as session:  # type: ignore
+            device = Device(
+                hostname="testdevice",
+                platform="eos",
+                management_ip=IPv4Address("10.0.1.22"),
+                state=DeviceState.MANAGED,
+                device_type=DeviceType.ACCESS,
+            )
+            session.add(device)
+            session.commit()
+            return device.id, device.hostname
+
+    def add_interfaces(self, device_id):
+        with sqla_session() as session:  # type: ignore
+            interface = Interface(
+                name="custom",
+                configtype=InterfaceConfigType.ACCESS_AUTO,
+                data={
+                    "patch_position": "3E-H12",
+                },
+                device_id=device_id,
+            )
+            interface2 = Interface(
+                name="downlink",
+                configtype=InterfaceConfigType.ACCESS_AUTO,
+                data={},
+                device_id=device_id,
+            )
+            session.add(interface)
+            session.add(interface2)
+            session.commit()
+
+    def test_get_interface(self):
+        result = self.client.get(f"/api/v1.0/device/{self.device_hostname}/interfaces")
+        self.assertEqual(result.status_code, 200)
+        json_data = json.loads(result.data.decode())
+        self.assertEqual(
+            ["custom", "downlink"],
+            [interface["name"] for interface in json_data["data"]["interfaces"]],
+        )
+
+    def test_update_interface_invalid_patch_position_not_unique(self):
+        modify_data = {
+            "interfaces": {
+                "custom": {
+                    "data": {
+                        "description": "new description",
+                    }
+                },
+                "downlink": {
+                    "data": {
+                        "patch_position": "3E-H12",
+                    }
+                },
+            }
+        }
+        result = self.client.put(f"/api/v1.0/device/{self.device_hostname}/interfaces", json=modify_data)
+        json_data = json.loads(result.data.decode())
+        self.assertEqual(result.status_code, 400)
+        self.assertEqual(json_data["status"], "error")
+
+    def test_update_interface(self):
+        modify_data = {
+            "interfaces": {
+                "custom": {
+                    "data": {
+                        "description": "test",
+                    }
+                },
+                "downlink": {
+                    "data": {
+                        "description": "test",
+                        "patch_position": "XW.H4.23",
+                    }
+                },
+            }
+        }
+        result = self.client.put(f"/api/v1.0/device/{self.device_hostname}/interfaces", json=modify_data)
+        json_data = json.loads(result.data.decode())
+        self.assertEqual(result.status_code, 200)
+        self.assertEqual(["custom", "downlink"], list(json_data["data"]["updated"].keys()))
+
+
+if __name__ == "__main__":
+    unittest.main()

src/cnaas_nms/devicehandler/sync_devices.py

@@ -271,7 +271,7 @@ def populate_device_vars(
                     ifindexnum = 0
                 if "ifclass" not in intf:
                     continue
-                extra_keys = ["aggregate_id", "enabled", "cli_append_str", "metric", "mtu", "tags"]
+                extra_keys = ["aggregate_id", "enabled", "cli_append_str", "metric", "mtu", "tags", "patch_postion"]
                 if intf["ifclass"] == "downlink":
                     data = {}
                     if intf["name"] in ifname_peer_map: