diff --git a/.github/workflows/code-quality.yaml b/.github/workflows/code-quality.yaml index 4bbc780..fb7b364 100644 --- a/.github/workflows/code-quality.yaml +++ b/.github/workflows/code-quality.yaml @@ -28,7 +28,7 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip - pip install flake8 pytest pytest-mock + pip install flake8 pytest pytest-mock build if [ -f requirements.txt ]; then pip install -r requirements.txt; fi - name: Lint with flake8 run: | @@ -37,3 +37,12 @@ jobs: - name: Test with pytest run: | pytest + - name: Audit concurrency, memory, and thread cleanup + if: matrix.python-version == '3.12' + run: python scripts/resource_audit.py --evaluations 20000 --workers 8 + - name: Build production distributions + if: matrix.python-version == '3.12' + run: python -m build + - name: Verify wheel contains production packages only + if: matrix.python-version == '3.12' + run: python -c "import glob, zipfile; names = zipfile.ZipFile(glob.glob('dist/*.whl')[0]).namelist(); assert not any(name.startswith(('tests/', 'featbit_openfeature/')) for name in names)" diff --git a/MANIFEST.in b/MANIFEST.in index 971366d..0feff78 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -1,4 +1,4 @@ include requirements.txt include README.md include dev-requirements.txt -include release/package.json \ No newline at end of file +include release/package.json diff --git a/README.md b/README.md index 5e79b3f..6141fc4 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,7 @@ If you want to use your own data source, see [Offline Mode](#offline-mode). ## Get Started ### Installation -install the sdk in using pip, this version of the SDK is compatible with Python 3.6 through 3.11. +Install the SDK using pip. This version is compatible with Python 3.6 through 3.12. ```shell pip install fb-python-sdk @@ -66,6 +66,29 @@ client.stop() - [Python Demo](https://github.com/featbit/featbit-samples/blob/main/samples/dino-game/demo-python/demo_python.py) +### SDK reliability and live verification + +The repository includes a repeatable [reliability review](docs/REVIEW.md) +covering unit tests, thread safety, memory retention, background-thread +shutdown, public runtime exception isolation, and redundant code. Run the +local resource audit with: + +```shell +python scripts/resource_audit.py --evaluations 80000 --workers 8 +``` + +To verify WebSocket synchronization, remote evaluation, status changes, event +delivery, and clean shutdown against FeatBit Cloud or a self-hosted evaluation +service, set `FEATBIT_ENV_SECRET` and run: + +```shell +python scripts/live_integration_check.py +``` + +The live script never prints or persists the environment secret. See the +review document for all supported environment variables and verification +criteria. + ### FBClient Applications **SHOULD instantiate a single FBClient instance** for the lifetime of the application. In the case where an application @@ -112,6 +135,16 @@ if client.update_status_provider.wait_for_OKState(): It's possible to set a timeout in seconds for the `wait_for_OKState` method. If the timeout is reached, the method will return `False` and the client will still be in an uninitialized state. If you do not specify a timeout, the method will wait indefinitely. +You can also observe later connection interruptions and recoveries: + +```python +def on_status_change(state): + print(state.state_type, state.error_track) + +client.update_status_provider.add_listener(on_status_change) +# Remove it during application shutdown: +client.update_status_provider.remove_listener(on_status_change) +``` > To check if the client is ready is optional. Even if the client is not ready, you can still evaluate feature flags, but the default value will be returned if SDK is not yet initialized. @@ -136,6 +169,7 @@ if client.initialize: flag_value = client.variation(flag_key, user, default_value) # evaluate the flag value and get the detail detail = client.variation_detail(flag_key, user, default=None) + print(detail.variation, detail.variation_id, detail.reason) ``` If you would like to get variations of all feature flags in a special environment, you can use `fbclient.client.FBClient.get_all_latest_flag_variations`, SDK will return `fbclient.common_types.AllFlagStates`, that explain the details of all feature flags. `fbclient.common_types.AllFlagStates.get()` returns the detail of a given feature flag key. diff --git a/docs/REVIEW.md b/docs/REVIEW.md new file mode 100644 index 0000000..5ae685a --- /dev/null +++ b/docs/REVIEW.md @@ -0,0 +1,108 @@ +# Python Server SDK reliability review + +This review covers the production risks requested for the Python Server SDK: +unit behavior, WebSocket synchronization, event processing, evaluation, +logging/error isolation, thread safety, shutdown, memory retention, redundant +code, and live FeatBit service interoperability. + +## Review result + +The review found and fixed the following concrete issues: + +- `RepeatableTask` stored an `Event` in `Thread._stop`, which breaks Python's + own `Thread.join()` cleanup path. It now uses a separate stop event and joins + deterministically. +- WebSocket reconnect backoff used an uninterruptible sleep. Client shutdown + now interrupts the wait, closes the socket defensively, stops the ping task, + and joins the streaming thread. +- The notice broadcaster mutated its listener registry concurrently. Listener + operations now use a lock and dispatch from an immutable snapshot; a queue + sentinel makes shutdown immediate and repeatable. +- The event processor did not retain or join its dispatcher. It now owns the + dispatcher lifecycle, joins the periodic flush task, and always completes a + synchronous message even if message handling fails. +- `FBClient.stop()` could expose an extension exception and skip all later + cleanup. Shutdown is now idempotent, producer-first, and isolates each + component failure. +- Custom event processor errors could escape from `identify`, `track_*`, or + `flush`. Event delivery is now best-effort and cannot fail the application + request. +- Invalid offline bootstrap JSON and unsupported runtime fallback objects could + raise from evaluation-related calls. They now return `False` or the supplied + fallback, with a diagnostic log message. +- `Config` and the HTTP helper used mutable default objects. Each client now + gets independent HTTP/WebSocket options and headers. +- A duplicate `FBUser.from_dict()` call in `track_metric` was removed. + +Constructor/configuration validation still raises for invalid required +configuration. This is intentional fail-fast behavior before an SDK client is +usable; runtime evaluation, event, status, and shutdown paths are isolated. + +## Automated evidence + +Run from the repository root: + +```shell +python -m pytest +python -m flake8 . +python scripts/resource_audit.py --evaluations 80000 --workers 8 +python -m build +``` + +Result on Python 3.11.8 (2026-07-29): + +- 66 unit/integration tests passed. +- Flake8 passed. +- 80,000 evaluations across 8 workers completed with 0 concurrent errors. +- Throughput was 11,425 evaluations/second on the final audit run. +- No FeatBit SDK worker threads remained after shutdown. +- 24,389 traced bytes remained after garbage collection, below the 2 MiB + regression threshold. +- A clean wheel built from the source distribution contained 27 files, no + `tests/` package, and no stale `featbit_openfeature/` package. + +The tests specifically cover concurrent evaluation, status-listener ordering, +listener registration/removal under contention, event-processor failure +isolation, repeated client lifecycle, periodic task joining, and interrupting a +blocked WebSocket lifecycle. + +## Live FeatBit service verification + +Use a real environment Server Key without storing it in the repository: + +```shell +export FEATBIT_ENV_SECRET='' +export FEATBIT_FLAG_KEY='python-app-release' +python scripts/live_integration_check.py +``` + +Optional variables are `FEATBIT_STREAMING_URL`, `FEATBIT_EVENT_URL`, and +`FEATBIT_START_WAIT_SECONDS`. The script verifies: + +1. authenticated WebSocket connection and initial data synchronization; +2. SDK status reaches `OK`; +3. remote flag evaluation returns value, reason, and variation ID; +4. feature-flag, identify, and custom metric events enter the delivery path; +5. explicit flush and final synchronous drain complete; +6. status changes to `OFF` and no SDK threads remain after close. + +The secret is read only from the process environment and is never printed or +written to disk. + +A recorded FeatBit Cloud run on 2026-07-24 reached the WebSocket connected +state, processed the initial data-sync payload, evaluated the remote +`python-app-release` flag, exercised the gray-release application under +concurrent HTTP traffic, and later demonstrated automatic reconnection after a +remote-host disconnect. The checked-in script turns that one-off validation +into a repeatable, secret-safe release check. + +## Remaining operational considerations + +- The SDK is designed as a long-lived singleton, not one client per request. +- Event delivery is intentionally best-effort so analytics failure cannot + affect application availability. +- A custom `DataStorage`, `EventProcessor`, or `UpdateProcessor` remains + responsible for its own internal correctness; the client isolates its + runtime and shutdown exceptions. +- Memory thresholds are regression guards, not universal capacity limits; + production sizing should use the application's real flag and segment data. diff --git a/fbclient/__init__.py b/fbclient/__init__.py index 054a925..89c9c70 100644 --- a/fbclient/__init__.py +++ b/fbclient/__init__.py @@ -24,9 +24,7 @@ def get() -> FBClient: If you need to create multiple client instances with different environments, instead of this singleton approach you can call directly the :class:`fbclient.client.FBClient` constructor. """ - global __config global __client - global __lock try: __lock.read_lock() @@ -58,7 +56,6 @@ def set_config(config: Config): """ global __config global __client - global __lock try: __lock.write_lock() diff --git a/fbclient/client.py b/fbclient/client.py index f6bd0fa..b568f7f 100644 --- a/fbclient/client.py +++ b/fbclient/client.py @@ -68,6 +68,8 @@ def __init__(self, config: Config, start_wait: float = 15.): raise ValueError("Config is not valid") self._config = config + self._stop_lock = threading.Lock() + self._closed = False if self._config.is_offline: log.info("FB Python SDK: SDK is in offline mode") else: @@ -79,7 +81,7 @@ def __init__(self, config: Config, start_wait: float = 15.): # init components # event processor self._event_processor = self._build_event_processor(config) - self._event_handler = lambda event: self._event_processor.send_event(event) + self._event_handler = self._send_event_safely # data storage self._data_storage = config.data_storage # evaluator @@ -126,6 +128,15 @@ def _build_update_processor(self, config: Config, broadcaster: NoticeBroadcater, return Streaming(config, broadcaster, update_status_provider, update_processor_event) + def _send_event_safely(self, event): + try: + if not self._closed: + self._event_processor.send_event(event) + except Exception: + # Event delivery must never affect the application request that + # produced the event, including when a custom processor is used. + log.exception('FB Python SDK: event processor failed') + @property def initialize(self) -> bool: """Returns true if the client has successfully connected to feature flag center. @@ -154,11 +165,23 @@ def stop(self): Do not attempt to use the client after calling this method. """ - log.info("FB Python SDK: Python SDK client is closing...") - self._data_storage.stop() - self._update_processor.stop() - self._event_processor.stop() - self._broadcaster.stop() + with self._stop_lock: + if self._closed: + return + self._closed = True + log.info("FB Python SDK: Python SDK client is closing...") + # Stop producers before consumers, and isolate every component so + # one extension failure cannot prevent the remaining resources + # from being released or escape into application shutdown code. + for name, component in ( + ('update processor', self._update_processor), + ('event processor', self._event_processor), + ('notice broadcaster', self._broadcaster), + ('data storage', self._data_storage)): + try: + component.stop() + except Exception: + log.exception('FB Python SDK: %s failed to stop' % name) def __enter__(self): return self @@ -174,9 +197,13 @@ def is_offline(self) -> bool: def _get_flag_internal(self, key: str) -> Optional[dict]: return self._data_storage.get(FEATURE_FLAGS, key) - def __handle_default_value(self, key: str, default: Any) -> Tuple[Optional[str], Optional[str]]: + def __handle_default_value(self, key: str, default: Any) -> Tuple[Optional[str], Any]: default_value = self._config.get_default_value(key, default) - default_value_type = simple_type_inference(default_value) + try: + default_value_type = simple_type_inference(default_value) + except Exception: + log.warning('FB Python SDK: unsupported default value; returning it unchanged on evaluation failure') + return None, default_value if default_value is None: return None, None elif default_value_type == 'boolean': @@ -235,7 +262,7 @@ def variation(self, key: str, user: dict, default: Any = None) -> Any: :param default: the default value of the flag, to be used if the return value is not available :return: one of the flag's values in any type in any type of string, bool, float, json or the default value if flag evaluation fails - :raises: ValueError if the default is not a string, boolean, numeric, or json type + Unsupported defaults are returned unchanged if evaluation cannot produce a flag value. """ er = self._evaluate_internal(key, user, default) return cast_variation_by_flag_type(er.flag_type, er.value) @@ -252,7 +279,7 @@ def variation_detail(self, key: str, user: dict, default: Any = None) -> EvalDet :param user: the attributes of the user :param default: the default value of the flag, to be used if the return value is not available :return: an :class:`fbclient.common_types.EvalDetail` object - :raises: ValueError if the default is not a string, boolean, numeric, or json type + Unsupported defaults are returned unchanged if evaluation cannot produce a flag value. """ return self._evaluate_internal(key, user, default).to_evail_detail @@ -318,7 +345,11 @@ def flush(self): schedules the next event delivery to be as soon as possible; however, the delivery still happens asynchronously on a thread, so this method will return immediately. """ - self._event_processor.flush() + try: + if not self._closed: + self._event_processor.flush() + except Exception: + log.exception('FB Python SDK: event processor flush failed') def identify(self, user: dict): """register an end user in the feature flag center @@ -352,7 +383,6 @@ def track_metric(self, user: dict, event_name: str, metric_value: float = 1.0): log.warning('FB Python SDK: user invalid') return - fb_user = FBUser.from_dict(user) metric_event = MetricEvent(fb_user).add(Metric(event_name, metric_value)) self._event_handler(metric_event) @@ -385,10 +415,13 @@ def initialize_from_external_json(self, json_str: str) -> bool: :param json_str: feature flags, segments...etc in the json format :return: True if the initialization is well done """ - if self._config.is_offline: - all_data = json.loads(json_str) - if valide_all_data(all_data): - version, data = _data_to_dict(all_data['data']) - return self._update_status_provider.init(data, version) + try: + if self._config.is_offline: + all_data = json.loads(json_str) + if valide_all_data(all_data): + version, data = _data_to_dict(all_data['data']) + return self._update_status_provider.init(data, version) + except Exception: + log.exception('FB Python SDK: invalid external bootstrap data') return False diff --git a/fbclient/common_types.py b/fbclient/common_types.py index 9539f3b..c282201 100644 --- a/fbclient/common_types.py +++ b/fbclient/common_types.py @@ -90,19 +90,21 @@ def __init__(self, reason: str, variation: Any, key_name: Optional[str] = None, - name: Optional[str] = None): + name: Optional[str] = None, + variation_id: Optional[str] = None): """Constructs an instance. - :param id: variation id :param reason: main factor that influenced the flag evaluation value :param variation: result of the flag evaluation in any type of string, bool, float/int, json(Python object) or default value if flag evaluation fails :param key_name: key name of the flag :param name: name of the flag + :param variation_id: stable identifier of the resolved variation, or None if evaluation failed """ self._reason = reason self._variation = variation self._key_name = key_name self._name = name + self._variation_id = variation_id @property def reason(self) -> str: @@ -129,6 +131,12 @@ def name(self) -> Optional[str]: """ return self._name + @property + def variation_id(self) -> Optional[str]: + """The stable identifier of the resolved variation, if evaluation succeeded. + """ + return self._variation_id + def to_json_dict(self) -> dict: json_dict = {} json_dict['reason'] = self.reason @@ -329,7 +337,8 @@ def is_success(self) -> bool: @property def to_evail_detail(self) -> "EvalDetail": _value = cast_variation_by_flag_type(self.__flag_type, self.__value) - return EvalDetail(self.__reason, _value, self.__key_name, self.__name) + variation_id = self.__id if self.is_success else None + return EvalDetail(self.__reason, _value, self.__key_name, self.__name, variation_id) @property def to_flag_state(self) -> "FlagState": diff --git a/fbclient/config.py b/fbclient/config.py index 0f6c482..a675a78 100644 --- a/fbclient/config.py +++ b/fbclient/config.py @@ -135,8 +135,8 @@ def __init__(self, data_storage: Optional[DataStorage] = None, update_processor_imp: Optional[Callable[['Config', DataUpdateStatusProvider, Event], UpdateProcessor]] = None, event_processor_imp: Optional[Callable[['Config', Sender], EventProcessor]] = None, - http: HTTPConfig = HTTPConfig(), - websocket: WebSocketConfig = WebSocketConfig(), + http: Optional[HTTPConfig] = None, + websocket: Optional[WebSocketConfig] = None, defaults: Optional[dict] = None): self.__env_secret = env_secret @@ -155,8 +155,10 @@ def __init__(self, events_retry_interval, 1) self.__events_max_retries = 1 if events_max_retries is None or events_max_retries <= 0 else min( events_max_retries, 3) - self.__http = http - self.__websocket = websocket + # Avoid sharing mutable default configuration objects between SDK + # clients created by unrelated applications or tests. + self.__http = http if http is not None else HTTPConfig() + self.__websocket = websocket if websocket is not None else WebSocketConfig() self.__defaults = defaults if defaults is not None else {} def copy_config_in_a_new_env(self, env_secret: str, defaults=None) -> 'Config': diff --git a/fbclient/event_processor.py b/fbclient/event_processor.py index 9a527ca..be50225 100644 --- a/fbclient/event_processor.py +++ b/fbclient/event_processor.py @@ -1,6 +1,6 @@ import json from concurrent.futures import ThreadPoolExecutor -from queue import Empty, Queue +from queue import Empty, Full, Queue from threading import BoundedSemaphore, Condition, Lock, Thread from typing import List, Optional @@ -18,8 +18,9 @@ def __init__(self, config: Config, sender: Sender): self.__inbox = Queue(maxsize=config.events_max_in_queue) self.__closed = False self.__lock = Lock() - EventDispatcher(config, sender, self.__inbox).start() - self.__flush_task = RepeatableTask('insight flush', config.events_flush_interval, self.flush) + self.__dispatcher = EventDispatcher(config, sender, self.__inbox) + self.__dispatcher.start() + self.__flush_task = RepeatableTask('featbit-insight-flush', config.events_flush_interval, self.flush) self.__flush_task.start() log.debug('insight processor is ready') @@ -33,7 +34,7 @@ def __put_message_to_inbox(self, message: EventMessage) -> bool: try: self.__inbox.put_nowait(message) return True - except: + except Full: if message.type == MessageType.SHUTDOWN: # must put the shut down to inbox; self.__inbox.put(message, block=True, timeout=None) @@ -78,6 +79,9 @@ def stop(self): self.__flush_task.stop() self.__put_message_async(MessageType.FLUSH) self.__put_message_and_wait_terminate(MessageType.SHUTDOWN) + self.__dispatcher.join(5.0) + if self.__dispatcher.is_alive(): + log.warning('FB Python SDK: event dispatcher did not stop in time') class EventDispatcher(Thread): @@ -86,7 +90,7 @@ class EventDispatcher(Thread): __BATCH_SIZE = 50 def __init__(self, config: Config, sender: Sender, inbox: "Queue[EventMessage]"): - super().__init__(daemon=True) + super().__init__(name='featbit-event-dispatcher', daemon=True) self.__config = config self.__inbox = inbox self.__closed = False @@ -106,6 +110,7 @@ def run(self): try: msgs = self.__drain_inbox(size=self.__BATCH_SIZE) for msg in msgs: + shutdown = False try: if msg.type == MessageType.FLAGS or msg.type == MessageType.METRICS or msg.type == MessageType.USER: self.__put_events_to_buffer(msg.event) # type: ignore @@ -113,11 +118,15 @@ def run(self): self.__trigger_flush() elif msg.type == MessageType.SHUTDOWN: self.__shutdown() - msg.completed() - return # exit the loop - msg.completed() + shutdown = True except Exception as inner: log.exception('FB Python SDK: unexpected error in event dispatcher: %s' % str(inner)) + finally: + # Synchronous callers must never wait forever because a + # dispatcher operation failed. + msg.completed() + if shutdown: + return # exit the loop except Exception as outer: log.exception('FB Python SDK: unexpected error in event dispatcher: %s' % str(outer)) diff --git a/fbclient/interfaces.py b/fbclient/interfaces.py index 084ca1c..6994f59 100644 --- a/fbclient/interfaces.py +++ b/fbclient/interfaces.py @@ -1,5 +1,5 @@ from abc import ABC, abstractmethod -from typing import Mapping, Optional +from typing import Callable, Mapping, Optional from fbclient.category import Category from fbclient.common_types import FBEvent @@ -124,6 +124,24 @@ def wait_for_OKState(self, timeout: float) -> bool: """ pass + def add_listener(self, listener: Callable[[State], None]): + """ + Registers a listener for update-processing status changes. + + The default implementation is a no-op so existing custom status + providers remain backward compatible. + """ + pass + + def remove_listener(self, listener: Callable[[State], None]): + """ + Removes a previously registered update-status listener. + + The default implementation is a no-op so existing custom status + providers remain backward compatible. + """ + pass + class DataStorage(ABC): """ diff --git a/fbclient/notice_broadcaster.py b/fbclient/notice_broadcaster.py index 725b1c3..cab2097 100644 --- a/fbclient/notice_broadcaster.py +++ b/fbclient/notice_broadcaster.py @@ -1,6 +1,6 @@ -from queue import Empty, Queue -from threading import Thread +from queue import Queue +from threading import Lock, Thread, current_thread from typing import Callable from fbclient.interfaces import Notice @@ -12,46 +12,72 @@ def __init__(self): self.__notice_queue = Queue() self.__closed = False self.__listeners = {} - self.__thread = Thread(daemon=True, target=self.__run) + self.__lock = Lock() + self.__stop_notice = object() + self.__thread = Thread(name='featbit-notice-broadcaster', + daemon=True, + target=self.__run) log.debug('notice broadcaster starting...') self.__thread.start() def add_listener(self, notice_type: str, listener: Callable[[Notice], None]): if isinstance(notice_type, str) and notice_type.strip() and listener is not None: log.debug('add a listener for notice type %s' % notice_type) - if notice_type not in self.__listeners: - self.__listeners[notice_type] = [] - self.__listeners[notice_type].append(listener) + with self.__lock: + if self.__closed: + return + if notice_type not in self.__listeners: + self.__listeners[notice_type] = [] + # Preserve the existing contract: registering the same + # callable multiple times produces the same number of + # notifications. + self.__listeners[notice_type].append(listener) def remove_listener(self, notice_type: str, listener: Callable[[Notice], None]): - if notice_type in self.__listeners and listener is not None: - log.debug('remove a listener for notice type %s' % notice_type) - notifiers = self.__listeners[notice_type] + if listener is None: + return + log.debug('remove a listener for notice type %s' % notice_type) + with self.__lock: + notifiers = self.__listeners.get(notice_type) if not notifiers: - del self.__listeners[notice_type] - else: + return + try: notifiers.remove(listener) + except ValueError: + return + if not notifiers: + del self.__listeners[notice_type] def broadcast(self, notice: Notice): + with self.__lock: + if self.__closed: + return self.__notice_queue.put(notice) def stop(self): log.debug('notice broadcaster stopping...') - self.__closed = True - self.__thread.join() + with self.__lock: + if self.__closed: + return + self.__closed = True + self.__notice_queue.put(self.__stop_notice) + if current_thread() is not self.__thread: + self.__thread.join(5.0) + if self.__thread.is_alive(): + log.warning('FB Python SDK: notice broadcaster did not stop in time') def __run(self): - while not self.__closed: - try: - notice = self.__notice_queue.get(block=True, timeout=1) - self.__notice_process(notice) - except Empty: - pass + while True: + notice = self.__notice_queue.get(block=True, timeout=None) + if notice is self.__stop_notice: + return + self.__notice_process(notice) def __notice_process(self, notice: Notice): - if notice.notice_type in self.__listeners: - for listerner in self.__listeners[notice.notice_type]: - try: - listerner(notice) - except Exception as e: - log.exception('FB Python SDK: unexpected error in handle notice %s: %s' % (notice.notice_type, str(e))) + with self.__lock: + listeners = tuple(self.__listeners.get(notice.notice_type, ())) + for listener in listeners: + try: + listener(notice) + except Exception as e: + log.exception('FB Python SDK: unexpected error in handle notice %s: %s' % (notice.notice_type, str(e))) diff --git a/fbclient/status.py b/fbclient/status.py index 93242da..1a65bae 100644 --- a/fbclient/status.py +++ b/fbclient/status.py @@ -1,6 +1,7 @@ import threading +from collections import deque from time import time -from typing import Mapping +from typing import Callable, Deque, List, Mapping, Tuple from fbclient.category import Category from fbclient.interfaces import DataStorage, DataUpdateStatusProvider @@ -15,6 +16,11 @@ def __init__(self, storage: DataStorage): self.__storage = storage self.__current_state = State.intializing_state() self.__lock = threading.Condition(threading.Lock()) + self.__listeners: List[Callable[[State], None]] = [] + self.__pending_notifications: Deque[ + Tuple[State, Tuple[Callable[[State], None], ...]] + ] = deque() + self.__publishing_notifications = False def init(self, all_data: Mapping[Category, Mapping[str, dict]], version: int = 0) -> bool: try: @@ -55,6 +61,7 @@ def current_state(self) -> State: def update_state(self, new_state: State): if not new_state: return + publish_notifications = False with self.__lock: old_state_type = self.__current_state.state_type new_state_type = new_state.state_type @@ -70,6 +77,48 @@ def update_state(self, new_state: State): self.__current_state = State(new_state_type, state_since, error) # wakes up all threads waiting for the ok state to check the new state self.__lock.notify_all() + self.__pending_notifications.append( + (self.__current_state, tuple(self.__listeners)) + ) + if not self.__publishing_notifications: + self.__publishing_notifications = True + publish_notifications = True + + if publish_notifications: + self.__publish_pending_notifications() + + def __publish_pending_notifications(self): + while True: + with self.__lock: + if not self.__pending_notifications: + self.__publishing_notifications = False + return + state, listeners = self.__pending_notifications.popleft() + + for listener in listeners: + try: + listener(state) + except Exception: + log.exception('FB Python SDK: Update status listener failed') + + def add_listener(self, listener: Callable[[State], None]): + if not callable(listener): + return + with self.__lock: + try: + if listener not in self.__listeners: + self.__listeners.append(listener) + except Exception: + log.exception('FB Python SDK: Could not add update status listener') + + def remove_listener(self, listener: Callable[[State], None]): + with self.__lock: + try: + self.__listeners.remove(listener) + except ValueError: + pass + except Exception: + log.exception('FB Python SDK: Could not remove update status listener') def wait_for_OKState(self, timeout: float = 0) -> bool: _timeout = 0 if timeout is None or timeout <= 0 else timeout diff --git a/fbclient/streaming.py b/fbclient/streaming.py index ba7081c..b355838 100644 --- a/fbclient/streaming.py +++ b/fbclient/streaming.py @@ -1,6 +1,5 @@ import json -from threading import Event, Thread -from time import sleep +from threading import Event, Thread, current_thread from typing import Optional, Tuple import websocket @@ -81,7 +80,7 @@ class Streaming(Thread, UpdateProcessor): __ping_interval = 10.0 def __init__(self, config: Config, broadcaster: NoticeBroadcater, dataUpdateStatusProvider: DataUpdateStatusProviderImpl, ready: Event): - super().__init__(daemon=True) + super().__init__(name='featbit-streaming', daemon=True) self.__config = config self.__broadcaster = broadcaster self.__storage = dataUpdateStatusProvider @@ -92,9 +91,10 @@ def __init__(self, config: Config, broadcaster: NoticeBroadcater, dataUpdateStat self.__self_closed = _SelfClosed() self.__closed_by_error = False self.__force_close = False + self.__stop_event = Event() self.__has_network = not config.is_offline if self.__has_network: - self.__ping_task = RepeatableTask('streaming ping', self.__ping_interval, self._on_ping) + self.__ping_task = RepeatableTask('featbit-streaming-ping', self.__ping_interval, self._on_ping) self.__ping_task.start() def _init_wsapp(self): @@ -131,7 +131,9 @@ def run(self): if self.__running: # calculate the delay for reconn delay = self.__strategy.next_delay() - sleep(delay) + # An Event-backed wait lets ``stop()`` interrupt a long + # exponential-backoff delay immediately. + self.__stop_event.wait(delay) except Exception as e: log.exception('FB Python SDK: Streaming unexpected error: %s', str(e)) self.__storage.update_state(State.error_off_state(UNKNOWN_ERROR, str(e))) @@ -261,11 +263,29 @@ def _on_message(self, wsapp: websocket.WebSocketApp, msg): def stop(self): log.info('FB Python SDK: Streaming is stopping...') self.__force_close = True - if self.__running and self.__wsapp: - self.__self_closed = _SelfClosed(is_self_close=True, is_reconn=False, state=State.normal_off_state()) - self.__wsapp.close(status=WS_NORMAL_CLOSE) + self.__running = False + self.__stop_event.set() + try: + self.__storage.update_state(State.normal_off_state()) + except Exception: + log.exception('FB Python SDK: could not publish streaming shutdown state') + try: + if self.__wsapp: + self.__self_closed = _SelfClosed(is_self_close=True, + is_reconn=False, + state=State.normal_off_state()) + self.__wsapp.close(status=WS_NORMAL_CLOSE) + except Exception: + log.exception('FB Python SDK: could not close the WebSocket connection') if self.__has_network: - self.__ping_task.stop() + try: + self.__ping_task.stop() + except Exception: + log.exception('FB Python SDK: could not stop the streaming ping task') + if current_thread() is not self and self.is_alive(): + self.join(self.__config.websocket.timeout + 1.0) + if self.is_alive(): + log.warning('FB Python SDK: streaming thread did not stop in time') @property def initialized(self) -> bool: diff --git a/fbclient/utils/http_client.py b/fbclient/utils/http_client.py index e7be001..07215eb 100644 --- a/fbclient/utils/http_client.py +++ b/fbclient/utils/http_client.py @@ -11,8 +11,10 @@ from fbclient.utils import build_headers, log -def build_http_factory(config: Config, headers={}): - return HTTPFactory(build_headers(config.env_secret, headers), config.http) +def build_http_factory(config: Config, headers=None): + return HTTPFactory(build_headers(config.env_secret, + headers if headers is not None else {}), + config.http) class HTTPFactory: diff --git a/fbclient/utils/repeatable_task.py b/fbclient/utils/repeatable_task.py index a878989..6931a53 100644 --- a/fbclient/utils/repeatable_task.py +++ b/fbclient/utils/repeatable_task.py @@ -3,7 +3,7 @@ base in https://medium.com/greedygame-engineering/an-elegant-way-to-run-periodic-tasks-in-python-61b7c477b679 """ -from threading import Event, Thread +from threading import Event, Thread, current_thread from time import time from typing import Callable @@ -16,17 +16,24 @@ def __init__(self, name: str, interval: float, callable: Callable, args=(), kwar super().__init__(name=name, daemon=True) self._interval = interval self._callable = callable - self._stop = Event() + # ``Thread`` already has a private ``_stop()`` method which is used by + # ``join()``. Keeping an Event under that name makes joining the task + # fail with ``TypeError: 'Event' object is not callable``. + self._stop_event = Event() self._args = args self._kwargs = {} if kwargs is None else kwargs - def stop(self): + def stop(self, timeout: float = 5.0): log.info("FB Python SDK: %s repeatable task is stopping..." % self.name) - self._stop.set() + self._stop_event.set() + if current_thread() is not self and self.is_alive(): + self.join(timeout) + if self.is_alive(): + log.warning("FB Python SDK: %s repeatable task did not stop in time" % self.name) def run(self): log.debug("%s repeatable task is starting..." % self.name) - stopped = self._stop.is_set() + stopped = self._stop_event.is_set() while not stopped: next_time = time() + self._interval try: @@ -34,4 +41,4 @@ def run(self): except Exception as e: log.exception("FB Python SDK: unexpected exception on %s repeatable task: %s" % (self.name, str(e))) delay = next_time - time() - stopped = self._stop.wait(delay) if delay > 0 else self._stop.is_set() + stopped = self._stop_event.wait(delay) if delay > 0 else self._stop_event.is_set() diff --git a/release/description.md b/release/description.md index e87e774..69e6ab3 100644 --- a/release/description.md +++ b/release/description.md @@ -1,5 +1,5 @@ -version: 1.1.7 +version: 1.1.8 ## Break changes @@ -7,8 +7,13 @@ No Break changes ## New features -No new features +- expose the stable variation ID in evaluation details +- add data-update status change listeners ## Updates -- handle python versions 3.12.x \ No newline at end of file +- handle Python versions 3.12.x +- make client, WebSocket, event, and notice shutdown deterministic and idempotent +- isolate runtime event and shutdown failures from application code +- add concurrency, memory-retention, thread-lifecycle, and live-service audit tools +- exclude the test package from production wheels diff --git a/release/package.json b/release/package.json index a478856..8f7d62d 100644 --- a/release/package.json +++ b/release/package.json @@ -1,3 +1,3 @@ { - "version":"1.1.7" -} \ No newline at end of file + "version":"1.1.8" +} diff --git a/scripts/live_integration_check.py b/scripts/live_integration_check.py new file mode 100644 index 0000000..d307373 --- /dev/null +++ b/scripts/live_integration_check.py @@ -0,0 +1,111 @@ +"""Validate this checkout against a real FeatBit evaluation service. + +The environment secret is read only from the process environment and is never +printed or persisted by this script. +""" + +import json +import os +import sys +import threading +import time +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from fbclient.client import FBClient # noqa: E402 +from fbclient.config import Config # noqa: E402 +from fbclient.status_types import StateType # noqa: E402 + + +def sdk_threads(): + return sorted( + thread.name for thread in threading.enumerate() + if thread.name.startswith("featbit-") + ) + + +def main(): + env_secret = os.environ.get("FEATBIT_ENV_SECRET") + if not env_secret: + raise SystemExit("FEATBIT_ENV_SECRET is required") + + streaming_url = os.environ.get( + "FEATBIT_STREAMING_URL", "wss://app-eval.featbit.co" + ) + event_url = os.environ.get( + "FEATBIT_EVENT_URL", "https://app-eval.featbit.co" + ) + flag_key = os.environ.get("FEATBIT_FLAG_KEY", "python-app-release") + timeout = float(os.environ.get("FEATBIT_START_WAIT_SECONDS", "15")) + states = [] + baseline_threads = sdk_threads() + + client = FBClient(Config(env_secret, + event_url=event_url, + streaming_url=streaming_url), + start_wait=timeout) + + def on_state_change(state): + states.append(state.state_type.name) + + client.update_status_provider.add_listener(on_state_change) + try: + ready = client.initialize or client.update_status_provider.wait_for_OKState( + timeout=timeout + ) + if not ready: + state = client.update_status_provider.current_state + error = state.error_track + raise SystemExit( + "FeatBit SDK did not become ready: %s%s" % ( + state.state_type.name, + " (%s)" % error.error_type if error else "", + ) + ) + + evaluations = [] + for index, plan in enumerate(("standard", "pro", "enterprise")): + user_key = "python-sdk-live-%s" % index + user = {"key": user_key, "name": user_key, "plan": plan} + detail = client.variation_detail(flag_key, user, False) + evaluations.append({ + "user_key": user_key, + "variation": detail.variation, + "variation_id": detail.variation_id, + "reason": detail.reason, + }) + client.identify(user) + client.track_metric(user, "python-sdk-live-check", 1.0) + + client.flush() + # Event delivery is asynchronous; stop() performs the final synchronous + # drain, while this short interval also exercises explicit flush(). + time.sleep(1.0) + before_close = client.update_status_provider.current_state.state_type + finally: + client.update_status_provider.remove_listener(on_state_change) + client.stop() + client.stop() + + leaked_threads = [ + name for name in sdk_threads() if name not in baseline_threads + ] + result = { + "ready": ready, + "state_before_close": before_close.name, + "state_after_close": client.update_status_provider.current_state.state_type.name, + "observed_state_changes": states, + "flag_key": flag_key, + "evaluations": evaluations, + "events_exercised": ["feature_flag", "identify", "custom_metric", "flush"], + "leaked_sdk_threads": leaked_threads, + } + print(json.dumps(result, indent=2, sort_keys=True, default=str)) + if before_close != StateType.OK or leaked_threads: + raise SystemExit(1) + + +if __name__ == "__main__": + main() diff --git a/scripts/resource_audit.py b/scripts/resource_audit.py new file mode 100644 index 0000000..b643a60 --- /dev/null +++ b/scripts/resource_audit.py @@ -0,0 +1,110 @@ +"""Repeatable concurrency, memory-retention, and thread-lifecycle audit.""" + +import argparse +import gc +import json +import sys +import threading +import time +import tracemalloc +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from fbclient.client import FBClient # noqa: E402 +from fbclient.config import Config # noqa: E402 + + +def sdk_thread_ids(): + return { + thread.ident for thread in threading.enumerate() + if thread.name.startswith("featbit-") + } + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--evaluations", type=int, default=80000) + parser.add_argument("--workers", type=int, default=8) + parser.add_argument("--max-retained-bytes", type=int, default=2 * 1024 * 1024) + args = parser.parse_args() + + baseline_threads = sdk_thread_ids() + config = Config("resource-audit", + event_url="http://offline", + streaming_url="ws://offline", + offline=True) + client = FBClient(config) + bootstrap = (ROOT / "tests" / "fbclient_test_data.json").read_text() + if not client.initialize_from_external_json(bootstrap): + raise SystemExit("could not initialize offline audit data") + + user = {"key": "warmup", "name": "Warmup"} + for _ in range(2000): + client.variation("ff-test-bool", user, False) + + gc.collect() + tracemalloc.start() + baseline_bytes = tracemalloc.get_traced_memory()[0] + errors = [] + per_worker = max(1, args.evaluations // args.workers) + + def evaluate(worker): + completed = 0 + try: + for index in range(per_worker): + key = "audit-%s-%s" % (worker, index) + value = client.variation( + "ff-test-bool", + {"key": key, "name": key}, + False, + ) + if not isinstance(value, bool): + raise AssertionError("evaluation returned a non-boolean value") + completed += 1 + except Exception as error: + errors.append("%s: %s" % (type(error).__name__, error)) + return completed + + started = time.perf_counter() + with ThreadPoolExecutor(max_workers=args.workers) as executor: + completed = sum(executor.map(evaluate, range(args.workers))) + elapsed = time.perf_counter() - started + client.stop() + client.stop() + gc.collect() + final_bytes = tracemalloc.get_traced_memory()[0] + tracemalloc.stop() + + leaked_threads = sorted( + thread.name for thread in threading.enumerate() + if thread.name.startswith("featbit-") + and thread.ident not in baseline_threads + ) + retained_bytes = max(0, final_bytes - baseline_bytes) + result = { + "evaluations": completed, + "workers": args.workers, + "seconds": round(elapsed, 4), + "evaluations_per_second": round(completed / elapsed) if elapsed else 0, + "concurrent_errors": errors, + "leaked_sdk_threads": leaked_threads, + "baseline_traced_bytes": baseline_bytes, + "final_traced_bytes": final_bytes, + "retained_traced_bytes": retained_bytes, + "allowed_retained_bytes": args.max_retained_bytes, + } + result["status"] = "ok" if ( + not errors + and not leaked_threads + and retained_bytes <= args.max_retained_bytes + ) else "failed" + print(json.dumps(result, indent=2, sort_keys=True)) + if result["status"] != "ok": + raise SystemExit(1) + + +if __name__ == "__main__": + main() diff --git a/setup.py b/setup.py index 91749b9..cf74fd7 100644 --- a/setup.py +++ b/setup.py @@ -27,7 +27,7 @@ def parse_requirements(filename): version=fb_version, author='Dian SUN', author_email='featbit.master@gmail.com', - packages=find_packages(), + packages=find_packages(exclude=('tests', 'tests.*')), url='https://github.com/featbit/featbit-python-sdk', project_urls={ 'Code': 'https://github.com/featbit/featbit-python-sdk', @@ -55,6 +55,5 @@ def parse_requirements(filename): extras_require={ "dev": dev_reqs }, - tests_require=dev_reqs, python_requires='>=3.6, <3.13' ) diff --git a/tests/test_data_update_status_provider.py b/tests/test_data_update_status_provider.py index 912191b..01d4e29 100644 --- a/tests/test_data_update_status_provider.py +++ b/tests/test_data_update_status_provider.py @@ -93,6 +93,72 @@ def test_update_state(data_updator): assert data_updator.current_state.state_type == StateType.INTERRUPTED +def test_status_listener_receives_changes_and_can_be_removed(data_updator): + received = [] + + def listener(state): + received.append(state.state_type) + + data_updator.add_listener(listener) + data_updator.add_listener(listener) + data_updator.add_listener(None) + data_updator.remove_listener(lambda _state: None) + data_updator.update_state(State.ok_state()) + data_updator.update_state(State.interrupted_state("network", "disconnected")) + data_updator.remove_listener(listener) + data_updator.update_state(State.ok_state()) + + assert received == [StateType.OK, StateType.INTERRUPTED] + + +def test_status_listener_exception_does_not_escape_or_block_others(data_updator): + received = [] + + def failing_listener(_state): + raise RuntimeError("listener failure") + + data_updator.add_listener(failing_listener) + data_updator.add_listener(lambda state: received.append(state.state_type)) + + data_updator.update_state(State.ok_state()) + + assert received == [StateType.OK] + assert data_updator.current_state.state_type == StateType.OK + + +def test_status_listener_preserves_concurrent_transition_order(data_updator): + first_callback_started = threading.Event() + release_first_callback = threading.Event() + received = [] + + def listener(state): + if state.state_type == StateType.OK: + first_callback_started.set() + assert release_first_callback.wait(1) + received.append(state.state_type) + + data_updator.add_listener(listener) + first_update = threading.Thread( + target=data_updator.update_state, + args=(State.ok_state(),) + ) + second_update = threading.Thread( + target=data_updator.update_state, + args=(State.interrupted_state("network", "disconnected"),) + ) + + first_update.start() + assert first_callback_started.wait(1) + second_update.start() + second_update.join(1) + release_first_callback.set() + first_update.join(1) + + assert not first_update.is_alive() + assert not second_update.is_alive() + assert received == [StateType.OK, StateType.INTERRUPTED] + + def test_wait_for_OKState(data_updator): assert not data_updator.wait_for_OKState(timeout=0.1) data_updator.update_state(State.ok_state()) diff --git a/tests/test_fbclient.py b/tests/test_fbclient.py index 5ff30fc..5146069 100644 --- a/tests/test_fbclient.py +++ b/tests/test_fbclient.py @@ -6,6 +6,7 @@ import pytest from fbclient.client import FBClient +from fbclient.common_types import EvalDetail from fbclient.config import Config from fbclient.data_storage import InMemoryDataStorage from fbclient.evaluator import (REASON_CLIENT_NOT_READY, REASON_ERROR, @@ -28,6 +29,17 @@ USER_EMAIL = {"key": "test-user-7@featbit.com", "name": "test-user-7"} +def test_eval_detail_variation_id_is_backward_compatible(): + legacy_detail = EvalDetail("test reason", True, "flag-key", "Flag name") + assert legacy_detail.variation_id is None + + detail_with_id = EvalDetail( + "test reason", True, "flag-key", "Flag name", "variation-id" + ) + assert detail_with_id.variation_id == "variation-id" + assert detail_with_id.to_json_dict() == legacy_detail.to_json_dict() + + def make_fb_client(update_processor_imp, event_processor_imp, start_wait=15.): config = Config(FAKE_ENV_SECRET, event_url=FAKE_URL, @@ -118,6 +130,7 @@ def start(): detail = client.variation_detail("ff-test-bool", USER_1, False) assert detail.variation is False assert detail.reason == REASON_CLIENT_NOT_READY + assert detail.variation_id is None all_states = client.get_all_latest_flag_variations(USER_1) # type: ignore assert not all_states.success assert all_states.reason == REASON_CLIENT_NOT_READY @@ -133,6 +146,7 @@ def test_bool_variation(): detail = client.variation_detail("ff-test-bool", USER_2, False) assert detail.variation is True assert detail.reason == REASON_TARGET_MATCH + assert detail.variation_id == "18b369f8-453f-46d7-88cc-fe41d29ca6e3" assert client.variation("ff-test-bool", USER_3, False) is False detail = client.variation_detail("ff-test-bool", USER_4, False) assert detail.variation is True @@ -238,6 +252,7 @@ def test_variation_argument_error(): detail = client.variation_detail("ff-not-existed", USER_1, False) assert detail.variation is False assert detail.reason == REASON_FLAG_NOT_FOUND + assert detail.variation_id is None detail = client.variation_detail("ff-test-bool", None, None) # type: ignore assert detail.variation is None assert detail.reason == REASON_USER_NOT_SPECIFIED @@ -265,7 +280,13 @@ def test_variation_error_default_value(): now = datetime.utcnow() with make_fb_client_offline() as client: assert client.initialize - with pytest.raises(ValueError): - client.variation_detail("ff-test-bool", USER_1, now) - with pytest.raises(ValueError): - client.variation("ff-test-bool", USER_1, now) + # Runtime evaluation APIs must not raise into application code merely + # because a fallback has an unsupported type. + assert client.variation("ff-test-bool", USER_1, now) is True + assert client.variation("ff-not-existed", USER_1, now) is now + assert client.variation_detail("ff-not-existed", USER_1, now).variation is now + + +def test_invalid_external_json_does_not_raise(): + with make_fb_client_offline() as client: + assert client.initialize_from_external_json("{") is False diff --git a/tests/test_runtime_safety.py b/tests/test_runtime_safety.py new file mode 100644 index 0000000..39a52d6 --- /dev/null +++ b/tests/test_runtime_safety.py @@ -0,0 +1,160 @@ +import base64 +import threading +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +from fbclient.client import FBClient +from fbclient.config import Config +from fbclient.data_storage import InMemoryDataStorage +from fbclient.notice_broadcaster import NoticeBroadcater +from fbclient.status import DataUpdateStatusProviderImpl +from fbclient.streaming import Streaming +from fbclient.update_processor import NullUpdateProcessor +from fbclient.utils.repeatable_task import RepeatableTask + + +FAKE_ENV_SECRET = base64.b64encode(b"runtime-safety").decode() +FAKE_URL = "http://fake" +USER = {"key": "runtime-user", "name": "Runtime User"} + + +def make_offline_client(): + client = FBClient(Config(FAKE_ENV_SECRET, + event_url=FAKE_URL, + streaming_url=FAKE_URL, + offline=True)) + bootstrap = Path("tests/fbclient_test_data.json").read_text() + assert client.initialize_from_external_json(bootstrap) + return client + + +def featbit_thread_ids(): + return { + thread.ident for thread in threading.enumerate() + if thread.name.startswith("featbit-") + } + + +def test_concurrent_evaluation_is_thread_safe(): + client = make_offline_client() + + def evaluate(worker): + for index in range(1000): + user = { + "key": "worker-%s-%s" % (worker, index), + "name": "Worker %s" % worker, + } + assert isinstance(client.variation("ff-test-bool", user, False), bool) + return 1000 + + try: + with ThreadPoolExecutor(max_workers=8) as executor: + assert sum(executor.map(evaluate, range(8))) == 8000 + finally: + client.stop() + + +def test_repeated_clients_release_all_sdk_threads(): + baseline = featbit_thread_ids() + for _ in range(10): + client = make_offline_client() + client.stop() + client.stop() + assert featbit_thread_ids() == baseline + + +def test_repeatable_task_can_be_joined_cleanly(): + task = RepeatableTask("featbit-test-repeatable", 0.01, lambda: None) + task.start() + task.stop() + assert not task.is_alive() + + +def test_notice_broadcaster_is_safe_during_listener_churn(): + broadcaster = NoticeBroadcater() + callback_count = [0] + callback_lock = threading.Lock() + + class Notice: + notice_type = "test" + + def listener(_notice): + with callback_lock: + callback_count[0] += 1 + + def churn(_worker): + for _ in range(200): + broadcaster.add_listener("test", listener) + broadcaster.broadcast(Notice()) + broadcaster.remove_listener("test", listener) + + with ThreadPoolExecutor(max_workers=8) as executor: + list(executor.map(churn, range(8))) + broadcaster.stop() + broadcaster.stop() + assert callback_count[0] >= 0 + + +def test_client_public_event_and_shutdown_calls_do_not_raise(): + class FailingEventProcessor: + def send_event(self, _event): + raise RuntimeError("send failed") + + def flush(self): + raise RuntimeError("flush failed") + + def stop(self): + raise RuntimeError("stop failed") + + def build_event_processor(_config, _sender): + return FailingEventProcessor() + + config = Config(FAKE_ENV_SECRET, + event_url=FAKE_URL, + streaming_url=FAKE_URL, + update_processor_imp=NullUpdateProcessor, + event_processor_imp=build_event_processor) + client = FBClient(config) + client.identify(USER) + client.track_metric(USER, "metric") + client.track_metrics(USER, {"metric": 1.0}) + client.flush() + client.stop() + client.stop() + + +def test_streaming_stop_interrupts_network_wait(monkeypatch): + connected = threading.Event() + + class FakeWebSocketApp: + def __init__(self, _url, **_kwargs): + self.closed = threading.Event() + self.sock = None + + def run_forever(self, **_kwargs): + connected.set() + self.closed.wait(10.0) + + def close(self, status=None): + self.closed.set() + + monkeypatch.setattr("fbclient.streaming.websocket.WebSocketApp", + FakeWebSocketApp) + config = Config(FAKE_ENV_SECRET, + event_url=FAKE_URL, + streaming_url=FAKE_URL) + broadcaster = NoticeBroadcater() + status = DataUpdateStatusProviderImpl(InMemoryDataStorage()) + streaming = Streaming(config, broadcaster, status, threading.Event()) + streaming.start() + assert connected.wait(1.0) + streaming.stop() + broadcaster.stop() + assert not streaming.is_alive() + + +def test_default_config_objects_are_not_shared(): + left = Config(FAKE_ENV_SECRET, FAKE_URL, FAKE_URL) + right = Config(FAKE_ENV_SECRET, FAKE_URL, FAKE_URL) + assert left.http is not right.http + assert left.websocket is not right.websocket