Add patch position

docker/api/Dockerfile

@@ -69,7 +69,7 @@ USER www-data
 RUN mkdir /tmp/devicecerts \
     && chmod -R u=rwX,g=,o= /tmp/devicecerts
 
-ARG BUILDBRANCH=develop
+ARG BUILDBRANCH=master
 ARG GITREPO_BASE=https://github.com/SUNET/cnaas-nms.git
 # Branch specific, don't cache and run cnaas setup script
 ARG GIT_COMMIT=unknown

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",
+        ),
     },
 )
 
@@ -101,6 +105,8 @@ def put(self, hostname):
                 return empty_result("error", "Device not found"), 404
 
             updated = False
+            interfaces = session.query(Interface).filter(Interface.device == dev).all()
+            patch_positions = [interface.data["patch_position"] for interface in interfaces if "patch_position" in interface.data]
             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):
@@ -115,12 +121,13 @@ def put(self, hostname):
                     if not intf:
                         errors.append(f"Interface {if_name} not found")
                         continue
+
+                    # init interface data
+                    intfdata_original: dict[str, Any] = {}
+                    intfdata: dict[str, Any] = {}
                     if intf.data and isinstance(intf.data, dict):
-                        intfdata_original: dict[str, Any] = dict(intf.data)
-                        intfdata: dict[str, Any] = dict(intf.data)
-                    else:
-                        intfdata_original = {}
-                        intfdata = {}
+                        intfdata_original = dict(intf.data)
+                        intfdata = dict(intf.data)
 
                     if "configtype" in if_dict and if_dict["configtype"]:
                         try:
@@ -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()