SIMPLE-6813 added deployed_mac_address property to Interface (#116)

Author: daniel-valent

virl2_client/models/interface.py

@@ -83,6 +83,7 @@ def __init__(
             "ipv4": None,
             "ipv6": None,
         }
+        self._deployed_mac_address = None
 
     def __eq__(self, other):
         if not isinstance(other, Interface):
@@ -252,6 +253,12 @@ def discovered_ipv6(self) -> str | None:
         self.node.sync_l3_addresses_if_outdated()
         return self._ip_snooped_info["ipv6"]
 
+    @property
+    def deployed_mac_address(self) -> str | None:
+        """Return the deployed MAC address of the interface."""
+        self.node.sync_interface_operational()
+        return self._deployed_mac_address
+
     @property
     def is_physical(self) -> bool:
         """

virl2_client/models/node.py

@@ -56,6 +56,7 @@ class Node:
         "vnc_key": "{lab}/nodes/{id}/keys/vnc",
         "layer3_addresses": "{lab}/nodes/{id}/layer3_addresses",
         "operational": "{lab}/nodes/{id}?operational=true&exclude_configurations=true",
+        "inteface_operational": "{lab}/nodes/{id}/interfaces?data=true&operational=true",
     }
 
     def __init__(
@@ -130,6 +131,7 @@ def __init__(
         self._pinned_compute_id = pinned_compute_id
         self._stale = False
         self._last_sync_l3_address_time = 0.0
+        self._last_sync_interface_operational_time = 0.0
         self._parameters = parameters
 
         self.statistics: dict[str, int | float] = {
@@ -195,6 +197,17 @@ def sync_l3_addresses_if_outdated(self) -> None:
         ):
             self.sync_layer3_addresses()
 
+    @check_stale
+    @locked
+    def sync_interface_operational_if_outdated(self) -> None:
+        timestamp = time.time()
+        if (
+            self.lab.auto_sync
+            and timestamp - self._last_sync_interface_operational_time
+            > self.lab.auto_sync_interval
+        ):
+            self.sync_interface_operational()
+
     @property
     @locked
     def state(self) -> str | None:
@@ -835,24 +848,6 @@ def sync_layer3_addresses(self) -> None:
         interfaces = result.get("interfaces", {})
         self.map_l3_addresses_to_interfaces(interfaces)
 
-    @check_stale
-    @locked
-    def sync_operational(self, response: dict[str, Any] = None):
-        """
-        Synchronize the operational state of the node.
-
-        :param response: If the operational data was fetched from the server elsewhere,
-            it can be passed here to save an API call. Will be fetched automatically
-            otherwise.
-        """
-        if response is None:
-            url = self._url_for("operational")
-            response = self._session.get(url).json()
-        self._pinned_compute_id = response.get("pinned_compute_id")
-        operational = response.get("operational", {})
-        self._compute_id = operational.get("compute_id")
-        self._resource_pool = operational.get("resource_pool")
-
     @check_stale
     @locked
     def map_l3_addresses_to_interfaces(
@@ -879,6 +874,37 @@ def map_l3_addresses_to_interfaces(
             }
         self._last_sync_l3_address_time = time.time()
 
+    @check_stale
+    @locked
+    def sync_operational(self, response: dict[str, Any] = None):
+        """
+        Synchronize the operational state of the node.
+
+        :param response: If the operational data was fetched from the server elsewhere,
+            it can be passed here to save an API call. Will be fetched automatically
+            otherwise.
+        """
+        if response is None:
+            url = self._url_for("operational")
+            response = self._session.get(url).json()
+        self._pinned_compute_id = response.get("pinned_compute_id")
+        operational = response.get("operational", {})
+        self._compute_id = operational.get("compute_id")
+        self._resource_pool = operational.get("resource_pool")
+
+    @check_stale
+    @locked
+    def sync_interface_operational(self):
+        """Synchronize the operational state of the node's interfaces."""
+        url = self._url_for("inteface_operational")
+        response = self._session.get(url).json()
+        self.lab.sync_topology_if_outdated()
+        for interface_data in response:
+            interface = self.lab._interfaces[interface_data["id"]]
+            operational = interface_data.get("operational", {})
+            interface._deployed_mac_address = operational.get("mac_address")
+        self._last_sync_interface_operational_time = time.time()
+
     def update(
         self,
         node_data: dict[str, Any],