New ifclass "mirror" on dist devices which copies settings from

interface on mgmtdomain peer device

src/cnaas_nms/db/helper.py

@@ -18,7 +18,16 @@ def canonical_mac(mac):
     return str(na_mac)
 
 
-def find_mgmtdomain_one_device(session, device0: Device) -> Optional[Mgmtdomain]:
+def find_mgmtdomain_peer(session, device0: Device) -> Device:
+    """Find the peer device in a mgmtdomain for a given device"""
+    mgmtdomain: Mgmtdomain = find_mgmtdomain_one_device(session, device0)
+    if mgmtdomain.device_a == device0:
+        return mgmtdomain.device_b
+    else:
+        return mgmtdomain.device_a
+
+
+def find_mgmtdomain_one_device(session, device0: Device) -> Mgmtdomain:
     if device0.device_type == DeviceType.DIST:
         mgmtdomain = (
             session.query(Mgmtdomain)
@@ -38,7 +47,7 @@ def find_mgmtdomain_one_device(session, device0: Device) -> Optional[Mgmtdomain]
     return mgmtdomain
 
 
-def find_mgmtdomain_two_devices(session, device0: Device, device1: Device) -> Optional[Mgmtdomain]:
+def find_mgmtdomain_two_devices(session, device0: Device, device1: Device) -> Mgmtdomain:
     if device0.device_type != device1.device_type:
         raise ValueError(
             "Both uplink devices must be of same device type: {}, {}".format(device0.hostname, device1.hostname)
@@ -89,7 +98,7 @@ def find_mgmtdomain_two_devices(session, device0: Device, device1: Device) -> Op
     return mgmtdomain
 
 
-def find_mgmtdomain(session, hostnames: List[str]) -> Optional[Mgmtdomain]:
+def find_mgmtdomain(session, hostnames: List[str]) -> Mgmtdomain:
     """Find the corresponding management domain for a pair of
     distribution switches.
 

src/cnaas_nms/db/tests/test_mgmtdomain.py

@@ -71,6 +71,13 @@ def test_find_mgmtdomain_invalid(self):
             self.assertRaises(ValueError, cnaas_nms.db.helper.find_mgmtdomain, session, [])
             self.assertRaises(ValueError, cnaas_nms.db.helper.find_mgmtdomain, session, [1, 2, 3])
 
+    def test_find_mgmtdomain_peer_device(self):
+        with sqla_session() as session:  # type: ignore
+            d_a = session.query(Device).filter(Device.hostname == "mgmtdomaintest1").one()
+            d_b = session.query(Device).filter(Device.hostname == "mgmtdomaintest2").one()
+            found_peer: Device = cnaas_nms.db.helper.find_mgmtdomain_peer(session, d_a)
+            self.assertEqual(d_b, found_peer, "Peer device not found")
+
     def test_find_mgmtdomain_twodist(self):
         with sqla_session() as session:  # type: ignore
             mgmtdomain = cnaas_nms.db.helper.find_mgmtdomain(session, ["eosdist1", "eosdist2"])

src/cnaas_nms/devicehandler/sync_devices.py

@@ -319,6 +319,21 @@ def populate_device_vars(
                                 "peer_asn": None,
                             }
                         )
+                elif intf["ifclass"] == "mirror":
+                    # Copy interface settings from mgmtdomain peer device
+                    if_dict = {"indexnum": ifindexnum}
+                    peer_device = cnaas_nms.db.helper.find_mgmtdomain_peer(session, dev)
+                    peer_settings, _ = get_settings(peer_device.hostname, devtype, peer_device.model)
+                    if "interfaces" in peer_settings and peer_settings["interfaces"]:
+                        for peer_intf in peer_settings["interfaces"]:
+                            if peer_intf["name"] == intf["name"]:
+                                for extra_key_name, value in peer_intf.items():
+                                    if_dict[extra_key_name] = value
+                                break
+                    # Description can be set separately from mirrored interface
+                    if "description" in intf:
+                        if_dict["description"] = intf["description"]
+                    fabric_device_variables["interfaces"].append(if_dict)
                 else:
                     if_dict = {"indexnum": ifindexnum}
                     for extra_key_name, value in intf.items():