diff --git a/src/cnaas_nms/db/helper.py b/src/cnaas_nms/db/helper.py index 803dd56a..f86eec82 100644 --- a/src/cnaas_nms/db/helper.py +++ b/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. diff --git a/src/cnaas_nms/db/tests/test_mgmtdomain.py b/src/cnaas_nms/db/tests/test_mgmtdomain.py index 208706ba..e4c9406a 100644 --- a/src/cnaas_nms/db/tests/test_mgmtdomain.py +++ b/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"]) diff --git a/src/cnaas_nms/devicehandler/sync_devices.py b/src/cnaas_nms/devicehandler/sync_devices.py index 85b3ceff..b45f5f61 100644 --- a/src/cnaas_nms/devicehandler/sync_devices.py +++ b/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():