diff --git a/.github/workflows/sdk-spec.yml b/.github/workflows/sdk-spec.yml index ca9c03d..5b5b483 100644 --- a/.github/workflows/sdk-spec.yml +++ b/.github/workflows/sdk-spec.yml @@ -11,7 +11,8 @@ permissions: jobs: sdk-spec: # Keep the ref in `uses:` and `spec-ref` in sync. - uses: Flagsmith/sdk-spec/.github/workflows/run-spec.yml@v0.7.0 + uses: Flagsmith/sdk-spec/.github/workflows/run-spec.yml@v0.9.0 with: - spec-ref: v0.7.0 + spec-ref: v0.9.0 + tags: core events local-evaluation language: python diff --git a/flagsmith/analytics.py b/flagsmith/analytics.py index f5af134..7e47400 100644 --- a/flagsmith/analytics.py +++ b/flagsmith/analytics.py @@ -1,11 +1,18 @@ import atexit +import enum import json import logging +import os +import random import threading +import time import typing +import weakref +from collections import deque from dataclasses import dataclass from datetime import datetime +import requests from requests_futures.sessions import FuturesSession # type: ignore from flagsmith.version import DEFAULT_USER_AGENT, __version__ @@ -21,6 +28,14 @@ DEFAULT_EVENT_API_URL: typing.Final[str] = "https://events.api.flagsmith.com/" +EVENTS_REQUEST_TIMEOUT_SECONDS: typing.Final[float] = 3 +EVENTS_MAX_ATTEMPTS: typing.Final[int] = 3 +EVENTS_RETRY_BASE_SECONDS: typing.Final[float] = 1 +DEFAULT_EVENTS_CLOSE_TIMEOUT_SECONDS: typing.Final[float] = 5 +RETRIED_STATUS_CODES: typing.Final[typing.FrozenSet[int]] = frozenset( + {408, 429, 500, 502, 503, 504} +) + session = FuturesSession(max_workers=4) @@ -96,11 +111,30 @@ class EventProcessorConfig: flush_interval_seconds: float = 10.0 +class _BufferedEvent(typing.NamedTuple): + payload: str + exposure_key: typing.Optional[ExposureKey] + + +class _DeliveryOutcome(enum.Enum): + DELIVERED = "delivered" + DROPPED = "dropped" + KEPT = "kept" + REFUSED = "refused" + + class EventProcessor: """ Buffered event processor that batches custom events and POSTs them to the - Flagsmith event endpoint. Flushes on a background timer or when the buffer - fills. + Flagsmith event endpoint, from one worker thread. The worker sends the + buffered events on a timer, when the buffer fills, on `flush`, and once + more on `close`. + + A batch is sent up to ``EVENTS_MAX_ATTEMPTS`` times, backing off with full + jitter between attempts. A batch whose last attempt fails transiently is + kept for the next send, unless that attempt got a 500. A batch the events + API rejects is dropped, and an events API refusing the environment key + stops event tracking. When the buffer is full, the oldest event is dropped. """ def __init__( @@ -113,13 +147,41 @@ def __init__( url = f"{url}/" self._batch_endpoint = f"{url}v1/events" self._environment_key = environment_key - self._max_buffer = config.max_buffer_items + if config.max_buffer_items < 1: + logger.warning( + "max_buffer_items must be at least 1, not %d: using 1", + config.max_buffer_items, + ) + self._max_buffer = max(config.max_buffer_items, 1) self._flush_interval_seconds = config.flush_interval_seconds + self._refused = False + self._closed = False + self._start_worker() + _processors.add(self) - self._buffer: typing.List[typing.Dict[str, typing.Any]] = [] + def _start_worker(self) -> None: + """ + Starts the worker, with empty state. Also runs in a forked child, where + the parent's worker doesn't exist and the parent sends its own events. + """ + self._session = requests.Session() + self._condition = threading.Condition() + self._buffer: typing.Deque[_BufferedEvent] = deque() self._buffered_exposure_keys: typing.Set[ExposureKey] = set() - self._lock = threading.Lock() - self._timer: typing.Optional[threading.Timer] = None + self._dropped_count = 0 + self._tracked_count = 0 + self._sent_count = 0 + # Events tracked up to this count are being sent, if a send is running. + self._sending_count: typing.Optional[int] = None + self._flush_requested = False + self._sends_started = 0 + self._sends_completed = 0 + # Set once `close` returns: the worker starts no more requests. + self._abandoned = threading.Event() + self._worker = threading.Thread( + target=self._run, name="flagsmith-event-processor", daemon=True + ) + self._worker.start() def track_event( self, @@ -164,8 +226,28 @@ def _buffer_event( traits: typing.Optional[typing.Dict[str, typing.Any]], metadata: typing.Optional[typing.Dict[str, typing.Any]], ) -> None: - should_flush = False - with self._lock: + try: + serialised_event = json.dumps( + { + "event": event, + "feature_name": feature_name, + "identifier": identifier, + "value": str(value) if value is not None else None, + "traits": dict(traits) if traits else None, + "metadata": {**(metadata or {}), "sdk_version": __version__}, + "timestamp": int(datetime.now().timestamp() * 1000), + } + ) + except (TypeError, ValueError) as e: + logger.warning("Not tracking %s event: not JSON serialisable: %s", event, e) + return + with self._condition: + if self._closed or self._refused: + logger.debug( + "Not tracking %s event: event tracking has stopped.", event + ) + return + exposure_key = None if event == FLAG_EXPOSURE_EVENT: # An exposure is defined by who saw which variant of which # feature; equal exposures within one flush interval add no @@ -184,79 +266,223 @@ def _buffer_event( ) return self._buffered_exposure_keys.add(exposure_key) - self._buffer.append( - { - "event": event, - "feature_name": feature_name, - "identifier": identifier, - "value": str(value) if value is not None else None, - "traits": dict(traits) if traits else None, - "metadata": {**(metadata or {}), "sdk_version": __version__}, - "timestamp": int(datetime.now().timestamp() * 1000), - } - ) - if len(self._buffer) >= self._max_buffer: - should_flush = True - - if should_flush: - self.flush() + self._buffer.append(_BufferedEvent(serialised_event, exposure_key)) + self._tracked_count += 1 + self._drop_oldest_events() + if self._has_full_buffer_of_new_events(): + self._condition.notify_all() def flush(self) -> None: - with self._lock: - if not self._buffer: + """ + Sends the buffered events, including any kept from a failed send, and + returns once each of them has been delivered, dropped, or kept for the + next send. If a send already running covers every event tracked so + far, waits for it instead. Does nothing once closed. + """ + with self._condition: + if self._closed: return - events = self._buffer - self._buffer = [] - self._buffered_exposure_keys.clear() - - payload = json.dumps({"events": events}) - try: - future = session.post( - self._batch_endpoint, - data=payload, - timeout=3, - headers={ - "Content-Type": "application/json; charset=utf-8", - "X-Environment-Key": self._environment_key, - "Flagsmith-SDK-User-Agent": DEFAULT_USER_AGENT, - }, + if ( + self._sending_count is not None + and self._sending_count >= self._tracked_count + ): + send = self._sends_started + else: + self._flush_requested = True + send = self._sends_started + 1 + self._condition.notify_all() + self._condition.wait_for( + lambda: self._sends_completed >= send or not self._worker.is_alive() ) - except RuntimeError: - logger.debug("Skipping flush: thread pool already shut down") - return - future.add_done_callback(lambda f: self._handle_flush_result(f, events)) - def _handle_flush_result( - self, - future: typing.Any, - events: typing.List[typing.Dict[str, typing.Any]], + def close( + self, timeout_seconds: float = DEFAULT_EVENTS_CLOSE_TIMEOUT_SECONDS ) -> None: - try: - response = future.result() - response.raise_for_status() - except Exception: + """ + Stops event tracking, and sends the buffered events, waiting for at + most `timeout_seconds`, also after `stop`. No request starts once this + returns. Calling it again does nothing. + """ + self.stop() + if self._abandoned.is_set(): + return + self._worker.join(timeout_seconds) + self._abandoned.set() + if self._worker.is_alive(): logger.warning( - "Failed to flush pipeline analytics, re-queuing events", exc_info=True + "Gave up sending events after %s seconds while closing", + timeout_seconds, ) - with self._lock: - self._buffer = events + self._buffer - self._buffer = self._buffer[: self._max_buffer] - - def start(self) -> None: - self._schedule_flush() - atexit.register(self.stop) def stop(self) -> None: - atexit.unregister(self.stop) - if self._timer is not None: - self._timer.cancel() - self.flush() - - def _schedule_flush(self) -> None: - self._timer = threading.Timer(self._flush_interval_seconds, self._timer_flush) - self._timer.daemon = True - self._timer.start() - - def _timer_flush(self) -> None: - self.flush() - self._schedule_flush() + """ + Stops event tracking, and has the worker send the buffered events, + without waiting for it. `close` still runs at exit, to wait for it, as + long as the worker is sending. + """ + with self._condition: + self._closed = True + self._condition.notify_all() + + def _run(self) -> None: + next_send_at = time.monotonic() + self._flush_interval_seconds + while True: + with self._condition: + while not ( + self._closed + or self._flush_requested + or self._has_full_buffer_of_new_events() + or time.monotonic() >= next_send_at + ): + self._condition.wait(next_send_at - time.monotonic()) + closing = self._closed + events = list(self._buffer) + self._buffer.clear() + self._buffered_exposure_keys.clear() + self._flush_requested = False + self._sends_started += 1 + sending_count = self._sending_count = self._tracked_count + dropped_count, self._dropped_count = self._dropped_count, 0 + next_send_at = time.monotonic() + self._flush_interval_seconds + if dropped_count: + logger.warning( + "Event buffer was full: dropped the %d oldest events", + dropped_count, + ) + + try: + outcome = ( + self._deliver([event.payload for event in events]) + if events + else _DeliveryOutcome.DELIVERED + ) + except Exception: + logger.exception("Dropping %d events: failed to send", len(events)) + outcome = _DeliveryOutcome.DROPPED + + with self._condition: + self._sent_count = sending_count + self._sending_count = None + self._sends_completed += 1 + if outcome is _DeliveryOutcome.KEPT: + self._buffer.extendleft(reversed(events)) + self._buffered_exposure_keys.update( + event.exposure_key for event in events if event.exposure_key + ) + self._drop_oldest_events() + elif outcome is _DeliveryOutcome.REFUSED: + self._refused = True + self._buffer.clear() + self._buffered_exposure_keys.clear() + self._condition.notify_all() + if closing: + self._session.close() + return + + def _has_full_buffer_of_new_events(self) -> bool: + """ + Whether a buffer's worth of events was tracked since the last send. A + buffer full of events kept from failed sends waits for the timer. + """ + return self._tracked_count - self._sent_count >= self._max_buffer + + def _drop_oldest_events(self) -> None: + while len(self._buffer) > self._max_buffer: + dropped = self._buffer.popleft() + self._buffered_exposure_keys.discard(dropped.exposure_key) + self._dropped_count += 1 + + def _deliver(self, events: typing.List[str]) -> _DeliveryOutcome: + payload = f'{{"events": [{", ".join(events)}]}}' + status: typing.Optional[int] = None + for attempt in range(1, EVENTS_MAX_ATTEMPTS + 1): + if attempt > 1: + backoff = EVENTS_RETRY_BASE_SECONDS * 2 ** (attempt - 2) + self._abandoned.wait(random.uniform(0, backoff)) + if self._abandoned.is_set(): + logger.warning( + "Not sending %d events: the client is closed", len(events) + ) + return _DeliveryOutcome.DROPPED + try: + response = self._post(payload) + except (requests.ConnectionError, requests.Timeout) as e: + status = None + failure = str(e) + except requests.RequestException as e: + logger.error("Dropping %d events: %s", len(events), e) + return _DeliveryOutcome.DROPPED + else: + status = response.status_code + if outcome := self._outcome_of(status, len(events)): + return outcome + failure = f"status {status}" + logger.warning( + "Failed to send %d events, attempt %d of %d: %s", + len(events), + attempt, + EVENTS_MAX_ATTEMPTS, + failure, + ) + if status == 500: + logger.warning( + "Dropping %d events the events API failed with 500", len(events) + ) + return _DeliveryOutcome.DROPPED + logger.warning("Keeping %d events for the next send", len(events)) + return _DeliveryOutcome.KEPT + + def _post(self, payload: str) -> requests.Response: + return self._session.post( + self._batch_endpoint, + data=payload, + timeout=EVENTS_REQUEST_TIMEOUT_SECONDS, + headers={ + "Content-Type": "application/json; charset=utf-8", + "X-Environment-Key": self._environment_key, + "Flagsmith-SDK-User-Agent": DEFAULT_USER_AGENT, + }, + ) + + @staticmethod + def _outcome_of(status: int, event_count: int) -> typing.Optional[_DeliveryOutcome]: + """The outcome of a request answered with `status`, or None to retry it.""" + if status < 400: + return _DeliveryOutcome.DELIVERED + if status in (401, 403): + logger.error( + "Events API refused the environment key with %d," + " stopping event tracking", + status, + ) + return _DeliveryOutcome.REFUSED + if status in RETRIED_STATUS_CODES: + return None + logger.warning( + "Events API rejected %d events with %d, dropping them", + event_count, + status, + ) + return _DeliveryOutcome.DROPPED + + +_processors: "weakref.WeakSet[EventProcessor]" = weakref.WeakSet() + + +def _restart_workers_after_fork() -> None: + for processor in list(_processors): + processor._start_worker() + + +def _close_processors_at_exit() -> None: + # One hook for every processor: a hook per processor would keep each one + # alive until exit. The worker keeps its processor alive while it runs, so + # a stopped processor is still closed here if its events are being sent. + for processor in list(_processors): + processor.close() + + +atexit.register(_close_processors_at_exit) + +if hasattr(os, "register_at_fork"): + os.register_at_fork(after_in_child=_restart_workers_after_fork) diff --git a/flagsmith/flagsmith.py b/flagsmith/flagsmith.py index f50ceec..7dd0026 100644 --- a/flagsmith/flagsmith.py +++ b/flagsmith/flagsmith.py @@ -9,6 +9,7 @@ from urllib3 import Retry from flagsmith.analytics import ( + DEFAULT_EVENTS_CLOSE_TIMEOUT_SECONDS, FLAG_EXPOSURE_EVENT, AnalyticsProcessor, EventProcessor, @@ -232,7 +233,6 @@ def _initialise_events( config=event_processor_config or EventProcessorConfig(), environment_key=environment_key, ) - self._event_processor.start() def _initialise_local_evaluation(self) -> None: # To ensure that the environment is set before allowing subsequent @@ -453,6 +453,38 @@ def track_exposure_event( metadata=metadata, ) + def flush_events(self) -> None: + """ + Send the events tracked so far, and return once each of them has been + delivered, dropped, or kept for the next send, retries included. Does + nothing when events are not enabled, or once the client is closed. + """ + if not self._event_processor: + logger.warning("Not flushing events: events are not enabled.") + return + self._event_processor.flush() + + def close( + self, timeout_seconds: float = DEFAULT_EVENTS_CLOSE_TIMEOUT_SECONDS + ) -> None: + """ + Stop the client's background work, and send the buffered events. Events + tracked afterwards are not sent. Calling it again does nothing. + + :param timeout_seconds: how long to wait for the buffered events to be + sent. Events not sent by then are lost. + """ + self._stop_background_work() + if self._event_processor: + self._event_processor.close(timeout_seconds) + + def _stop_background_work(self) -> None: + if hasattr(self, "environment_data_polling_manager_thread"): + self.environment_data_polling_manager_thread.stop() + + if hasattr(self, "event_stream_thread"): + self.event_stream_thread.stop() + def update_environment(self) -> None: try: environment_data = self._get_json_response( @@ -592,11 +624,6 @@ def _get_json_response( ) from e def __del__(self) -> None: - if hasattr(self, "environment_data_polling_manager_thread"): - self.environment_data_polling_manager_thread.stop() - - if hasattr(self, "event_stream_thread"): - self.event_stream_thread.stop() - + self._stop_background_work() if self._event_processor: self._event_processor.stop() diff --git a/sdk-spec/steps.py b/sdk-spec/steps.py index 7c7700a..3f2f221 100644 --- a/sdk-spec/steps.py +++ b/sdk-spec/steps.py @@ -50,6 +50,19 @@ def create_client_with_local_evaluation( ) +# Create the SDK client with local evaluation and events enabled +def create_client_with_local_evaluation_and_events( + environment_key: str, api_url: str, events_api_url: str +) -> Flagsmith: + return Flagsmith( + environment_key=environment_key, + api_url=api_url, + enable_local_evaluation=True, + enable_events=True, + event_processor_config=EventProcessorConfig(events_api_url=events_api_url), + ) + + # When the application gets the environment flags def get_environment_flags(flagsmith: Flagsmith) -> list[SpecFlag]: return as_spec_flags(flagsmith.get_environment_flags()) @@ -92,3 +105,21 @@ def track_exposure_event( flagsmith.track_exposure_event( feature, identifier=identifier, value=value, metadata=metadata ) + + +# When the application tracks events "event_{first}" to "event_{last}" for identity "{identifier}" +def track_numbered_events( + flagsmith: Flagsmith, first: int, last: int, identifier: str +) -> None: + for number in range(first, last + 1): + flagsmith.track_event(f"event_{number}", identifier=identifier) + + +# When the application flushes events +def flush_events(flagsmith: Flagsmith) -> None: + flagsmith.flush_events() + + +# When the application closes the client with a timeout of {timeout} seconds +def close_client(flagsmith: Flagsmith, timeout: float) -> None: + flagsmith.close(timeout_seconds=timeout) diff --git a/tests/conftest.py b/tests/conftest.py index 0663be9..9bd8b58 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -51,7 +51,8 @@ def tracking_init(self: Flagsmith, *args: typing.Any, **kwargs: typing.Any) -> N stream.stop() stream.join(timeout=5) if flagsmith._event_processor: - flagsmith._event_processor.stop() + flagsmith._event_processor.close(timeout_seconds=0) + flagsmith._event_processor._worker.join(5) @pytest.fixture() @@ -69,11 +70,14 @@ def event_processor_config() -> EventProcessorConfig: @pytest.fixture() def event_processor( event_processor_config: EventProcessorConfig, -) -> EventProcessor: - return EventProcessor( +) -> Generator[EventProcessor, None, None]: + processor = EventProcessor( config=event_processor_config, environment_key="test_key", ) + yield processor + processor.close(timeout_seconds=0) + processor._worker.join(5) @pytest.fixture(scope="session") diff --git a/tests/test_event_processor.py b/tests/test_event_processor.py index 52595a7..2667839 100644 --- a/tests/test_event_processor.py +++ b/tests/test_event_processor.py @@ -1,16 +1,35 @@ +import gc import json -from concurrent.futures import Future +import logging +import os +import threading +import time +import typing +import weakref from datetime import datetime -from unittest import mock + +import pytest +import requests +import responses +from pytest_mock import MockerFixture from flagsmith.analytics import ( FLAG_EXPOSURE_EVENT, EventProcessor, EventProcessorConfig, + _close_processors_at_exit, ) -def test_track_event_buffers_event(event_processor: EventProcessor) -> None: +def test_flush__tracked_event__sent_to_events_api( + mocker: MockerFixture, + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + post = mocker.spy(event_processor._session, "post") + mocked_responses.post("http://test_analytics/v1/events", status=202) + before = int(datetime.now().timestamp() * 1000) event_processor.track_event( event="purchase", identifier="user1", @@ -18,10 +37,23 @@ def test_track_event_buffers_event(event_processor: EventProcessor) -> None: traits={"plan": "premium"}, metadata={"sku": "abc"}, ) - event_processor.track_event(event="purchase", identifier="user1") + after = int(datetime.now().timestamp() * 1000) - assert len(event_processor._buffer) == 2 - event = event_processor._buffer[0] + # When + event_processor.flush() + + # Then + assert len(mocked_responses.calls) == 1 + assert post.call_args.kwargs["timeout"] == 3 + request = mocked_responses.calls[0].request + assert request.headers["X-Environment-Key"] == "test_key" + assert request.headers["Content-Type"] == "application/json; charset=utf-8" + assert "flagsmith-python-sdk/" in request.headers["Flagsmith-SDK-User-Agent"] + request_body = request.body + assert request_body + body = json.loads(request_body) + assert "environment_key" not in body + [event] = body["events"] assert event["event"] == "purchase" assert event["feature_name"] is None assert event["identifier"] == "user1" @@ -29,19 +61,29 @@ def test_track_event_buffers_event(event_processor: EventProcessor) -> None: assert event["traits"] == {"plan": "premium"} assert event["metadata"]["sku"] == "abc" assert "sdk_version" in event["metadata"] - assert isinstance(event["timestamp"], int) + assert before <= event["timestamp"] <= after -def test_buffer_stringifies_non_string_values( +def test_flush__event_values__sent_as_strings( event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, ) -> None: + # Given + mocked_responses.post("http://test_analytics/v1/events", status=202) event_processor.track_event(event="purchase", value=99.5) event_processor.track_event(event="opt_in", value=True) event_processor.track_event(event="count", value=3) event_processor.track_event(event="variant", value="control") event_processor.track_event(event="empty") - assert [e["value"] for e in event_processor._buffer] == [ + # When + event_processor.flush() + + # Then + request_body = mocked_responses.calls[0].request.body + assert request_body + body = json.loads(request_body) + assert [e["value"] for e in body["events"]] == [ "99.5", "True", "3", @@ -50,19 +92,12 @@ def test_buffer_stringifies_non_string_values( ] -def test_track_event_defaults_timestamp_to_now( - event_processor: EventProcessor, -) -> None: - before = int(datetime.now().timestamp() * 1000) - event_processor.track_event(event="ping") - after = int(datetime.now().timestamp() * 1000) - - assert before <= event_processor._buffer[0]["timestamp"] <= after - - -def test_track_exposure_event_buffers_with_flag_exposure_event_name( +def test_flush__exposure_event__sent_as_flag_exposure( event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, ) -> None: + # Given + mocked_responses.post("http://test_analytics/v1/events", status=202) event_processor.track_exposure_event( feature_name="checkout_v2", identifier="user1", @@ -71,8 +106,13 @@ def test_track_exposure_event_buffers_with_flag_exposure_event_name( metadata={"source": "homepage"}, ) - assert len(event_processor._buffer) == 1 - event = event_processor._buffer[0] + # When + event_processor.flush() + + # Then + request_body = mocked_responses.calls[0].request.body + assert request_body + [event] = json.loads(request_body)["events"] assert event["event"] == FLAG_EXPOSURE_EVENT == "$flag_exposure" assert event["feature_name"] == "checkout_v2" assert event["identifier"] == "user1" @@ -82,25 +122,32 @@ def test_track_exposure_event_buffers_with_flag_exposure_event_name( assert "sdk_version" in event["metadata"] -def test_track_exposure_event__equal_events_in_flush_interval__deduplicated( +def test_flush__equal_exposures_in_flush_interval__sent_once( event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, ) -> None: - # Given / When + # Given + mocked_responses.post("http://test_analytics/v1/events", status=202) for _ in range(3): event_processor.track_exposure_event( - feature_name="checkout_v2", - identifier="user1", - value="variant_b", + feature_name="checkout_v2", identifier="user1", value="variant_b" ) + # When + event_processor.flush() + # Then - assert len(event_processor._buffer) == 1 + request_body = mocked_responses.calls[0].request.body + assert request_body + assert len(json.loads(request_body)["events"]) == 1 -def test_track_exposure_event__differing_events__not_deduplicated( +def test_flush__differing_exposures__each_sent( event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, ) -> None: - # Given / When + # Given + mocked_responses.post("http://test_analytics/v1/events", status=202) event_processor.track_exposure_event( feature_name="checkout_v2", identifier="user1", value="variant_b" ) @@ -114,103 +161,782 @@ def test_track_exposure_event__differing_events__not_deduplicated( feature_name="banner_test", identifier="user1", value="variant_b" ) + # When + event_processor.flush() + # Then - assert len(event_processor._buffer) == 4 + request_body = mocked_responses.calls[0].request.body + assert request_body + assert len(json.loads(request_body)["events"]) == 4 -def test_track_exposure_event__equal_event_after_flush__buffered_again( +def test_flush__equal_exposure_after_flush__sent_again( event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, ) -> None: # Given - with mock.patch("flagsmith.analytics.session"): - event_processor.track_exposure_event( - feature_name="checkout_v2", identifier="user1", value="variant_b" - ) - event_processor.flush() + mocked_responses.post("http://test_analytics/v1/events", status=202) + event_processor.track_exposure_event( + feature_name="checkout_v2", identifier="user1", value="variant_b" + ) + event_processor.flush() + event_processor.track_exposure_event( + feature_name="checkout_v2", identifier="user1", value="variant_b" + ) - # When - event_processor.track_exposure_event( - feature_name="checkout_v2", identifier="user1", value="variant_b" - ) + # When + event_processor.flush() # Then - assert len(event_processor._buffer) == 1 + assert len(mocked_responses.calls) == 2 -def test_track_event__equal_events__not_deduplicated( +def test_flush__equal_custom_events__each_sent( event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, ) -> None: - # Given / When + # Given + mocked_responses.post("http://test_analytics/v1/events", status=202) event_processor.track_event(event="purchase", identifier="user1", value="99.5") event_processor.track_event(event="purchase", identifier="user1", value="99.5") + # When + event_processor.flush() + # Then - assert len(event_processor._buffer) == 2 + request_body = mocked_responses.calls[0].request.body + assert request_body + assert len(json.loads(request_body)["events"]) == 2 + + +def test_track_event__metadata_not_json_serialisable__event_not_tracked( + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, +) -> None: + # Given + mocked_responses.post("http://test_analytics/v1/events", status=202) + # When + event_processor.track_event(event="purchase", metadata={"at": datetime.now()}) + event_processor.track_event(event="checkout") + event_processor.flush() -def test_auto_flush_on_buffer_full() -> None: - config = EventProcessorConfig(events_api_url="http://test/", max_buffer_items=5) - processor = EventProcessor(config=config, environment_key="key") + # Then + request_body = mocked_responses.calls[0].request.body + assert request_body + assert [e["event"] for e in json.loads(request_body)["events"]] == ["checkout"] + assert caplog.messages == [ + "Not tracking purchase event: not JSON serialisable:" + " Object of type datetime is not JSON serializable" + ] - with mock.patch("flagsmith.analytics.session"): - for i in range(5): - processor.track_event(event=f"event_{i}") - assert len(processor._buffer) == 0 +def test_track_event__full_buffer__sent_without_flush( + mocked_responses: responses.RequestsMock, +) -> None: + # Given + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", max_buffer_items=5 + ), + environment_key="test_key", + ) + mocked_responses.post("http://test_analytics/v1/events", status=202) + + # When + for i in range(5): + processor.track_event(event=f"event_{i}") + deadline = time.monotonic() + 5 + while not mocked_responses.calls and time.monotonic() < deadline: + time.sleep(0.01) + processor.flush() + processor.close(timeout_seconds=0) + processor._worker.join(5) + # Then + assert len(mocked_responses.calls) == 1 + request_body = mocked_responses.calls[0].request.body + assert request_body + body = json.loads(request_body) + assert [e["event"] for e in body["events"]] == [f"event_{i}" for i in range(5)] -def test_flush_sends_correct_http_request(event_processor: EventProcessor) -> None: - with mock.patch("flagsmith.analytics.session") as mock_session: - event_processor.track_event(event="purchase", identifier="user1") - event_processor.flush() - mock_session.post.assert_called_once() - call_kwargs = mock_session.post.call_args - assert call_kwargs[0][0] == "http://test_analytics/v1/events" +def test_track_event__buffer_full_of_kept_events__waits_for_flush_interval( + mocker: MockerFixture, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + mocker.patch("flagsmith.analytics.random.uniform", return_value=0) + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", max_buffer_items=2 + ), + environment_key="test_key", + ) + mocked_responses.post("http://test_analytics/v1/events", status=503) + processor.track_event(event="event_1") + processor.track_event(event="event_2") + processor.flush() - headers = call_kwargs[1]["headers"] - assert headers["X-Environment-Key"] == "test_key" - assert headers["Content-Type"] == "application/json; charset=utf-8" - assert "flagsmith-python-sdk/" in headers["Flagsmith-SDK-User-Agent"] + # When + processor.track_event(event="event_3") + time.sleep(0.3) - body = json.loads(call_kwargs[1]["data"]) - assert "environment_key" not in body - assert len(body["events"]) == 1 - assert body["events"][0]["event"] == "purchase" + # Then + assert len(mocked_responses.calls) == 3 + processor.close(timeout_seconds=0) + processor._worker.join(5) + + +@pytest.mark.parametrize("max_buffer_items", [0, -1]) +def test_track_event__max_buffer_items_below_one__event_sent_without_flush( + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, + max_buffer_items: int, +) -> None: + # Given + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", + max_buffer_items=max_buffer_items, + ), + environment_key="test_key", + ) + mocked_responses.post("http://test_analytics/v1/events", status=202) + + # When + processor.track_event(event="purchase") + deadline = time.monotonic() + 5 + while not mocked_responses.calls and time.monotonic() < deadline: + time.sleep(0.01) + processor.close(timeout_seconds=0) + processor._worker.join(5) + + # Then + assert len(mocked_responses.calls) == 1 + request_body = mocked_responses.calls[0].request.body + assert request_body + assert [e["event"] for e in json.loads(request_body)["events"]] == ["purchase"] + assert caplog.messages[0] == ( + f"max_buffer_items must be at least 1, not {max_buffer_items}: using 1" + ) + + +def test_flush__full_buffer__drops_oldest_event( + mocker: MockerFixture, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, +) -> None: + # Given + mocker.patch("flagsmith.analytics.random.uniform", return_value=0) + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", max_buffer_items=3 + ), + environment_key="test_key", + ) + for status in (503, 503, 503, 202): + mocked_responses.post("http://test_analytics/v1/events", status=status) + for i in range(1, 4): + processor.track_event(event=f"event_{i}") + processor.flush() + + # When + processor.track_event(event="event_4") + processor.flush() + processor.close(timeout_seconds=0) + processor._worker.join(5) + + # Then + assert len(mocked_responses.calls) == 4 + request_body = mocked_responses.calls[3].request.body + assert request_body + body = json.loads(request_body) + assert [e["event"] for e in body["events"]] == ["event_2", "event_3", "event_4"] + assert "Event buffer was full: dropped the 1 oldest events" in caplog.messages + + +def test_flush__no_events__sends_nothing( + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, +) -> None: + # When + event_processor.flush() + + # Then + assert len(mocked_responses.calls) == 0 + + +def test_flush__transient_failures__retried_with_backoff( + mocker: MockerFixture, + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, +) -> None: + # Given + uniform = mocker.patch("flagsmith.analytics.random.uniform", return_value=0) + mocked_responses.post("http://test_analytics/v1/events", status=503) + mocked_responses.post( + "http://test_analytics/v1/events", body=requests.ConnectionError("reset") + ) + mocked_responses.post("http://test_analytics/v1/events", status=202) + event_processor.track_event(event="purchase") + + # When + event_processor.flush() + + # Then + assert len(mocked_responses.calls) == 3 + assert uniform.call_args_list == [mocker.call(0, 1), mocker.call(0, 2)] + assert caplog.messages == [ + "Failed to send 1 events, attempt 1 of 3: status 503", + "Failed to send 1 events, attempt 2 of 3: reset", + ] + + +@pytest.mark.parametrize( + "failures", + [ + pytest.param([503, 503, 503], id="unavailable"), + pytest.param([500, 500, 503], id="unavailable-after-server-errors"), + pytest.param([429, 408, 504], id="throttled-and-timed-out"), + pytest.param( + [502, requests.ConnectionError("reset"), requests.Timeout("timed out")], + id="unreachable", + ), + ], +) +def test_flush__last_attempt_fails_transiently__kept_ahead_of_newer_events( + mocker: MockerFixture, + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, + failures: typing.List[typing.Union[int, Exception]], +) -> None: + # Given + mocker.patch("flagsmith.analytics.random.uniform", return_value=0) + for failure in failures: + if isinstance(failure, Exception): + mocked_responses.post("http://test_analytics/v1/events", body=failure) + else: + mocked_responses.post("http://test_analytics/v1/events", status=failure) + mocked_responses.post("http://test_analytics/v1/events", status=202) + event_processor.track_event(event="add_to_cart") + event_processor.flush() + + # When + event_processor.track_event(event="checkout") + event_processor.flush() + + # Then + assert len(mocked_responses.calls) == 4 + request_body = mocked_responses.calls[3].request.body + assert request_body + body = json.loads(request_body) + assert [e["event"] for e in body["events"]] == ["add_to_cart", "checkout"] + assert "Keeping 1 events for the next send" in caplog.messages + + +def test_flush__kept_batch_and_no_new_events__kept_batch_sent( + mocker: MockerFixture, + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + mocker.patch("flagsmith.analytics.random.uniform", return_value=0) + for status in (503, 503, 503, 202): + mocked_responses.post("http://test_analytics/v1/events", status=status) + event_processor.track_event(event="add_to_cart") + event_processor.flush() + + # When + event_processor.flush() + + # Then + assert len(mocked_responses.calls) == 4 + request_body = mocked_responses.calls[3].request.body + assert request_body + assert [e["event"] for e in json.loads(request_body)["events"]] == ["add_to_cart"] + + +def test_flush__send_in_flight_covers_tracked_events__waits_without_resending( + mocker: MockerFixture, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + mocker.patch("flagsmith.analytics.random.uniform", return_value=0) + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", max_buffer_items=2 + ), + environment_key="test_key", + ) + requested = threading.Event() + answer = threading.Event() + + def answer_when_released( + request: requests.PreparedRequest, + ) -> typing.Tuple[int, typing.Dict[str, str], str]: + requested.set() + answer.wait(5) + return 503, {}, "" + + mocked_responses.add_callback( + responses.POST, + "http://test_analytics/v1/events", + callback=answer_when_released, + ) + processor.track_event(event="event_1") + processor.track_event(event="event_2") + requested.wait(5) + threading.Timer(0.1, answer.set).start() + + # When + processor.flush() + + # Then + assert len(mocked_responses.calls) == 3 + processor.close(timeout_seconds=0) + processor._worker.join(5) -def test_flush_noop_when_empty(event_processor: EventProcessor) -> None: - with mock.patch("flagsmith.analytics.session") as mock_session: - event_processor.flush() +def test_track_exposure_event__equal_exposure_dropped_from_full_buffer__sent( + mocked_responses: responses.RequestsMock, +) -> None: + # Given + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", max_buffer_items=2 + ), + environment_key="test_key", + ) + requested = threading.Event() + answer = threading.Event() + + def answer_when_released( + request: requests.PreparedRequest, + ) -> typing.Tuple[int, typing.Dict[str, str], str]: + requested.set() + answer.wait(5) + return 202, {}, "" + + mocked_responses.add_callback( + responses.POST, + "http://test_analytics/v1/events", + callback=answer_when_released, + ) + processor.track_event(event="event_1") + processor.track_event(event="event_2") + requested.wait(5) + processor.track_exposure_event( + feature_name="checkout", identifier="user1", value="variant_b" + ) + processor.track_event(event="event_3") + processor.track_event(event="event_4") - mock_session.post.assert_not_called() + # When + processor.track_exposure_event( + feature_name="checkout", identifier="user1", value="variant_b" + ) + answer.set() + processor.flush() + processor.close(timeout_seconds=0) + processor._worker.join(5) + # Then + sent_events = [] + for call in mocked_responses.calls: + request_body = call.request.body + assert request_body + sent_events += [e["event"] for e in json.loads(request_body)["events"]] + assert sent_events == ["event_1", "event_2", "event_4", FLAG_EXPOSURE_EVENT] + + +@pytest.mark.parametrize( + "statuses, expected_log", + [ + pytest.param( + [400], + "Events API rejected 1 events with 400, dropping them", + id="rejected", + ), + pytest.param( + [501], + "Events API rejected 1 events with 501, dropping them", + id="not-implemented", + ), + pytest.param( + [503, 503, 500], + "Dropping 1 events the events API failed with 500", + id="server-error-last", + ), + ], +) +def test_flush__batch_rejected_or_last_failed_with_server_error__dropped( + mocker: MockerFixture, + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, + statuses: typing.List[int], + expected_log: str, +) -> None: + # Given + mocker.patch("flagsmith.analytics.random.uniform", return_value=0) + for status in (*statuses, 202): + mocked_responses.post("http://test_analytics/v1/events", status=status) + event_processor.track_event(event="add_to_cart") + event_processor.flush() -def test_failed_flush_requeues_events(event_processor: EventProcessor) -> None: - future: Future[None] = Future() - future.set_exception(Exception("connection error")) + # When + event_processor.track_event(event="checkout") + event_processor.flush() - with mock.patch("flagsmith.analytics.session") as mock_session: - mock_session.post.return_value = future - event_processor.track_event(event="purchase") - event_processor.flush() + # Then + assert len(mocked_responses.calls) == len(statuses) + 1 + request_body = mocked_responses.calls[-1].request.body + assert request_body + body = json.loads(request_body) + assert [e["event"] for e in body["events"]] == ["checkout"] + assert expected_log in caplog.messages - assert len(event_processor._buffer) == 1 - assert event_processor._buffer[0]["event"] == "purchase" +def test_flush__request_cannot_be_made__dropped_without_retrying( + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, +) -> None: + # Given + mocked_responses.post( + "http://test_analytics/v1/events", + body=requests.exceptions.InvalidHeader("invalid header"), + ) + event_processor.track_event(event="purchase") -def test_start_stop_lifecycle() -> None: - config = EventProcessorConfig( - events_api_url="http://test/", flush_interval_seconds=100 + # When + event_processor.flush() + + # Then + assert len(mocked_responses.calls) == 1 + assert caplog.messages == ["Dropping 1 events: invalid header"] + + +def test_flush__unexpected_error__returns_and_keeps_tracking( + mocker: MockerFixture, + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, +) -> None: + # Given + mocker.patch.object( + event_processor._session, + "post", + side_effect=[RuntimeError("unexpected"), mocker.DEFAULT], + wraps=event_processor._session.post, ) - processor = EventProcessor(config=config, environment_key="key") + mocked_responses.post("http://test_analytics/v1/events", status=202) + event_processor.track_event(event="purchase") + event_processor.flush() - processor.start() - assert processor._timer is not None - assert processor._timer.is_alive() + # When + event_processor.track_event(event="checkout") + event_processor.flush() + + # Then + assert len(mocked_responses.calls) == 1 + assert caplog.messages == ["Dropping 1 events: failed to send"] - with mock.patch("flagsmith.analytics.session"): - processor.track_event(event="purchase") - processor.stop() - assert len(processor._buffer) == 0 +@pytest.mark.parametrize("status", [401, 403]) +def test_flush__environment_key_refused__stops_event_tracking( + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, + status: int, +) -> None: + # Given + mocked_responses.post("http://test_analytics/v1/events", status=status) + event_processor.track_event(event="add_to_cart") + event_processor.flush() + + # When + with caplog.at_level(logging.DEBUG, logger="flagsmith.analytics"): + event_processor.track_event(event="checkout") + event_processor.flush() + + # Then + assert len(mocked_responses.calls) == 1 + assert caplog.messages == [ + f"Events API refused the environment key with {status}," + " stopping event tracking", + "Not tracking checkout event: event tracking has stopped.", + ] + + +def test_events__flush_interval_elapsed__sent_without_flush( + mocked_responses: responses.RequestsMock, +) -> None: + # Given + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", flush_interval_seconds=0.01 + ), + environment_key="test_key", + ) + mocked_responses.post("http://test_analytics/v1/events", status=202) + + # When + processor.track_event(event="purchase") + deadline = time.monotonic() + 5 + while not mocked_responses.calls and time.monotonic() < deadline: + time.sleep(0.01) + processor.close(timeout_seconds=0) + processor._worker.join(5) + + # Then + assert len(mocked_responses.calls) == 1 + + +def test_close__sends_buffered_events_and_stops_event_tracking( + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, +) -> None: + # Given + mocked_responses.post("http://test_analytics/v1/events", status=202) + event_processor.track_event(event="purchase") + + # When + event_processor.close() + with caplog.at_level(logging.DEBUG, logger="flagsmith.analytics"): + event_processor.track_event(event="checkout") + event_processor.flush() + event_processor.close() + + # Then + assert len(mocked_responses.calls) == 1 + request_body = mocked_responses.calls[0].request.body + assert request_body + assert [e["event"] for e in json.loads(request_body)["events"]] == ["purchase"] + assert not event_processor._worker.is_alive() + assert caplog.messages == [ + "Not tracking checkout event: event tracking has stopped." + ] + + +def test_close__events_api_not_answering__returns_within_timeout( + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, +) -> None: + # Given + answer = threading.Event() + mocked_responses.add_callback( + responses.POST, + "http://test_analytics/v1/events", + callback=lambda request: (answer.wait(5), (503, {}, ""))[1], + ) + event_processor.track_event(event="purchase") + + # When + started = time.monotonic() + event_processor.close(timeout_seconds=0.1) + elapsed = time.monotonic() - started + answer.set() + event_processor._worker.join(5) + + # Then + assert elapsed < 1 + assert len(mocked_responses.calls) == 1 + assert caplog.messages == [ + "Gave up sending events after 0.1 seconds while closing", + "Failed to send 1 events, attempt 1 of 3: status 503", + "Not sending 1 events: the client is closed", + ] + + +def test_stop__events_api_not_answering__returns_without_waiting( + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + answer = threading.Event() + mocked_responses.add_callback( + responses.POST, + "http://test_analytics/v1/events", + callback=lambda request: (answer.wait(5), (202, {}, ""))[1], + ) + event_processor.track_event(event="purchase") + + # When + started = time.monotonic() + event_processor.stop() + elapsed = time.monotonic() - started + answer.set() + event_processor._worker.join(5) + + # Then + assert elapsed < 1 + assert len(mocked_responses.calls) == 1 + assert not event_processor._worker.is_alive() + + +def test_close__after_stop__waits_for_buffered_events( + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + answer = threading.Event() + mocked_responses.add_callback( + responses.POST, + "http://test_analytics/v1/events", + callback=lambda request: (answer.wait(5), (202, {}, ""))[1], + ) + event_processor.track_event(event="purchase") + event_processor.stop() + threading.Timer(0.2, answer.set).start() + + # When + event_processor.close(timeout_seconds=5) + + # Then + assert len(mocked_responses.calls) == 1 + assert not event_processor._worker.is_alive() + + +def test_track_event__full_buffer_while_sending__drops_logged_once( + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, +) -> None: + # Given + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", max_buffer_items=2 + ), + environment_key="test_key", + ) + requested = threading.Event() + answer = threading.Event() + + def answer_when_released( + request: requests.PreparedRequest, + ) -> typing.Tuple[int, typing.Dict[str, str], str]: + requested.set() + answer.wait(5) + return 202, {}, "" + + mocked_responses.add_callback( + responses.POST, + "http://test_analytics/v1/events", + callback=answer_when_released, + ) + processor.track_event(event="event_1") + processor.track_event(event="event_2") + requested.wait(5) + + # When + for i in range(3, 8): + processor.track_event(event=f"event_{i}") + answer.set() + processor.flush() + processor.close(timeout_seconds=0) + processor._worker.join(5) + + # Then + assert [m for m in caplog.messages if m.startswith("Event buffer")] == [ + "Event buffer was full: dropped the 3 oldest events" + ] + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="needs os.fork") +@pytest.mark.filterwarnings( + "ignore:This process .* is multi-threaded:DeprecationWarning" +) +def test_track_event__forked_child__each_process_sends_its_own_events( + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + mocked_responses.post("http://test_analytics/v1/events", status=202) + event_processor.track_event(event="tracked_before_fork") + + # When + pid = os.fork() + if pid == 0: + exit_code = 1 + try: + event_processor.track_event(event="tracked_in_child") + event_processor.flush() + child_sent_events = [] + for call in mocked_responses.calls: + request_body = call.request.body + assert request_body + child_sent_events += [ + e["event"] for e in json.loads(request_body)["events"] + ] + exit_code = 0 if child_sent_events == ["tracked_in_child"] else 2 + finally: + os._exit(exit_code) + _, wait_status = os.waitpid(pid, 0) + event_processor.flush() + + # Then + assert os.waitstatus_to_exitcode(wait_status) == 0 + assert len(mocked_responses.calls) == 1 + request_body = mocked_responses.calls[0].request.body + assert request_body + assert [e["event"] for e in json.loads(request_body)["events"]] == [ + "tracked_before_fork" + ] + + +def test_stop__processor_no_longer_referenced__garbage_collected( + event_processor_config: EventProcessorConfig, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + mocked_responses.post("http://test_analytics/v1/events", status=202) + processor = EventProcessor( + config=event_processor_config, environment_key="test_key" + ) + processor.track_event(event="purchase") + worker = processor._worker + collected = weakref.ref(processor) + + # When + processor.stop() + del processor + worker.join(5) + gc.collect() + + # Then + assert len(mocked_responses.calls) == 1 + assert collected() is None + + +def test_close_processors_at_exit__stopped_processor_sending__waits_for_it( + event_processor_config: EventProcessorConfig, + mocked_responses: responses.RequestsMock, +) -> None: + # Given + answer = threading.Event() + mocked_responses.add_callback( + responses.POST, + "http://test_analytics/v1/events", + callback=lambda request: (answer.wait(5), (202, {}, ""))[1], + ) + processor = EventProcessor( + config=event_processor_config, environment_key="test_key" + ) + processor.track_event(event="purchase") + worker = processor._worker + processor.stop() + del processor + threading.Timer(0.2, answer.set).start() + + # When + _close_processors_at_exit() + + # Then + assert len(mocked_responses.calls) == 1 + assert not worker.is_alive() diff --git a/tests/test_flagsmith.py b/tests/test_flagsmith.py index 7998c23..b6879fc 100644 --- a/tests/test_flagsmith.py +++ b/tests/test_flagsmith.py @@ -1090,6 +1090,112 @@ def test_track_event_delegates_to_event_processor( ) +def test_flush_events__events_tracked__sent_to_events_api( + api_key: str, mocked_responses: responses.RequestsMock +) -> None: + # Given + flagsmith = Flagsmith( + environment_key=api_key, + enable_events=True, + event_processor_config=EventProcessorConfig(events_api_url="http://test/"), + ) + mocked_responses.post("http://test/v1/events", status=202) + flagsmith.track_event("purchase", identifier="user1") + + # When + flagsmith.flush_events() + + # Then + assert len(mocked_responses.calls) == 1 + request_body = mocked_responses.calls[0].request.body + assert request_body + body = json.loads(request_body) + assert [e["event"] for e in body["events"]] == ["purchase"] + + +def test_flush_events__events_disabled__does_nothing( + api_key: str, caplog: pytest.LogCaptureFixture +) -> None: + # Given + flagsmith = Flagsmith(environment_key=api_key) + + # When + flagsmith.flush_events() + + # Then + assert caplog.messages == ["Not flushing events: events are not enabled."] + + +def test_close__events_tracked__sent_within_timeout( + mocker: MockerFixture, api_key: str +) -> None: + # Given + flagsmith = Flagsmith(environment_key=api_key, enable_events=True) + assert flagsmith._event_processor + mock_close = mocker.patch.object(flagsmith._event_processor, "close") + + # When + flagsmith.close(timeout_seconds=2) + + # Then + mock_close.assert_called_once_with(2) + + +def test_del__events_tracked__stops_event_processor_without_waiting( + mocker: MockerFixture, api_key: str +) -> None: + # Given + flagsmith = Flagsmith(environment_key=api_key, enable_events=True) + assert flagsmith._event_processor + mock_stop = mocker.patch.object(flagsmith._event_processor, "stop") + mock_close = mocker.patch.object(flagsmith._event_processor, "close") + + # When + flagsmith.__del__() + + # Then + mock_stop.assert_called_once_with() + mock_close.assert_not_called() + + +def test_close__local_evaluation__stops_polling( + mocker: MockerFixture, + local_eval_flagsmith: Flagsmith, +) -> None: + # Given + mock_stop = mocker.patch.object( + local_eval_flagsmith.environment_data_polling_manager_thread, "stop" + ) + + # When + local_eval_flagsmith.close() + + # Then + mock_stop.assert_called_once_with() + + +def test_close__realtime_updates__stops_streaming( + requests_session_response_ok: None, server_api_key: str, mocker: MockerFixture +) -> None: + # Given + mocker.patch( + "flagsmith.streaming_manager.requests.get", + side_effect=requests.exceptions.ReadTimeout(), + ) + flagsmith = Flagsmith( + environment_key=server_api_key, + enable_local_evaluation=True, + enable_realtime_updates=True, + ) + mock_stop = mocker.patch.object(flagsmith.event_stream_thread, "stop") + + # When + flagsmith.close() + + # Then + mock_stop.assert_called_once_with() + + def test_track_exposure_event_raises_without_config(api_key: str) -> None: flagsmith = Flagsmith(environment_key=api_key) with pytest.raises(ValueError, match="Events must be enabled"):