tests/test_analytics.py

"""
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()