""" tests/test_analytics.py — Unit tests for backend/analytics.py Run with: python3 -m pytest tests/test_analytics.py -q """ import importlib import os import sys import threading import time import types import unittest.mock as mock import pytest # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _reload_analytics(monkeypatch, api_key="", posthog_module=None): """ Reload analytics with a clean global state and optionally inject a fake posthog module into sys.modules. """ # Clear cached state if "backend.analytics" in sys.modules: del sys.modules["backend.analytics"] if posthog_module is not None: monkeypatch.setitem(sys.modules, "posthog", posthog_module) elif "posthog" in sys.modules: monkeypatch.delitem(sys.modules, "posthog", raising=False) if api_key: monkeypatch.setenv("POSTHOG_API_KEY", api_key) else: monkeypatch.delenv("POSTHOG_API_KEY", raising=False) import backend.analytics as analytics return analytics def _make_fake_posthog(): """Return a mock posthog module with capture/identify spies.""" ph = types.ModuleType("posthog") ph.api_key = None ph.host = None ph.sync_mode = None ph.capture = mock.MagicMock() ph.identify = mock.MagicMock() return ph def _wait_threads(timeout=2.0): """Wait for all non-main daemon threads spawned by analytics to finish.""" deadline = time.monotonic() + timeout while time.monotonic() < deadline: alive = [ t for t in threading.enumerate() if t.daemon and t is not threading.current_thread() ] if not alive: break time.sleep(0.05) # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- class TestTrackNoop: """track() must be a no-op when POSTHOG_API_KEY is not configured.""" def test_track_no_key_does_not_raise(self, monkeypatch): analytics = _reload_analytics(monkeypatch, api_key="") # Should not raise analytics.track("test_event", user_id="user-1") def test_track_no_key_returns_none(self, monkeypatch): analytics = _reload_analytics(monkeypatch, api_key="") result = analytics.track("test_event", user_id="user-1", properties={"k": "v"}) assert result is None def test_identify_no_key_does_not_raise(self, monkeypatch): analytics = _reload_analytics(monkeypatch, api_key="") analytics.identify("user-1", {"email": "test@example.com"}) def test_track_analysis_no_key_does_not_raise(self, monkeypatch): analytics = _reload_analytics(monkeypatch, api_key="") analytics.track_analysis("user-1", mode="auto", instrument_count=3, duration_s=4.2) def test_track_subscription_no_key_does_not_raise(self, monkeypatch): analytics = _reload_analytics(monkeypatch, api_key="") analytics.track_subscription("user-1", plan="pro", action="checkout") def test_track_error_no_key_does_not_raise(self, monkeypatch): analytics = _reload_analytics(monkeypatch, api_key="") analytics.track_error("user-1", error_type="ValueError", endpoint="/api/analyze") class TestTrackWithPosthog: """track() must call posthog.capture when configured.""" def test_track_calls_posthog_capture(self, monkeypatch): ph = _make_fake_posthog() analytics = _reload_analytics(monkeypatch, api_key="ph_test_key", posthog_module=ph) analytics.track("my_event", user_id="u-42", properties={"foo": "bar"}) _wait_threads() ph.capture.assert_called_once() call_kwargs = ph.capture.call_args assert call_kwargs.kwargs.get("event") == "my_event" or ( len(call_kwargs.args) >= 2 and call_kwargs.args[1] == "my_event" ) def test_identify_calls_posthog_identify(self, monkeypatch): ph = _make_fake_posthog() analytics = _reload_analytics(monkeypatch, api_key="ph_test_key", posthog_module=ph) analytics.identify("u-99", {"email": "user@example.com", "plan": "pro"}) _wait_threads() ph.identify.assert_called_once() call_kwargs = ph.identify.call_args # Verify user_id forwarded correctly distinct_id = ( call_kwargs.kwargs.get("distinct_id") or (call_kwargs.args[0] if call_kwargs.args else None) ) assert str(distinct_id) == "u-99" def test_track_analysis_sends_correct_event_name(self, monkeypatch): ph = _make_fake_posthog() analytics = _reload_analytics(monkeypatch, api_key="ph_test_key", posthog_module=ph) analytics.track_analysis("u-7", mode="ml", instrument_count=2, duration_s=3.5) _wait_threads() ph.capture.assert_called_once() call_kwargs = ph.capture.call_args event_name = ( call_kwargs.kwargs.get("event") or (call_kwargs.args[1] if len(call_kwargs.args) >= 2 else None) ) assert event_name == "analysis_completed" def test_track_analysis_sends_correct_properties(self, monkeypatch): ph = _make_fake_posthog() analytics = _reload_analytics(monkeypatch, api_key="ph_test_key", posthog_module=ph) analytics.track_analysis("u-7", mode="offline", instrument_count=5, duration_s=10.0) _wait_threads() call_kwargs = ph.capture.call_args props = ( call_kwargs.kwargs.get("properties") or (call_kwargs.args[2] if len(call_kwargs.args) >= 3 else {}) ) assert props.get("mode") == "offline" assert props.get("instrument_count") == 5 assert props.get("duration_s") == 10.0 def test_track_subscription_event_name_and_props(self, monkeypatch): ph = _make_fake_posthog() analytics = _reload_analytics(monkeypatch, api_key="ph_test_key", posthog_module=ph) analytics.track_subscription("u-3", plan="starter", action="cancel") _wait_threads() ph.capture.assert_called_once() call_kwargs = ph.capture.call_args event_name = ( call_kwargs.kwargs.get("event") or (call_kwargs.args[1] if len(call_kwargs.args) >= 2 else None) ) assert event_name == "subscription_event" props = ( call_kwargs.kwargs.get("properties") or (call_kwargs.args[2] if len(call_kwargs.args) >= 3 else {}) ) assert props.get("plan") == "starter" assert props.get("action") == "cancel" def test_track_error_event_name_and_props(self, monkeypatch): ph = _make_fake_posthog() analytics = _reload_analytics(monkeypatch, api_key="ph_test_key", posthog_module=ph) analytics.track_error("u-5", error_type="HTTPException", endpoint="/api/analyze") _wait_threads() ph.capture.assert_called_once() call_kwargs = ph.capture.call_args props = ( call_kwargs.kwargs.get("properties") or (call_kwargs.args[2] if len(call_kwargs.args) >= 3 else {}) ) assert props.get("error_type") == "HTTPException" assert props.get("endpoint") == "/api/analyze" class TestNeverRaises: """All public functions must swallow every exception.""" def test_track_with_none_user_id(self, monkeypatch): analytics = _reload_analytics(monkeypatch, api_key="") analytics.track("event", user_id=None) # type: ignore[arg-type] def test_track_analysis_with_bad_inputs(self, monkeypatch): analytics = _reload_analytics(monkeypatch, api_key="") analytics.track_analysis(None, mode=None, instrument_count=None, duration_s=None) # type: ignore[arg-type] def test_identify_with_none_properties(self, monkeypatch): analytics = _reload_analytics(monkeypatch, api_key="") analytics.identify("u-1", None) # type: ignore[arg-type] def test_track_with_posthog_raising(self, monkeypatch): """Even when posthog.capture raises, track() must not propagate.""" ph = _make_fake_posthog() ph.capture.side_effect = RuntimeError("network failure") analytics = _reload_analytics(monkeypatch, api_key="ph_test_key", posthog_module=ph) # Should not raise analytics.track("event", user_id="u-1") _wait_threads() def test_identify_with_posthog_raising(self, monkeypatch): ph = _make_fake_posthog() ph.identify.side_effect = Exception("timeout") analytics = _reload_analytics(monkeypatch, api_key="ph_test_key", posthog_module=ph) analytics.identify("u-1", {"email": "test@example.com"}) _wait_threads()