SIMPLE-6615 custom mac addresses implementation (#106)

* SIMPLE-6615 custom mac address
* Backwards compatibility for older schemas

---------

Co-authored-by: Tomas Mikuska <tmikuska@cisco.com>

virl2_client/event_handling.py

@@ -310,13 +310,11 @@ def _handle_element_modified(self, event: Event) -> None:
  )
 
  elif event.element_type == "interface":
- # it seems only port change info arrives here,
- # which the client doesn't use, so this message can be discarded
- pass
+ event.element._update(event.data, push_to_server=False)
 
  elif event.element_type == "link":
- # same as above, only sends link_capture_key which is not used
- # by the client, so we discard the message
+ # only sends link_capture_key which is not used by the client,
+ # so we discard the message
  pass
 
  else:

virl2_client/models/interface.py

@@ -22,9 +22,9 @@
 
 import logging
 import warnings
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
-from ..utils import check_stale, get_url_from_template
+from ..utils import check_stale, get_url_from_template, locked
 from ..utils import property_s as property
 
 if TYPE_CHECKING:
@@ -51,6 +51,7 @@ def __init__(
  label: str,
  slot: int | None,
  iface_type: str = "physical",
+ mac_address: str | None = None,
  ) -> None:
  """
  A CML 2 network interface, part of a node.
@@ -60,12 +61,14 @@ def __init__(
  :param label: The label of the interface.
  :param slot: The slot of the interface.
  :param iface_type: The type of the interface.
+ :param mac_address: The MAC address of the interface.
  """
  self._id = iid
  self._node = node
  self._type = iface_type
  self._label = label
  self._slot = slot
+ self._mac_address = mac_address
  self._state: str | None = None
  self._session: httpx.Client = node.lab._session
  self._stale = False
@@ -144,6 +147,20 @@ def physical(self) -> bool:
  """Check if the interface is physical."""
  return self.type == "physical"
 
+ @property
+ def mac_address(self) -> str | None:
+ """Return the MAC address set to the interface.
+ This is the address that will be used when the device is started."""
+ self.node.lab.sync_topology_if_outdated()
+ return self._mac_address
+
+ @mac_address.setter
+ @locked
+ def mac_address(self, value: str | None) -> None:
+ """Set the MAC address of the node to the given value."""
+ self._set_interface_property("mac_address", value)
+ self._mac_address = value
+
  @property
  def connected(self) -> bool:
  """Check if the interface is connected to a link."""
@@ -385,3 +402,43 @@ def is_connected(self):
  DeprecationWarning,
  )
  return self.connected
+
+ @check_stale
+ @locked
+ def _update(
+ self,
+ interface_data: dict[str, Any],
+ push_to_server: bool = True,
+ ) -> None:
+ """
+ Update the interface_data with the provided data.
+
+ :param interface_data: The data to update the interface with.
+ :param push_to_server: Whether to push the changes to the server.
+ """
+ if push_to_server:
+ self._set_interface_properties(interface_data)
+ if "data" in interface_data:
+ interface_data = interface_data["data"]
+ for key, value in interface_data.items():
+ setattr(self, f"_{key}", value)
+
+ def _set_interface_property(self, key: str, val: Any) -> None:
+ """
+ Set a property of the interface.
+
+ :param key: The key of the property to set.
+ :param val: The value to set.
+ """
+ _LOGGER.debug(f"Setting node property {self} {key}: {val}")
+ self._set_interface_properties({key: val})
+
+ @check_stale
+ def _set_interface_properties(self, interface_data: dict[str, Any]) -> None:
+ """
+ Set multiple properties of the interface.
+
+ :param node_data: A dictionary containing the properties to set.
+ """
+ url = self._url_for("interface")
+ self._session.patch(url, json=interface_data)

virl2_client/models/lab.py

@@ -935,17 +935,19 @@ def _create_interface_local(
  node: Node,
  slot: int | None,
  iface_type: str = "physical",
+ mac_address: str | None = None,
  ) -> Interface:
  """Helper function to create an interface in the client library."""
  if iface_id not in self._interfaces:
- iface = Interface(iface_id, node, label, slot, iface_type)
+ iface = Interface(iface_id, node, label, slot, iface_type, mac_address)
  self._interfaces[iface_id] = iface
  else: # update the interface if it already exists:
  iface = self._interfaces[iface_id]
  iface._node = node
  iface._label = label
  iface._slot = slot
  iface._type = iface_type
+ iface._mac_address = mac_address
  return iface
 
  @check_stale
@@ -954,7 +956,8 @@ def create_annotation(self, annotation_type: str, **kwargs) -> AnnotationType:
  """
  Create a lab annotation.
 
- :param type: Type of the annotation (rectangle, ellipse, line or text).
+ :param annotation_type: Type of the annotation (rectangle, ellipse, line or
+ text).
  :returns: The created annotation.
  """
  url = self._url_for("annotations")
@@ -1465,8 +1468,11 @@ def _import_interface(
  label = iface_data["label"]
  slot = iface_data.get("slot")
  iface_type = iface_data["type"]
+ mac_address = iface_data.get("mac_address")
  node = self._nodes[node_id]
- return self._create_interface_local(iface_id, label, node, slot, iface_type)
+ return self._create_interface_local(
+ iface_id, label, node, slot, iface_type, mac_address
+ )
 
  @locked
  def _import_node(self, node_id: str, node_data: dict) -> Node:
@@ -1579,11 +1585,12 @@ def update_lab(self, topology: dict, exclude_configurations: bool) -> None:
  # kept elements
  kept_nodes = update_node_ids.intersection(existing_node_ids)
  # kept_links = update_link_ids.intersection(existing_link_ids)
- # kept_interfaces = update_interface_ids.intersection(existing_interface_ids)
+ kept_interfaces = update_interface_ids.intersection(existing_interface_ids)
  kept_annotations = update_annotation_ids.intersection(existing_annotation_ids)
  self._update_elements(
  topology=topology,
  kept_nodes=kept_nodes,
+ kept_interfaces=kept_interfaces,
  kept_annotations=kept_annotations,
  exclude_configurations=exclude_configurations,
  )
@@ -1721,6 +1728,7 @@ def _update_elements(
  self,
  topology: dict,
  kept_nodes: Iterable[str] | None = None,
+ kept_interfaces: Iterable[str] | None = None,
  kept_annotations: Iterable[str] | None = None,
  exclude_configurations: bool = False,
  ) -> None:
@@ -1738,13 +1746,20 @@ def _update_elements(
  lab_node = self._nodes[node_id]
  lab_node._update(node, exclude_configurations, push_to_server=False)
 
- # For now, can't update interface data server-side, this will change with tags
- # for interface_id in kept_interfaces:
- # interface_data = self._find_interface_in_topology(interface_id, topology)
+ if kept_interfaces:
+ for interface_id in kept_interfaces:
+ interface_data = self._find_interface_in_topology(
+ interface_id, topology
+ )
+ interface = self._interfaces[interface_id]
+ interface._update(interface_data, push_to_server=False)
 
  # For now, can't update link data server-side, this will change with tags
- # for link_id in kept_links:
- # link_data = self._find_link_in_topology(link_id, topology)
+ # if kept_links:
+ # for link_id in kept_links:
+ # link_data = self._find_link_in_topology(link_id, topology)
+ # link = self._links[link_id]
+ # link._update(link_data, push_to_server=False)
 
  if kept_annotations:
  for ann_id in kept_annotations:
@@ -1782,14 +1797,27 @@ def _find_link_in_topology(link_id: str, topology: dict) -> dict:
  # if it cannot be found, it is an internal structure error
  raise LinkNotFound
 
- # @staticmethod
- # def _find_interface_in_topology(interface_id: str, topology: dict) -> dict:
- # for node in topology["nodes"]:
- # for interface in node["interfaces"]:
- # if interface["id"] == interface_id:
- # return interface
- # # if it cannot be found, it is an internal structure error
- # raise InterfaceNotFound
+ @staticmethod
+ def _find_interface_in_topology(interface_id: str, topology: dict) -> dict:
+ """
+ Find an interface in the given topology.
+
+ :param interface_id: The ID of the interface to find.
+ :param topology: Dictionary containing the lab topology.
+ :returns: The interface with the specified ID.
+ :raises InterfaceNotFound: If the interface cannot be found in the topology.
+ """
+ if "interfaces" in topology:
+ for interface in topology["interfaces"]:
+ if interface["id"] == interface_id:
+ return interface
+ else:
+ for node in topology["nodes"]:
+ for interface in node["interfaces"]:
+ if interface["id"] == interface_id:
+ return interface
+ # if it cannot be found, it is an internal structure error
+ raise InterfaceNotFound
 
  @staticmethod
  def _find_node_in_topology(node_id: str, topology: dict) -> dict:

virl2_client/utils.py

@@ -144,7 +144,7 @@ def __get__(self, instance, owner):
 def locked(func: TCallable) -> TCallable:
  """
  A decorator that makes a method threadsafe.
- Parent class instance must have a `session.lock` property for locking to occur.
+ Parent class instance must have a `_session.lock` property for locking to occur.
  """
 
  @wraps(func)