CMLDEV-655 Refactored opt-in to enum (#178)

Added telemetry support

Author: tmikuska

virl2_client/models/system.py

@@ -27,7 +27,7 @@
 
 from virl2_client.exceptions import ControllerNotFound
 
-from ..utils import _deprecated_argument, get_url_from_template
+from ..utils import OptInStatus, _deprecated_argument, get_url_from_template
 
 if TYPE_CHECKING:
     import httpx
@@ -43,6 +43,8 @@ class SystemManagement:
         "external_connectors": "system/external_connectors",
         "web_session_timeout": "web_session_timeout",
         "host_configuration": "system/compute_hosts/configuration",
+        "telemetry": "telemetry",
+        "telemetry_events": "telemetry/events",
     }
 
     def __init__(
@@ -139,6 +141,28 @@ def maintenance_notice(self, notice: SystemNotice | None) -> None:
             notice._update(resolved, push_to_server=False)
         self._maintenance_notice = notice
 
+    @property
+    def telemetry(self) -> dict[str, OptInStatus | str]:
+        """Return the telemetry state."""
+        url = self._url_for("telemetry")
+        return self._session.get(url).json()
+
+    @property
+    def telemetry_state(self) -> OptInStatus:
+        """Return the telemetry state."""
+        return OptInStatus(self.telemetry["opt_in"])
+
+    @telemetry_state.setter
+    def telemetry_state(self, mode: OptInStatus) -> None:
+        """Set the telemetry state."""
+        url = self._url_for("telemetry")
+        self._session.put(url, json={"opt_in": mode.name})
+
+    def get_telemetry_events(self) -> list[dict[str, Any]]:
+        """Return the list of telemetry events."""
+        url = self._url_for("telemetry_events")
+        return self._session.get(url).json()
+
     def sync_compute_hosts_if_outdated(self) -> None:
         """Synchronize compute hosts if they are outdated."""
         timestamp = time.time()

virl2_client/models/user.py

@@ -20,9 +20,10 @@
 
 from __future__ import annotations
 
+import warnings
 from typing import TYPE_CHECKING, Any
 
-from ..utils import UNCHANGED, _Sentinel, get_url_from_template
+from ..utils import UNCHANGED, OptInStatus, _Sentinel, get_url_from_template
 
 if TYPE_CHECKING:
     import httpx
@@ -142,9 +143,9 @@ def _prepare_body(
         password_dict: dict[str, str] | None = None,
         pubkey: str | None = None,
         resource_pool: str | None | _Sentinel = UNCHANGED,
-        opt_in: bool | None | _Sentinel = UNCHANGED,
+        opt_in: OptInStatus | bool | None | _Sentinel = UNCHANGED,
         tour_version: str | None = None,
-    ) -> dict[str, Any]:
+    ) -> None:
         optional_data = {
             "fullname": fullname,
             "description": description,
@@ -156,16 +157,26 @@ def _prepare_body(
             "pubkey": pubkey,
             "tour_version": tour_version,
         }
-        sentinel_data = {
-            "resource_pool": resource_pool,
-            "opt_in": opt_in,
-        }
         for key, value in optional_data.items():
             if value is not None:
                 data[key] = value
-        for key, value in sentinel_data.items():
-            if value != UNCHANGED:
-                data[key] = value
+        if resource_pool is not UNCHANGED:
+            data["resource_pool"] = resource_pool
+        if opt_in is UNCHANGED:
+            return
+        if not isinstance(opt_in, OptInStatus):
+            warnings.warn(
+                "Using boolean or None values for opt_in is deprecated. "
+                f"Use one of {[status.value for status in OptInStatus]} instead.",
+                DeprecationWarning,
+            )
+            if opt_in is True:
+                opt_in = OptInStatus.ACCEPTED
+            elif opt_in is False:
+                opt_in = OptInStatus.DECLINED
+            else:
+                opt_in = OptInStatus.UNSET
+        data["opt_in"] = opt_in.value
 
     def user_groups(self, user_id: str) -> list[str]:
         """

virl2_client/utils.py

@@ -23,6 +23,7 @@
 import warnings
 from collections.abc import Callable
 from contextlib import nullcontext
+from enum import Enum
 from functools import wraps
 from typing import TYPE_CHECKING, Any, Type, TypeVar, cast
 
@@ -56,6 +57,12 @@ def __repr__(self):
 _CONFIG_MODE = "exclude_configurations=false"
 
 
+class OptInStatus(Enum):
+    ACCEPTED = "accepted"
+    DECLINED = "declined"
+    UNSET = "unset"
+
+
 def _make_not_found(instance: Element) -> ElementNotFound:
     """Composes and raises an ElementNotFound error for the given instance."""
     class_name = type(instance).__name__