SIMPLE-6506 removed/deprecated offline mode leftovers (#97)

tests/test_client_library.py

@@ -77,6 +77,7 @@ def test_import_lab_from_path_virl(
 
  cl._session.post.assert_called_once_with(
  "import/virl-1x",
+ params=None,
  content="<?xml version='1.0' encoding='UTF-8'?>",
  )
  cl._session.post.assert_called_once()

virl2_client/event_handling.py

@@ -305,7 +305,7 @@ def _handle_element_created(self, event: Event) -> None:
 
  def _handle_element_modified(self, event: Event) -> None:
  if event.element_type == "node":
- event.element.update(
+ event.element._update(
  event.data, exclude_configurations=False, push_to_server=False
  )
 

virl2_client/models/__init__.py

@@ -22,6 +22,7 @@
 node and image definition and helper classes for automation
 and authentication."""
 
+from .annotation import Annotation
 from .auth_management import AuthManagement
 from .authentication import TokenAuth
 from .groups import GroupManagement
@@ -48,4 +49,5 @@
  "TokenAuth",
  "ResourcePoolManagement",
  "AuthManagement",
+ "Annotation",
 )

virl2_client/models/annotation.py

@@ -24,7 +24,7 @@
 from typing import TYPE_CHECKING, Any, Literal
 
 from ..exceptions import InvalidProperty
-from ..utils import check_stale, get_url_from_template, locked
+from ..utils import _deprecated_argument, check_stale, get_url_from_template, locked
 from ..utils import property_s as property
 
 if TYPE_CHECKING:
@@ -353,17 +353,25 @@ def _remove_on_server(self) -> None:
  url = self._url_for("annotation")
  self._session.delete(url)
 
+ def update(self, annotation_data: dict[str, Any], push_to_server=None) -> None:
+ """
+ Update annotation properties.
+
+ :param annotation_data: JSON dict with new annotation property:value pairs.
+ :param push_to_server: DEPRECATED: Was only used by internal methods
+ and should otherwise always be True.
+ """
+ _deprecated_argument(self.update, push_to_server, "push_to_server")
+ self._update(annotation_data, push_to_server=True)
+
  @check_stale
  @locked
- def update(
- self, annotation_data: dict[str, Any], push_to_server: bool = True
- ) -> None:
+ def _update(self, annotation_data: dict[str, Any], push_to_server: bool) -> None:
  """
  Update annotation properties.
 
  :param annotation_data: JSON dict with new annotation property:value pairs.
  :param push_to_server: Whether to push the changes to the server.
- Defaults to True; should only be False when used by internal methods.
  """
  if annotation_data.get("type") not in (None, self._type):
  raise ValueError("Can't change annotation type.")
@@ -424,7 +432,7 @@ def __init__(
  self._y2 = 100
  self._rotation = 0
  if annotation_data:
- self.update(annotation_data, push_to_server=False)
+ self._update(annotation_data, push_to_server=False)
 
  @property
  def border_radius(self) -> int:
@@ -499,7 +507,7 @@ def __init__(
  self._y2 = 100
  self._rotation = 0
  if annotation_data:
- self.update(annotation_data, push_to_server=False)
+ self._update(annotation_data, push_to_server=False)
 
  @property
  def x2(self) -> int:
@@ -562,7 +570,7 @@ def __init__(
  self._line_start = None
  self._line_end = None
  if annotation_data:
- self.update(annotation_data, push_to_server=False)
+ self._update(annotation_data, push_to_server=False)
 
  @property
  def x2(self) -> int:
@@ -643,7 +651,7 @@ def __init__(
  self._text_size = 12
  self._text_unit = "pt"
  if annotation_data:
- self.update(annotation_data, push_to_server=False)
+ self._update(annotation_data, push_to_server=False)
 
  @property
  def rotation(self) -> int:

virl2_client/models/lab.py

@@ -1736,7 +1736,7 @@ def _update_elements(
  for node_id in kept_nodes:
  node = self._find_node_in_topology(node_id, topology)
  lab_node = self._nodes[node_id]
- lab_node.update(node, exclude_configurations, push_to_server=False)
+ 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:
@@ -1750,7 +1750,7 @@ def _update_elements(
  for ann_id in kept_annotations:
  annotation = self._find_annotation_in_topology(ann_id, topology)
  lab_annotation = self._annotations[ann_id]
- lab_annotation.update(annotation, push_to_server=False)
+ lab_annotation._update(annotation, push_to_server=False)
 
  @locked
  def update_lab_properties(self, properties: dict[str, Any]):

virl2_client/models/node.py

@@ -28,7 +28,7 @@
 from typing import TYPE_CHECKING, Any
 
 from ..exceptions import InterfaceNotFound
-from ..utils import check_stale, get_url_from_template, locked
+from ..utils import _deprecated_argument, check_stale, get_url_from_template, locked
 from ..utils import property_s as property
 
 if TYPE_CHECKING:
@@ -879,9 +879,26 @@ def map_l3_addresses_to_interfaces(
  }
  self._last_sync_l3_address_time = time.time()
 
+ def update(
+ self,
+ node_data: dict[str, Any],
+ exclude_configurations: bool,
+ push_to_server=None,
+ ) -> None:
+ """
+ Update the node with the provided data.
+
+ :param node_data: The data to update the node with.
+ :param exclude_configurations: Whether to exclude configuration updates.
+ :param push_to_server: DEPRECATED: Was only used by internal methods
+ and should otherwise always be True.
+ """
+ _deprecated_argument(self.update, push_to_server, "push_to_server")
+ self._update(node_data, exclude_configurations, push_to_server=True)
+
  @check_stale
  @locked
- def update(
+ def _update(
  self,
  node_data: dict[str, Any],
  exclude_configurations: bool,
@@ -893,7 +910,6 @@ def update(
  :param node_data: The data to update the node with.
  :param exclude_configurations: Whether to exclude configuration updates.
  :param push_to_server: Whether to push the changes to the server.
- Defaults to True; should only be False when used by internal methods.
  """
  if push_to_server:
  self._set_node_properties(node_data)

virl2_client/models/resource_pools.py

@@ -26,7 +26,7 @@
 from typing import TYPE_CHECKING, Any, Dict, Iterable
 
 from ..exceptions import InvalidProperty
-from ..utils import get_url_from_template
+from ..utils import _deprecated_argument, get_url_from_template
 
 if TYPE_CHECKING:
  import httpx
@@ -93,7 +93,7 @@ def sync_resource_pools(self) -> None:
  pool_id = res_pool.pop("id")
  res_pool["pool_id"] = pool_id
  if pool_id in self._resource_pools:
- self._resource_pools[pool_id].update(res_pool, push_to_server=False)
+ self._resource_pools[pool_id]._update(res_pool, push_to_server=False)
  else:
  self._add_resource_pool_local(**res_pool)
  res_pool_ids.append(pool_id)
@@ -417,13 +417,23 @@ def remove(self) -> None:
  url = self._url_for("resource_pool")
  self._session.delete(url)
 
- def update(self, pool_data: dict[str, Any], push_to_server: bool = True):
+ def update(self, pool_data: dict[str, Any], push_to_server=None) -> None:
+ """
+ Update multiple properties of the pool at once.
+
+ :param pool_data: A dictionary of the properties to update.
+ :param push_to_server: DEPRECATED: Was only used by internal methods
+ and should otherwise always be True.
+ """
+ _deprecated_argument(self.update, push_to_server, "push_to_server")
+ self._update(pool_data, push_to_server=True)
+
+ def _update(self, pool_data: dict[str, Any], push_to_server: bool = True) -> None:
  """
  Update multiple properties of the pool at once.
 
  :param pool_data: A dictionary of the properties to update.
  :param push_to_server: Whether to push the changes to the server.
- Defaults to True; should only be False when used by internal methods.
  """
  if push_to_server:
  self._set_resource_pool_properties(pool_data)

virl2_client/models/system.py

@@ -26,7 +26,7 @@
 
 from virl2_client.exceptions import ControllerNotFound, InvalidMacAddressBlock
 
-from ..utils import get_url_from_template
+from ..utils import _deprecated_argument, get_url_from_template
 
 if TYPE_CHECKING:
  import httpx
@@ -136,7 +136,7 @@ def maintenance_notice(self, notice: SystemNotice | None) -> None:
  else:
  notice = self._system_notices.get(resolved["id"])
  if notice is not None and resolved is not None:
- notice.update(resolved, push_to_server=False)
+ notice._update(resolved, push_to_server=False)
  self._maintenance_notice = notice
 
  def sync_compute_hosts_if_outdated(self) -> None:
@@ -167,7 +167,7 @@ def sync_compute_hosts(self) -> None:
  compute_id = compute_host.pop("id")
  compute_host["compute_id"] = compute_id
  if compute_id in self._compute_hosts:
- self._compute_hosts[compute_id].update(
+ self._compute_hosts[compute_id]._update(
  compute_host, push_to_server=False
  )
  else:
@@ -188,7 +188,7 @@ def sync_system_notices(self) -> None:
  for system_notice in system_notices:
  notice_id = system_notice.get("id")
  if notice_id in self._system_notices:
- self._system_notices[notice_id].update(
+ self._system_notices[notice_id]._update(
  system_notice, push_to_server=False
  )
  else:
@@ -512,13 +512,23 @@ def remove(self) -> None:
  url = self._url_for("compute_host")
  self._session.delete(url)
 
- def update(self, host_data: dict[str, Any], push_to_server: bool = True) -> None:
+ def update(self, host_data: dict[str, Any], push_to_server=None) -> None:
+ """
+ Update the compute host with the given data.
+
+ :param host_data: The data to update the compute host.
+ :param push_to_server: DEPRECATED: Was only used by internal methods
+ and should otherwise always be True.
+ """
+ _deprecated_argument(self.update, push_to_server, "push_to_server")
+ self._update(host_data, push_to_server=True)
+
+ def _update(self, host_data: dict[str, Any], push_to_server: bool = True) -> None:
  """
  Update the compute host with the given data.
 
  :param host_data: The data to update the compute host.
  :param push_to_server: Whether to push the changes to the server.
- Defaults to True; should only be False when used by internal methods.
  """
  if push_to_server:
  self._set_compute_host_properties(host_data)
@@ -545,7 +555,7 @@ def _set_compute_host_properties(self, host_data: dict[str, Any]) -> None:
  """
  url = self._url_for("compute_host")
  new_data = self._session.patch(url, json=host_data).json()
- self.update(new_data, push_to_server=False)
+ self._update(new_data, push_to_server=False)
 
 
 class SystemNotice:
@@ -642,13 +652,23 @@ def remove(self) -> None:
  url = self._url_for("notice")
  self._session.delete(url)
 
- def update(self, notice_data: dict[str, Any], push_to_server: bool = True) -> None:
+ def update(self, notice_data: dict[str, Any], push_to_server=None) -> None:
+ """
+ Update the system notice with the given data.
+
+ :param notice_data: The data to update the system notice with.
+ :param push_to_server: DEPRECATED: Was only used by internal methods
+ and should otherwise always be True.
+ """
+ _deprecated_argument(self.update, push_to_server, "push_to_server")
+ self._update(notice_data, push_to_server=True)
+
+ def _update(self, notice_data: dict[str, Any], push_to_server: bool = True) -> None:
  """
  Update the system notice with the given data.
 
  :param notice_data: The data to update the system notice with.
  :param push_to_server: Whether to push the changes to the server.
- Defaults to True; should only be False when used by internal methods.
  """
  if push_to_server:
  self._set_notice_properties(notice_data)
@@ -675,4 +695,4 @@ def _set_notice_properties(self, notice_data: dict[str, Any]) -> None:
  """
  url = self._url_for("notice")
  new_data = self._session.patch(url, json=notice_data).json()
- self.update(new_data, push_to_server=False)
+ self._update(new_data, push_to_server=False)

virl2_client/utils.py

@@ -20,9 +20,10 @@
 
 from __future__ import annotations
 
+import warnings
 from contextlib import nullcontext
 from functools import wraps
-from typing import TYPE_CHECKING, Callable, Type, TypeVar, Union, cast
+from typing import TYPE_CHECKING, Any, Callable, Type, TypeVar, Union, cast
 
 import httpx
 
@@ -178,3 +179,19 @@ def get_url_from_template(
  values = {}
  values["CONFIG_MODE"] = _CONFIG_MODE
  return endpoint_url_template.format(**values)
+
+
+_DEPRECATION_MESSAGES = {
+ "push_to_server": "meant to be used only by internal methods",
+ "offline": "offline mode has been removed",
+}
+
+
+def _deprecated_argument(func, argument: Any, argument_name: str):
+ if argument is not None:
+ reason = _DEPRECATION_MESSAGES[argument_name]
+ warnings.warn(
+ f"{type(func.__self__).__name__}.{func.__name__}: "
+ f"The argument '{argument_name}' is deprecated. Reason: {reason}",
+ DeprecationWarning,
+ )