add tests

Author: Tymek

tests/integration_tests/integration_unleash.py

@@ -0,0 +1,35 @@
+# ---
+import os
+import logging
+import sys
+import time
+
+from UnleashClient import UnleashClient
+
+root = logging.getLogger()
+root.setLevel(logging.DEBUG)
+
+handler = logging.StreamHandler(sys.stdout)
+handler.setLevel(logging.DEBUG)
+formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+handler.setFormatter(formatter)
+root.addHandler(handler)
+# ---
+
+api_url = os.getenv('UNLEASH_API_URL', 'https://app.unleash-hosted.com/demo/api')
+api_token = os.getenv('UNLEASH_API_TOKEN', 'demo-app:dev.9fc74dd72d2b88bea5253c04240b21a54841f08d9918046ed55a06b5')
+flag = "example-flag"
+use_streaming = os.getenv("USE_STREAMING", "true").lower() == "true"
+
+client = UnleashClient(
+    url=api_url,
+    app_name="integration-python",
+    custom_headers={'Authorization': api_token},
+    experimental_mode={"type": "streaming"} if use_streaming else None,
+    metrics_interval=1)
+
+client.initialize_client()
+
+while True:
+    print(f"'{flag}' is enabled: {client.is_enabled(flag)}")
+    time.sleep(3)

tests/integration_tests/integration_unleashheroku.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from UnleashClient.streaming.event_processor import StreamingEventProcessor
+
+
+class FakeEngine:
+    def __init__(self):
+        self.states = []
+
+    def take_state(self, state):
+        self.states.append(state)
+
+
+class FakeEvent:
+    def __init__(self, event: str, data):
+        self.event = event
+        self.data = data
+
+
+def test_processor_hydrates_on_connected():
+    engine = FakeEngine()
+    processor = StreamingEventProcessor(engine)
+
+    assert processor.hydrated is False
+
+    payload = {"version": 1, "features": [], "segments": []}
+    processor.process(FakeEvent("unleash-connected", payload))
+
+    assert processor.hydrated is True
+    assert engine.states == [payload]
+
+
+def test_processor_applies_updates():
+    engine = FakeEngine()
+    processor = StreamingEventProcessor(engine)
+
+    connected_payload = {"version": 1, "features": ["f1"], "segments": []}
+    update_payload = {"version": 2, "features": ["f1", "f2"], "segments": []}
+
+    processor.process(FakeEvent("unleash-connected", connected_payload))
+    processor.process(FakeEvent("unleash-updated", update_payload))
+
+    assert processor.hydrated is True
+    assert engine.states == [connected_payload, update_payload]
+
+
+def test_processor_ignores_unknown_event_types():
+    engine = FakeEngine()
+    processor = StreamingEventProcessor(engine)
+
+    processor.process(FakeEvent("heartbeat", {}))
+    processor.process(FakeEvent("message", {}))
+
+    # No states should be applied
+    assert engine.states == []

tests/unit_tests/streaming/test_event_processor.py

@@ -0,0 +1,123 @@
+from __future__ import annotations
+
+import json
+import threading
+import time
+from typing import Iterable
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from UnleashClient.streaming import StreamingConnector
+from UnleashClient.streaming.event_processor import StreamingEventProcessor
+
+
+class FakeEngine:
+    def __init__(self):
+        self.states = []
+
+    def take_state(self, state):
+        self.states.append(state)
+
+
+class FakeEvent:
+    def __init__(self, event: str, data):
+        self.event = event
+        self.data = data
+
+
+class FiniteSSEClient:
+    """SSE client that yields given events then stops."""
+
+    def __init__(self, events: Iterable[FakeEvent]):
+        self._events = list(events)
+        self.closed = False
+
+    @property
+    def events(self):
+        for e in self._events:
+            if self.closed:
+                break
+            yield e
+
+    def close(self):
+        self.closed = True
+
+    def interrupt(self):
+        self.close()
+
+
+class FailingSSEClient:
+    """SSE client that fails immediately when iterating events."""
+
+    def __init__(self):
+        self.closed = False
+
+    @property
+    def events(self):
+        raise ConnectionError("Simulated connection failure")
+
+    def close(self):
+        self.closed = True
+
+    def interrupt(self):
+        self.close()
+
+def test_successful_connection_calls_ready():
+    engine = FakeEngine()
+    processor = StreamingEventProcessor(engine)
+
+    ready_calls = {"n": 0}
+
+    def on_ready():
+        ready_calls["n"] += 1
+
+    events = [
+        FakeEvent("unleash-connected", {"version": 1, "features": [], "segments": []}),
+        FakeEvent("unleash-updated", {"version": 2, "features": [], "segments": []}),
+    ]
+
+    controller = StreamingConnector(
+        url="http://unleash.example",
+        headers={},
+        request_timeout=5,
+        event_processor=processor,
+        on_ready=on_ready,
+        sse_client_factory=lambda url, headers, timeout: FiniteSSEClient(events),
+    )
+
+    th = threading.Thread(target=controller._run, daemon=True)
+    th.start()
+    time.sleep(0.05)
+    controller.stop()
+    th.join(timeout=1)
+
+    assert engine.states  # at least one state applied
+    assert ready_calls["n"] == 1
+
+
+def test_connection_failures_trigger_retries():
+    engine = FakeEngine()
+    processor = StreamingEventProcessor(engine)
+
+    attempts = []
+
+    def factory(url, headers, timeout):
+        attempts.append(time.time())
+        return FailingSSEClient()
+
+    controller = StreamingConnector(
+        url="http://unleash.example",
+        headers={},
+        request_timeout=5,
+        event_processor=processor,
+        sse_client_factory=factory,
+    )
+
+    th = threading.Thread(target=controller._run, daemon=True)
+    th.start()
+    time.sleep(1.5)
+    controller.stop()
+    th.join(timeout=1)
+
+    assert len(attempts) >= 1