diff --git a/.github/workflows/sdk-spec.yml b/.github/workflows/sdk-spec.yml index ca9c03d..d23ee3f 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.8.0 with: - spec-ref: v0.7.0 + spec-ref: v0.8.0 + tags: core or events or local-evaluation language: python diff --git a/flagsmith/analytics.py b/flagsmith/analytics.py index f5af134..77d5e9c 100644 --- a/flagsmith/analytics.py +++ b/flagsmith/analytics.py @@ -1,11 +1,14 @@ import atexit +import enum import json import logging +import random import threading import typing 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 +24,12 @@ 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}) + session = FuturesSession(max_workers=4) @@ -96,11 +105,25 @@ class EventProcessorConfig: flush_interval_seconds: float = 10.0 +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. + + A batch is sent up to ``EVENTS_MAX_ATTEMPTS`` times, backing off with full + jitter between attempts, one batch at a time. A batch whose last attempt + fails transiently is kept for the next flush, 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__( @@ -116,9 +139,18 @@ def __init__( self._max_buffer = config.max_buffer_items self._flush_interval_seconds = config.flush_interval_seconds + self._session = requests.Session() self._buffer: typing.List[typing.Dict[str, typing.Any]] = [] self._buffered_exposure_keys: typing.Set[ExposureKey] = set() self._lock = threading.Lock() + # Held while a batch is in flight, so that one batch is sent at a time. + self._send_lock = threading.Lock() + self._tracked_count = 0 + self._attempted_count = 0 + self._full_buffer_flush_pending = False + self._refused = False + self._stopped = threading.Event() + self._closed = threading.Event() self._timer: typing.Optional[threading.Timer] = None def track_event( @@ -164,8 +196,14 @@ 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: + if self._refused: + logger.debug( + "Not tracking %s event: the events API refused the" + " environment key.", + event, + ) + return 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,6 +222,12 @@ def _buffer_event( ) return self._buffered_exposure_keys.add(exposure_key) + if len(self._buffer) >= self._max_buffer: + dropped = self._buffer.pop(0) + logger.warning( + "Event buffer is full, dropping the oldest %s event", + dropped["event"], + ) self._buffer.append( { "event": event, @@ -195,68 +239,159 @@ def _buffer_event( "timestamp": int(datetime.now().timestamp() * 1000), } ) - if len(self._buffer) >= self._max_buffer: - should_flush = True + self._tracked_count += 1 + if len(self._buffer) < self._max_buffer or ( + self._full_buffer_flush_pending + ): + return + self._full_buffer_flush_pending = True + tracked_count = self._tracked_count - if should_flush: - self.flush() + threading.Thread( + target=self._flush_full_buffer, args=(tracked_count,), daemon=True + ).start() def flush(self) -> None: + """ + Sends the events tracked so far, and returns once each of them has + been delivered, dropped, or kept for the next flush. Events already + sent while this was waiting for another batch in flight are not sent + again. + """ with self._lock: - if not self._buffer: - return - events = self._buffer - self._buffer = [] - self._buffered_exposure_keys.clear() + tracked_count = self._tracked_count + self._send(tracked_count) - 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, - }, - ) - 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 _flush_full_buffer(self, tracked_count: int) -> None: + with self._lock: + self._full_buffer_flush_pending = False + self._send(tracked_count) - def _handle_flush_result( - self, - future: typing.Any, - events: typing.List[typing.Dict[str, typing.Any]], - ) -> None: - try: - response = future.result() - response.raise_for_status() - except Exception: + def _send(self, tracked_count: typing.Optional[int] = None) -> None: + """ + Sends the buffered events as one batch. Given `tracked_count`, does + nothing if every event tracked up to it has been sent already. + """ + with self._send_lock: + with self._lock: + if not self._buffer or ( + tracked_count is not None and self._attempted_count >= tracked_count + ): + return + events = self._buffer + self._buffer = [] + self._buffered_exposure_keys.clear() + attempted_count = self._tracked_count + + outcome = self._deliver(events) + + with self._lock: + self._attempted_count = attempted_count + if outcome is _DeliveryOutcome.KEPT: + max_buffer = self._max_buffer + self._buffer = (events + self._buffer)[-max_buffer:] + elif outcome is _DeliveryOutcome.REFUSED: + self._refused = True + self._buffer = [] + self._buffered_exposure_keys.clear() + + def _deliver( + self, events: typing.List[typing.Dict[str, typing.Any]] + ) -> _DeliveryOutcome: + payload = json.dumps({"events": events}) + for attempt in range(1, EVENTS_MAX_ATTEMPTS + 1): + if attempt > 1: + backoff = EVENTS_RETRY_BASE_SECONDS * 2 ** (attempt - 2) + if self._closed.wait(random.uniform(0, backoff)): + logger.warning( + "Not retrying %d events: the client is closed", len(events) + ) + return _DeliveryOutcome.KEPT + try: + response = 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, + }, + ) + except requests.RequestException as e: + logger.warning( + "Failed to send %d events, attempt %d of %d: %s", + len(events), + attempt, + EVENTS_MAX_ATTEMPTS, + e, + ) + status = None + else: + status = response.status_code + if response.ok: + 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 not in RETRIED_STATUS_CODES and status < 500: + logger.warning( + "Events API rejected %d events with %d, dropping them", + len(events), + status, + ) + return _DeliveryOutcome.DROPPED + logger.warning( + "Failed to send %d events, attempt %d of %d: status %d", + len(events), + attempt, + EVENTS_MAX_ATTEMPTS, + status, + ) + if status == 500: logger.warning( - "Failed to flush pipeline analytics, re-queuing events", exc_info=True + "Dropping %d events the events API failed with 500", len(events) ) - with self._lock: - self._buffer = events + self._buffer - self._buffer = self._buffer[: self._max_buffer] + return _DeliveryOutcome.DROPPED + logger.warning("Keeping %d events for the next flush", len(events)) + return _DeliveryOutcome.KEPT def start(self) -> None: self._schedule_flush() atexit.register(self.stop) - def stop(self) -> None: + def stop( + self, timeout_seconds: float = DEFAULT_EVENTS_CLOSE_TIMEOUT_SECONDS + ) -> None: + """ + Stops flushing on a timer, and sends the buffered events, waiting for + at most `timeout_seconds`. Events not sent by then are not retried. + """ atexit.unregister(self.stop) + self._stopped.set() if self._timer is not None: self._timer.cancel() - self.flush() + final_flush = threading.Thread(target=self._send, daemon=True) + final_flush.start() + final_flush.join(timeout_seconds) + self._closed.set() + if final_flush.is_alive(): + logger.warning( + "Gave up sending events after %s seconds while closing", + timeout_seconds, + ) def _schedule_flush(self) -> None: + if self._stopped.is_set(): + return 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._send() self._schedule_flush() diff --git a/flagsmith/flagsmith.py b/flagsmith/flagsmith.py index f50ceec..fd80e57 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, @@ -453,6 +454,33 @@ 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 flush, retries included. + """ + if not self._event_processor: + raise ValueError("Events must be enabled to flush events.") + 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. + + :param timeout_seconds: how long to wait for the buffered events to be + sent. Events not sent by then are lost. + """ + 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() + + if self._event_processor: + self._event_processor.stop(timeout_seconds) + def update_environment(self) -> None: try: environment_data = self._get_json_response( @@ -592,11 +620,4 @@ 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() - - if self._event_processor: - self._event_processor.stop() + self.close() 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/test_event_processor.py b/tests/test_event_processor.py index 52595a7..19cab36 100644 --- a/tests/test_event_processor.py +++ b/tests/test_event_processor.py @@ -1,7 +1,14 @@ import json -from concurrent.futures import Future +import logging +import threading +import time +import typing 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, @@ -120,18 +127,19 @@ def test_track_exposure_event__differing_events__not_deduplicated( def test_track_exposure_event__equal_event_after_flush__buffered_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() - # When - 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" + ) # Then assert len(event_processor._buffer) == 1 @@ -148,69 +156,337 @@ def test_track_event__equal_events__not_deduplicated( assert len(event_processor._buffer) == 2 -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") +def test_flush__full_buffer__sends_events_once( + 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) + for i in range(5): + processor.track_event(event=f"event_{i}") - with mock.patch("flagsmith.analytics.session"): - for i in range(5): - processor.track_event(event=f"event_{i}") + # When + processor.flush() - assert len(processor._buffer) == 0 + # 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)] + assert processor._buffer == [] + + +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() + + # 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 is full, dropping the oldest event_1 event" in ( + caplog.messages + ) -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_flush__sends_events_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) + event_processor.track_event(event="purchase", identifier="user1") - 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 + event_processor.flush() - body = json.loads(call_kwargs[1]["data"]) + # Then + assert len(mocked_responses.calls) == 1 + request = mocked_responses.calls[0].request + assert post.call_args.kwargs["timeout"] == 3 + 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 assert len(body["events"]) == 1 assert body["events"][0]["event"] == "purchase" + assert event_processor._buffer == [] -def test_flush_noop_when_empty(event_processor: EventProcessor) -> None: - with mock.patch("flagsmith.analytics.session") as mock_session: - event_processor.flush() +def test_flush__no_events__sends_nothing( + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, +) -> None: + # When + event_processor.flush() - mock_session.post.assert_not_called() + # Then + assert len(mocked_responses.calls) == 0 -def test_failed_flush_requeues_events(event_processor: EventProcessor) -> None: - future: Future[None] = Future() - future.set_exception(Exception("connection error")) +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") - with mock.patch("flagsmith.analytics.session") as mock_session: - mock_session.post.return_value = future - event_processor.track_event(event="purchase") - event_processor.flush() + # 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 event_processor._buffer == [] + assert caplog.messages == [ + "Failed to send 1 events, attempt 1 of 3: status 503", + "Failed to send 1 events, attempt 2 of 3: reset", + ] - assert len(event_processor._buffer) == 1 - assert event_processor._buffer[0]["event"] == "purchase" +@pytest.mark.parametrize( + "statuses", + [ + pytest.param([503, 503, 503], id="unavailable"), + pytest.param([500, 500, 503], id="unavailable-after-server-errors"), + pytest.param([429, 429, 429], id="throttled"), + ], +) +def test_flush__last_attempt_fails_transiently__kept_ahead_of_newer_events( + mocker: MockerFixture, + event_processor: EventProcessor, + mocked_responses: responses.RequestsMock, + caplog: pytest.LogCaptureFixture, + statuses: typing.List[int], +) -> 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_start_stop_lifecycle() -> None: - config = EventProcessorConfig( - events_api_url="http://test/", flush_interval_seconds=100 + # 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 flush" in caplog.messages + + +@pytest.mark.parametrize( + "statuses, expected_log", + [ + pytest.param( + [400], + "Events API rejected 1 events with 400, dropping them", + id="rejected", + ), + 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() + + # When + event_processor.track_event(event="checkout") + 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 + + +@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 event_processor._buffer == [] + assert caplog.messages == [ + f"Events API refused the environment key with {status}," + " stopping event tracking", + "Not tracking checkout event: the events API refused the environment key.", + ] + + +def test_stop__timer_flush_in_progress__no_further_timer_flushes( + mocked_responses: responses.RequestsMock, +) -> None: + # Given + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", flush_interval_seconds=100 + ), + environment_key="test_key", ) - processor = EventProcessor(config=config, environment_key="key") + processor.start() + timer = processor._timer + processor.stop() + # When + processor._timer_flush() + + # Then + assert processor._timer is timer + assert len(mocked_responses.calls) == 0 + + +def test_start__flushes_on_timer( + 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) + processor.track_event(event="purchase") + + # When + processor.start() + deadline = time.monotonic() + 5 + while not mocked_responses.calls and time.monotonic() < deadline: + time.sleep(0.01) + processor.stop() + + # Then + assert len(mocked_responses.calls) == 1 + assert processor._buffer == [] + + +def test_stop__sends_buffered_events( + mocked_responses: responses.RequestsMock, +) -> None: + # Given + processor = EventProcessor( + config=EventProcessorConfig( + events_api_url="http://test_analytics/", flush_interval_seconds=100 + ), + environment_key="test_key", + ) processor.start() + mocked_responses.post("http://test_analytics/v1/events", status=202) + processor.track_event(event="purchase") + + # When + processor.stop() + + # Then assert processor._timer is not None - assert processor._timer.is_alive() + assert processor._timer.finished.is_set() + assert len(mocked_responses.calls) == 1 + assert processor._buffer == [] - with mock.patch("flagsmith.analytics.session"): - processor.track_event(event="purchase") - processor.stop() - assert len(processor._buffer) == 0 +def test_stop__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.stop(timeout_seconds=0.1) + elapsed = time.monotonic() - started + answer.set() + with event_processor._send_lock: + pass + + # 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 retrying 1 events: the client is closed", + ] diff --git a/tests/test_flagsmith.py b/tests/test_flagsmith.py index 7998c23..9084211 100644 --- a/tests/test_flagsmith.py +++ b/tests/test_flagsmith.py @@ -1090,6 +1090,69 @@ 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__raises(api_key: str) -> None: + # Given + flagsmith = Flagsmith(environment_key=api_key) + + # When / Then + with pytest.raises(ValueError, match="Events must be enabled to flush events."): + flagsmith.flush_events() + + +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_stop = mocker.patch.object(flagsmith._event_processor, "stop") + + # When + flagsmith.close(timeout_seconds=2) + + # Then + mock_stop.assert_called_once_with(2) + + +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_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"):