Add hostname collision check in DeviceInitApi.arg_check

src/cnaas_nms/api/device.py

@@ -527,9 +527,20 @@ def post(self, device_id: int):
         # Set default delay for init_jobs
         init_delay_seconds = 1
 
-        # If device init is already in progress, reschedule a new step2 (connectivity check)
-        # instead of trying to restart initialization
         with sqla_session() as session:  # type: ignore
+            # Check for duplicate hostname only when not replacing device
+            if not job_kwargs.get("replace_hostname") and (
+                used_dev := session.query(Device)
+                .filter(Device.hostname == job_kwargs.get("new_hostname", ""))
+                .one_or_none()
+            ):
+                return empty_result(
+                    status="error",
+                    data=f"Hostname {job_kwargs['new_hostname']} is already used for device with id: {used_dev.id}",
+                ), 400
+
+            # If device init is already in progress, reschedule a new step2 (connectivity check)
+            # instead of trying to restart initialization
             dev: Optional[Device] = session.query(Device).filter(Device.id == device_id).one_or_none()
             if (
                 dev
@@ -638,7 +649,7 @@ def arg_check(cls, device_id: int, json_data: dict) -> dict:
                 parsed_args["neighbors"] = json_data["neighbors"]
             else:
                 raise ValueError(
-                    "Neighbors must be specified as either a list of hostnames,an empty list, or not specified at all"
+                    "Neighbors must be specified as either a list of hostnames, an empty list, or not specified at all"
                 )
         else:
             parsed_args["neighbors"] = None
@@ -672,6 +683,17 @@ def post(self, device_id: int):
             return empty_result(status="error", data="Error parsing arguments: {}".format(e)), 400
 
         with sqla_session() as session:  # type: ignore
+            # Check for duplicate hostname only when not replacing device
+            if not parsed_args.get("replace_hostname") and (
+                used_dev := session.query(Device)
+                .filter(Device.hostname == parsed_args.get("new_hostname", ""))
+                .one_or_none()
+            ):
+                return empty_result(
+                    status="error",
+                    data=f"Hostname {parsed_args['new_hostname']} is already used for device with id: {used_dev.id}",
+                ), 400
+
             try:
                 dev: Device = cnaas_nms.devicehandler.init_device.pre_init_checks(session, device_id)
                 linknets_all = dev.get_linknets_as_dict(session)

src/cnaas_nms/api/tests/test_device.py

@@ -46,6 +46,7 @@ def cleandb(self):
                 "test-dist2",
                 "access-old-name",
                 "access-new-name",
+                "mac-000000000000",
                 "discovered_device",
             ]:
                 device = session.query(Device).filter(Device.hostname == hostname).one_or_none()
@@ -505,6 +506,35 @@ def test_initcheck_distdevice(self):
         #        self.assertEqual(json_data['data']['compatible'], False)
         self.assertEqual(json_data["status"], "error")
 
+    def test_init_access_hostname_collision(self):
+        """
+        Test that the api returns an error when trying to
+        init a device with the same hostname as another switch
+        """
+        with sqla_session() as session:  # type: ignore
+            device = Device(
+                hostname="mac-000000000000",
+                platform="eos",
+                dhcp_ip=IPv4Address("10.0.1.22"),  # noqa: S1313
+                state=DeviceState.DISCOVERED,
+                device_type=DeviceType.UNKNOWN,
+            )
+            session.add(device)
+            session.commit()
+            session.refresh(device)
+            device_id = device.id
+
+        # Check for both device_initcheck and device_init that the error shows up.
+        for endpoint in ["device_initcheck", "device_init"]:
+            device_data = {"hostname": "eosaccess", "device_type": "ACCESS"}
+            result = self.client.post(f"/api/v1.0/{endpoint}/{device_id}", json=device_data)
+
+            json_data = result.json
+            self.assertIsNotNone(json_data)
+
+            self.assertEqual(result.status_code, 400)
+            self.assertIn("Hostname eosaccess is already used for device with id:", result.json.get("message"))
+
     def test_get_stackmembers_invalid_device(self):
         result = self.client.get(f"/api/v1.0/device/{'nonexisting'}/stackmember")
         json_data = json.loads(result.data.decode())