diff --git a/packages/uipath/pyproject.toml b/packages/uipath/pyproject.toml index ec813ed17..b35e51aee 100644 --- a/packages/uipath/pyproject.toml +++ b/packages/uipath/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "uipath" -version = "2.13.14" +version = "2.13.15" description = "Python SDK and CLI for UiPath Platform, enabling programmatic interaction with automation services, process management, and deployment tools." readme = { file = "README.md", content-type = "text/markdown" } requires-python = ">=3.11" diff --git a/packages/uipath/src/uipath/telemetry/__init__.py b/packages/uipath/src/uipath/telemetry/__init__.py index 92b3a354c..77b79bf1d 100644 --- a/packages/uipath/src/uipath/telemetry/__init__.py +++ b/packages/uipath/src/uipath/telemetry/__init__.py @@ -1,4 +1,7 @@ -from ._track import ( # noqa: D104 +"""UiPath telemetry tracking.""" + +from ._constants import PERIODIC_TELEMETRY_FLUSH_FEATURE_FLAG +from ._track import ( flush_events, is_telemetry_enabled, reset_event_client, @@ -8,6 +11,7 @@ ) __all__ = [ + "PERIODIC_TELEMETRY_FLUSH_FEATURE_FLAG", "track", "track_event", "is_telemetry_enabled", diff --git a/packages/uipath/src/uipath/telemetry/_constants.py b/packages/uipath/src/uipath/telemetry/_constants.py index 2583c6715..7a9104deb 100644 --- a/packages/uipath/src/uipath/telemetry/_constants.py +++ b/packages/uipath/src/uipath/telemetry/_constants.py @@ -1,5 +1,7 @@ _CONNECTION_STRING = "$CONNECTION_STRING" +PERIODIC_TELEMETRY_FLUSH_FEATURE_FLAG = "EnablePeriodicTelemetryFlush" + _APP_INSIGHTS_EVENT_MARKER_ATTRIBUTE = "APPLICATION_INSIGHTS_EVENT_MARKER_ATTRIBUTE" _OTEL_RESOURCE_ATTRIBUTES = "OTEL_RESOURCE_ATTRIBUTES" _SDK_VERSION = "SdkVersion" diff --git a/packages/uipath/src/uipath/telemetry/_track.py b/packages/uipath/src/uipath/telemetry/_track.py index efde079c4..8d09b69fc 100644 --- a/packages/uipath/src/uipath/telemetry/_track.py +++ b/packages/uipath/src/uipath/telemetry/_track.py @@ -1,14 +1,16 @@ import atexit import json import os +import threading from functools import wraps from importlib.metadata import version from logging import INFO, WARNING, LogRecord, getLogger -from typing import Any, Callable, ClassVar, Dict, Mapping, Optional, Union +from typing import Any, Callable, ClassVar, Dict, Mapping from opentelemetry.sdk._logs import LoggingHandler from opentelemetry.util.types import AnyValue +from uipath.core.feature_flags import FeatureFlags from uipath.platform.constants import ( ENV_BASE_URL, ENV_ORGANIZATION_ID, @@ -32,6 +34,7 @@ _SDK_VERSION, _TELEMETRY_CONFIG_FILE, _UNKNOWN, + PERIODIC_TELEMETRY_FLUSH_FEATURE_FLAG, ) # Try to import Application Insights client for custom events @@ -59,7 +62,7 @@ def _parse_connection_string( connection_string: str, -) -> Optional[Dict[str, str]]: +) -> Dict[str, str] | None: """Parse Azure Application Insights connection string. Args: @@ -90,6 +93,8 @@ def _parse_connection_string( _logger = getLogger(__name__) _logger.propagate = False +_PERIODIC_TELEMETRY_FLUSH_INTERVAL_SECONDS = 5.0 + def _get_connection_string() -> str | None: """Get the Application Insights connection string. @@ -230,13 +235,19 @@ class _AppInsightsEventClient: """ _initialized = False - _client: Optional[Any] = None + _client: Any | None = None _atexit_registered = False - _connection_string_provider: ClassVar[Optional[Callable[[], Optional[str]]]] = None + _connection_string_provider: ClassVar[Callable[[], str | None] | None] = None + _lifecycle_lock = threading.RLock() + _flush_lock = threading.Lock() + _flush_thread: threading.Thread | None = None + _flush_stop_event: threading.Event | None = None + _pending_events = threading.Event() + _shutdown_requested = False @staticmethod def set_connection_string_provider( - provider: Callable[[], Optional[str]], + provider: Callable[[], str | None], ) -> None: """Override how the connection string is resolved. @@ -248,59 +259,101 @@ def set_connection_string_provider( @staticmethod def _initialize() -> None: """Initialize Application Insights client for custom events.""" - if _AppInsightsEventClient._initialized: - return - - _AppInsightsEventClient._initialized = True + with _AppInsightsEventClient._lifecycle_lock: + if _AppInsightsEventClient._shutdown_requested: + return + if _AppInsightsEventClient._initialized: + return - # Suppress verbose logging from Application Insights SDK - # The SDK logs telemetry ingestion details which should not be user-facing - getLogger("applicationinsights").setLevel(WARNING) - getLogger("applicationinsights.channel").setLevel(WARNING) + _AppInsightsEventClient._initialized = True - if not _HAS_APPINSIGHTS: - return + # Suppress verbose logging from Application Insights SDK + # The SDK logs telemetry ingestion details which should not be user-facing + getLogger("applicationinsights").setLevel(WARNING) + getLogger("applicationinsights.channel").setLevel(WARNING) - if _AppInsightsEventClient._connection_string_provider: - connection_string = _AppInsightsEventClient._connection_string_provider() - else: - connection_string = _get_connection_string() - if not connection_string: - return + if not _HAS_APPINSIGHTS: + return - try: - parsed = _parse_connection_string(connection_string) - if not parsed: + if _AppInsightsEventClient._connection_string_provider: + connection_string = ( + _AppInsightsEventClient._connection_string_provider() + ) + else: + connection_string = _get_connection_string() + if not connection_string: return - instrumentation_key = parsed["InstrumentationKey"] - ingestion_endpoint = parsed.get("IngestionEndpoint") + try: + parsed = _parse_connection_string(connection_string) + if not parsed: + return - # Build custom channel: DiagnosticSender → SynchronousQueue → TelemetryChannel - if ingestion_endpoint: - endpoint_url = ingestion_endpoint.rstrip("/") + "/v2/track" - else: - endpoint_url = None # SDK default + instrumentation_key = parsed["InstrumentationKey"] + ingestion_endpoint = parsed.get("IngestionEndpoint") - sender = _DiagnosticSender(service_endpoint_uri=endpoint_url) - queue = SynchronousQueue(sender) - channel = TelemetryChannel(queue=queue) + # Build custom channel: DiagnosticSender → SynchronousQueue → TelemetryChannel + if ingestion_endpoint: + endpoint_url = ingestion_endpoint.rstrip("/") + "/v2/track" + else: + endpoint_url = None # SDK default + + sender = _DiagnosticSender(service_endpoint_uri=endpoint_url) + queue = SynchronousQueue(sender) + channel = TelemetryChannel(queue=queue) + + _AppInsightsEventClient._client = AppInsightsTelemetryClient( + instrumentation_key, telemetry_channel=channel + ) + + # Set application version + _AppInsightsEventClient._client.context.application.ver = version( + "uipath" + ) + except Exception as e: + # Log but don't raise - telemetry should never break the main application + _logger.warning( + f"Failed to initialize Application Insights client: {e}" + ) + _logger.debug( + "Application Insights initialization error", exc_info=True + ) + + @staticmethod + def _ensure_periodic_flush_worker() -> None: + """Start the feature-gated periodic flush worker once.""" + if not FeatureFlags.is_flag_enabled(PERIODIC_TELEMETRY_FLUSH_FEATURE_FLAG): + return - _AppInsightsEventClient._client = AppInsightsTelemetryClient( - instrumentation_key, telemetry_channel=channel + with _AppInsightsEventClient._lifecycle_lock: + if _AppInsightsEventClient._shutdown_requested: + return + current = _AppInsightsEventClient._flush_thread + if current and current.is_alive(): + return + + stop_event = threading.Event() + worker = threading.Thread( + target=_AppInsightsEventClient._periodic_flush_worker, + args=(stop_event,), + name="uipath-appinsights-flush", + daemon=True, ) + _AppInsightsEventClient._flush_stop_event = stop_event + _AppInsightsEventClient._flush_thread = worker + worker.start() - # Set application version - _AppInsightsEventClient._client.context.application.ver = version("uipath") - except Exception as e: - # Log but don't raise - telemetry should never break the main application - _logger.warning(f"Failed to initialize Application Insights client: {e}") - _logger.debug("Application Insights initialization error", exc_info=True) + @staticmethod + def _periodic_flush_worker(stop_event: threading.Event) -> None: + """Flush periodically until shutdown is requested.""" + while not stop_event.wait(_PERIODIC_TELEMETRY_FLUSH_INTERVAL_SECONDS): + if _AppInsightsEventClient._pending_events.is_set(): + _AppInsightsEventClient.flush() @staticmethod def track_event( name: str, - properties: Optional[Dict[str, Any]] = None, + properties: Dict[str, Any] | None = None, ) -> None: """Track a custom event to Application Insights customEvents table. @@ -310,62 +363,100 @@ def track_event( """ _AppInsightsEventClient._initialize() - if not _AppInsightsEventClient._client: - return + with _AppInsightsEventClient._lifecycle_lock: + if _AppInsightsEventClient._shutdown_requested: + return - try: - safe_properties: Dict[str, str] = {} - if properties: - for key, value in properties.items(): - if value is not None: - safe_properties[key] = str(value) - - _AppInsightsEventClient._client.track_event( - name=name, properties=safe_properties, measurements={} - ) - # Note: We don't flush after every event to avoid blocking. - # Events will be sent in batches by the SDK. - except Exception as e: - # Log but don't raise - telemetry should never break the main application - _logger.warning(f"Failed to track event '{name}': {e}") - _logger.debug(f"Event tracking error for '{name}'", exc_info=True) + if not _AppInsightsEventClient._client: + return + + _AppInsightsEventClient._ensure_periodic_flush_worker() + + try: + safe_properties: Dict[str, str] = {} + if properties: + for key, value in properties.items(): + if value is not None: + safe_properties[key] = str(value) + + client = _AppInsightsEventClient._client + if not client: + return + client.track_event( + name=name, properties=safe_properties, measurements={} + ) + _AppInsightsEventClient._pending_events.set() + # Note: We don't flush after every event to avoid blocking. + # Events are sent when the queue fills, on explicit/shutdown flush, + # or periodically when the periodic flush feature is enabled. + except Exception as e: + # Log but don't raise - telemetry should never break the main application + _logger.warning(f"Failed to track event '{name}': {e}") + _logger.debug(f"Event tracking error for '{name}'", exc_info=True) @staticmethod def flush() -> None: """Flush any pending telemetry events.""" - if _AppInsightsEventClient._client: - try: - _AppInsightsEventClient._client.flush() - # Check if items remain after flush (indicates send failure) + with _AppInsightsEventClient._flush_lock: + client = _AppInsightsEventClient._client + if client: + _AppInsightsEventClient._pending_events.clear() try: - remaining = ( - _AppInsightsEventClient._client.channel.queue._queue.qsize() - ) - if remaining > 0: - _logger.warning( - "AppInsights flush: %d items still in queue after flush", - remaining, - ) - except Exception: - pass - except Exception as e: - # Log but don't raise - telemetry should never break the main application - _logger.warning(f"Failed to flush telemetry events: {e}") - _logger.debug("Telemetry flush error", exc_info=True) + client.flush() + # Check if items remain after flush (indicates send failure) + try: + remaining = client.channel.queue._queue.qsize() + if remaining > 0: + _AppInsightsEventClient._pending_events.set() + _logger.warning( + "AppInsights flush: %d items still in queue after flush", + remaining, + ) + except Exception: + pass + except Exception as e: + _AppInsightsEventClient._pending_events.set() + # Log but don't raise - telemetry should never break the main application + _logger.warning(f"Failed to flush telemetry events: {e}") + _logger.debug("Telemetry flush error", exc_info=True) + + @staticmethod + def _shutdown(*, reset_client: bool = False) -> None: + """Stop the periodic worker and perform a synchronized final flush.""" + with _AppInsightsEventClient._lifecycle_lock: + _AppInsightsEventClient._shutdown_requested = True + stop_event = _AppInsightsEventClient._flush_stop_event + worker = _AppInsightsEventClient._flush_thread + if stop_event: + stop_event.set() + + if worker and worker is not threading.current_thread(): + worker.join() + + _AppInsightsEventClient.flush() + + if _AppInsightsEventClient._flush_thread is worker: + _AppInsightsEventClient._flush_thread = None + _AppInsightsEventClient._flush_stop_event = None + + if reset_client: + _AppInsightsEventClient._client = None + _AppInsightsEventClient._initialized = False + _AppInsightsEventClient._pending_events.clear() + _AppInsightsEventClient._shutdown_requested = False @staticmethod def register_atexit_flush() -> None: - """Register an atexit handler to flush events on process exit.""" - if not _AppInsightsEventClient._atexit_registered: - atexit.register(_AppInsightsEventClient.flush) - _AppInsightsEventClient._atexit_registered = True + """Register an atexit handler to stop the worker and flush events.""" + with _AppInsightsEventClient._lifecycle_lock: + if not _AppInsightsEventClient._atexit_registered: + atexit.register(_AppInsightsEventClient._shutdown) + _AppInsightsEventClient._atexit_registered = True @staticmethod def reset() -> None: - """Flush pending events and reset so the next call re-initializes.""" - _AppInsightsEventClient.flush() - _AppInsightsEventClient._client = None - _AppInsightsEventClient._initialized = False + """Flush pending events, stop the worker, and reset client state.""" + _AppInsightsEventClient._shutdown(reset_client=True) class _TelemetryClient: @@ -405,7 +496,7 @@ def _initialize(): _logger.debug("Telemetry initialization error", exc_info=True) @staticmethod - def _track_method(name: str, attrs: Optional[Dict[str, Any]] = None): + def _track_method(name: str, attrs: Dict[str, Any] | None = None): """Track function invocations using OpenTelemetry.""" if not _TelemetryClient._is_enabled(): return @@ -417,7 +508,7 @@ def _track_method(name: str, attrs: Optional[Dict[str, Any]] = None): @staticmethod def track_event( name: str, - properties: Optional[Dict[str, Any]] = None, + properties: Dict[str, Any] | None = None, ) -> None: """Track a custom event to Application Insights customEvents table. @@ -453,7 +544,7 @@ def track_event( def track_event( name: str, - properties: Optional[Dict[str, Any]] = None, + properties: Dict[str, Any] | None = None, ) -> None: """Track a custom event. @@ -494,7 +585,7 @@ def flush_events() -> None: def set_event_connection_string_provider( - provider: Callable[[], Optional[str]], + provider: Callable[[], str | None], ) -> None: """Override how the Application Insights connection string is resolved. @@ -511,7 +602,7 @@ def reset_event_client() -> None: def track_cli_event( name: str, - properties: Optional[Dict[str, Any]] = None, + properties: Dict[str, Any] | None = None, ) -> None: """Track a CLI event. @@ -527,10 +618,10 @@ def track_cli_event( def track( - name_or_func: Optional[Union[str, Callable[..., Any]]] = None, + name_or_func: str | Callable[..., Any] | None = None, *, - when: Optional[Union[bool, Callable[..., bool]]] = True, - extra: Optional[Dict[str, Any]] = None, + when: bool | Callable[..., bool] | None = True, + extra: Dict[str, Any] | None = None, ): """Decorator that will trace function invocations. diff --git a/packages/uipath/tests/telemetry/test_track.py b/packages/uipath/tests/telemetry/test_track.py index ed44846b9..738c4ea2b 100644 --- a/packages/uipath/tests/telemetry/test_track.py +++ b/packages/uipath/tests/telemetry/test_track.py @@ -2,15 +2,20 @@ import json import os +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from queue import Empty, Queue from unittest.mock import MagicMock, patch import pytest +from uipath.core.feature_flags import FeatureFlags from uipath.platform.constants import ( ENV_PROJECT_KEY, ENV_UIPATH_AGENT_ID, ENV_UIPATH_PROJECT_ID, ) +from uipath.telemetry import PERIODIC_TELEMETRY_FLUSH_FEATURE_FLAG from uipath.telemetry._track import ( _AppInsightsEventClient, _DiagnosticSender, @@ -26,6 +31,49 @@ ) +@pytest.fixture +def appinsights_ingestion_server(): + """Capture App Insights ingestion requests on a local HTTP server.""" + received: Queue[list[dict[str, object]]] = Queue() + + class Handler(BaseHTTPRequestHandler): + def do_POST(self) -> None: + content_length = int(self.headers.get("Content-Length", "0")) + payload = json.loads(self.rfile.read(content_length)) + received.put(payload) + self.send_response(200) + self.end_headers() + + def log_message(self, format: str, *args: object) -> None: + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + server_thread = threading.Thread(target=server.serve_forever, daemon=True) + server_thread.start() + try: + yield f"http://127.0.0.1:{server.server_port}", received + finally: + server.shutdown() + server.server_close() + server_thread.join() + + +def _event_names(payload: list[dict[str, object]]) -> list[str]: + """Extract custom-event names from an App Insights ingestion payload.""" + names: list[str] = [] + for envelope in payload: + data = envelope.get("data") + if not isinstance(data, dict): + continue + base_data = data.get("baseData") + if not isinstance(base_data, dict): + continue + name = base_data.get("name") + if isinstance(name, str): + names.append(name) + return names + + class TestGetProjectKey: """`_get_project_key` resolution: uipath.json#id, then legacy telemetry file.""" @@ -128,15 +176,15 @@ class TestAppInsightsEventClient: def setup_method(self): """Reset AppInsightsEventClient state before each test.""" - _AppInsightsEventClient._initialized = False - _AppInsightsEventClient._client = None + _AppInsightsEventClient.reset() _AppInsightsEventClient._connection_string_provider = None + FeatureFlags.reset_flags() def teardown_method(self): """Clean up after each test.""" - _AppInsightsEventClient._initialized = False - _AppInsightsEventClient._client = None + _AppInsightsEventClient.reset() _AppInsightsEventClient._connection_string_provider = None + FeatureFlags.reset_flags() @patch("uipath.telemetry._track._CONNECTION_STRING", "$CONNECTION_STRING") def test_initialize_no_connection_string(self): @@ -428,6 +476,196 @@ def test_reset_allows_reinitialization_with_new_connection_string( assert _AppInsightsEventClient._client is mock_client_2 assert mock_client_class.call_count == 2 + def test_single_event_remains_buffered_when_periodic_flush_disabled( + self, monkeypatch, appinsights_ingestion_server + ): + """Default behavior buffers one event until an explicit flush.""" + endpoint, received = appinsights_ingestion_server + monkeypatch.setenv( + "TELEMETRY_CONNECTION_STRING", + f"InstrumentationKey=test-key;IngestionEndpoint={endpoint}", + ) + monkeypatch.setattr( + "uipath.telemetry._track._PERIODIC_TELEMETRY_FLUSH_INTERVAL_SECONDS", + 0.05, + ) + + track_event("single-buffered-event") + + with pytest.raises(Empty): + received.get(timeout=0.15) + + flush_events() + payload = received.get(timeout=1) + assert _event_names(payload) == ["single-buffered-event"] + + def test_single_event_is_sent_by_periodic_flush( + self, monkeypatch, appinsights_ingestion_server + ): + """Enabled periodic flushing sends one event without an explicit flush.""" + endpoint, received = appinsights_ingestion_server + monkeypatch.setenv( + "TELEMETRY_CONNECTION_STRING", + f"InstrumentationKey=test-key;IngestionEndpoint={endpoint}", + ) + monkeypatch.setattr( + "uipath.telemetry._track._PERIODIC_TELEMETRY_FLUSH_INTERVAL_SECONDS", + 0.05, + ) + monkeypatch.setenv( + f"UIPATH_FEATURE_{PERIODIC_TELEMETRY_FLUSH_FEATURE_FLAG}", "true" + ) + + track_event("single-periodic-event") + + payload = received.get(timeout=1) + assert _event_names(payload) == ["single-periodic-event"] + + def test_periodic_worker_skips_flush_without_pending_events(self, monkeypatch): + """Periodic ticks do not flush while no event is queued.""" + monkeypatch.setattr( + "uipath.telemetry._track._PERIODIC_TELEMETRY_FLUSH_INTERVAL_SECONDS", + 0.01, + ) + FeatureFlags.configure_flags({PERIODIC_TELEMETRY_FLUSH_FEATURE_FLAG: True}) + mock_client = MagicMock() + monkeypatch.setattr(_AppInsightsEventClient, "_initialized", True) + monkeypatch.setattr(_AppInsightsEventClient, "_client", mock_client) + + _AppInsightsEventClient._ensure_periodic_flush_worker() + threading.Event().wait(0.05) + + mock_client.flush.assert_not_called() + + def test_shutdown_waits_for_worker_and_serializes_final_flush(self, monkeypatch): + """Shutdown joins an active worker before performing its final flush.""" + monkeypatch.setattr( + "uipath.telemetry._track._PERIODIC_TELEMETRY_FLUSH_INTERVAL_SECONDS", + 0.01, + ) + FeatureFlags.configure_flags({PERIODIC_TELEMETRY_FLUSH_FEATURE_FLAG: True}) + + flush_entered = threading.Event() + release_flush = threading.Event() + counter_lock = threading.Lock() + active_flushes = 0 + max_active_flushes = 0 + + def blocking_flush() -> None: + nonlocal active_flushes, max_active_flushes + with counter_lock: + active_flushes += 1 + max_active_flushes = max(max_active_flushes, active_flushes) + flush_entered.set() + release_flush.wait(timeout=1) + with counter_lock: + active_flushes -= 1 + + mock_client = MagicMock() + mock_client.flush.side_effect = blocking_flush + monkeypatch.setattr(_AppInsightsEventClient, "_initialized", True) + monkeypatch.setattr(_AppInsightsEventClient, "_client", mock_client) + _AppInsightsEventClient.track_event("test-event") + assert flush_entered.wait(timeout=1) + + shutdown_thread = threading.Thread(target=_AppInsightsEventClient._shutdown) + shutdown_thread.start() + shutdown_thread.join(timeout=0.05) + assert shutdown_thread.is_alive() + + release_flush.set() + shutdown_thread.join(timeout=1) + + assert not shutdown_thread.is_alive() + assert max_active_flushes == 1 + assert mock_client.flush.call_count == 2 + assert _AppInsightsEventClient._flush_thread is None + + def test_periodic_flush_does_not_block_event_tracking(self, monkeypatch): + """A slow flush must not block the event-producing thread.""" + FeatureFlags.configure_flags({PERIODIC_TELEMETRY_FLUSH_FEATURE_FLAG: True}) + flush_entered = threading.Event() + release_flush = threading.Event() + + def blocking_flush() -> None: + flush_entered.set() + release_flush.wait(timeout=1) + + mock_client = MagicMock() + mock_client.flush.side_effect = blocking_flush + monkeypatch.setattr(_AppInsightsEventClient, "_initialized", True) + monkeypatch.setattr(_AppInsightsEventClient, "_client", mock_client) + + flush_thread = threading.Thread(target=_AppInsightsEventClient.flush) + flush_thread.start() + assert flush_entered.wait(timeout=1) + + track_thread = threading.Thread( + target=_AppInsightsEventClient.track_event, + args=("event-during-flush",), + ) + track_thread.start() + track_thread.join(timeout=0.2) + + assert not track_thread.is_alive() + mock_client.track_event.assert_called_once() + + release_flush.set() + flush_thread.join(timeout=1) + assert not flush_thread.is_alive() + + def test_reset_prevents_worker_restart_during_shutdown(self, monkeypatch): + """Reset keeps worker lifecycle serialized until client state is cleared.""" + monkeypatch.setattr( + "uipath.telemetry._track._PERIODIC_TELEMETRY_FLUSH_INTERVAL_SECONDS", + 0.01, + ) + FeatureFlags.configure_flags({PERIODIC_TELEMETRY_FLUSH_FEATURE_FLAG: True}) + flush_entered = threading.Event() + release_flush = threading.Event() + + def blocking_flush() -> None: + flush_entered.set() + release_flush.wait(timeout=1) + + mock_client = MagicMock() + mock_client.flush.side_effect = blocking_flush + monkeypatch.setattr(_AppInsightsEventClient, "_initialized", True) + monkeypatch.setattr(_AppInsightsEventClient, "_client", mock_client) + _AppInsightsEventClient.track_event("start-worker") + assert flush_entered.wait(timeout=1) + + reset_thread = threading.Thread(target=_AppInsightsEventClient.reset) + reset_thread.start() + reset_thread.join(timeout=0.05) + assert reset_thread.is_alive() + + track_thread = threading.Thread( + target=_AppInsightsEventClient.track_event, + args=("event-during-reset",), + ) + track_thread.start() + track_thread.join(timeout=0.05) + assert track_thread.is_alive() + + release_flush.set() + reset_thread.join(timeout=1) + track_thread.join(timeout=1) + + assert not reset_thread.is_alive() + assert not track_thread.is_alive() + assert _AppInsightsEventClient._client is None + assert _AppInsightsEventClient._flush_thread is None + + def test_atexit_registration_uses_synchronized_shutdown(self, monkeypatch): + """The idempotent atexit callback owns worker shutdown and final flush.""" + monkeypatch.setattr(_AppInsightsEventClient, "_atexit_registered", False) + with patch("uipath.telemetry._track.atexit.register") as register: + _AppInsightsEventClient.register_atexit_flush() + _AppInsightsEventClient.register_atexit_flush() + + register.assert_called_once_with(_AppInsightsEventClient._shutdown) + class TestPublicProviderAndResetFunctions: """Test the public set_event_connection_string_provider and reset_event_client.""" diff --git a/packages/uipath/uv.lock b/packages/uipath/uv.lock index cca93a5f9..796ffb972 100644 --- a/packages/uipath/uv.lock +++ b/packages/uipath/uv.lock @@ -2598,7 +2598,7 @@ wheels = [ [[package]] name = "uipath" -version = "2.13.14" +version = "2.13.15" source = { editable = "." } dependencies = [ { name = "applicationinsights" },