mem0/tests/test_telemetry.py

import threading
from unittest.mock import MagicMock, patch

import pytest

import mem0.memory.telemetry as telemetry_module


class TestTelemetryDisabled:
    """Verify PostHog is never instantiated when telemetry is disabled."""

    def test_posthog_not_created_when_disabled(self):
        """Posthog() constructor should never be called when MEM0_TELEMETRY=False."""
        with patch.object(telemetry_module, "MEM0_TELEMETRY", False):
            with patch("mem0.memory.telemetry.Posthog") as mock_posthog:
                at = telemetry_module.AnonymousTelemetry()
                mock_posthog.assert_not_called()
                assert at.posthog is None
                assert at.user_id is None

    def test_capture_event_noop_when_disabled(self):
        """capture_event() should return immediately without creating AnonymousTelemetry."""
        with patch.object(telemetry_module, "MEM0_TELEMETRY", False):
            with patch("mem0.memory.telemetry.AnonymousTelemetry") as mock_cls:
                telemetry_module.capture_event("test.event", MagicMock())
                mock_cls.assert_not_called()

    def test_capture_client_event_noop_when_disabled(self):
        """capture_client_event() should return immediately without calling posthog."""
        with patch.object(telemetry_module, "MEM0_TELEMETRY", False):
            mock_instance = MagicMock()
            mock_client_telemetry = MagicMock()
            telemetry_module._client_telemetry = mock_client_telemetry
            telemetry_module.capture_client_event("test.event", mock_instance)
            mock_client_telemetry.capture_event.assert_not_called()
            telemetry_module._client_telemetry = None

    def test_instance_capture_event_noop_when_posthog_is_none(self):
        """AnonymousTelemetry.capture_event() should be a no-op when posthog is None."""
        with patch.object(telemetry_module, "MEM0_TELEMETRY", False):
            at = telemetry_module.AnonymousTelemetry()
            at.capture_event("test.event", {"key": "value"})  # should not raise

    def test_close_noop_when_posthog_is_none(self):
        """close() should not raise when posthog is None."""
        with patch.object(telemetry_module, "MEM0_TELEMETRY", False):
            at = telemetry_module.AnonymousTelemetry()
            at.close()  # should not raise

    def test_no_threads_spawned_when_disabled(self):
        """No consumer threads should be created when telemetry is disabled."""
        with patch.object(telemetry_module, "MEM0_TELEMETRY", False):
            threads_before = threading.active_count()
            telemetry_module.AnonymousTelemetry()
            threads_after = threading.active_count()
            assert threads_after == threads_before


class TestTelemetryEnabled:
    """Verify PostHog works normally when telemetry is enabled."""

    def test_posthog_created_when_enabled(self):
        """Posthog() should be instantiated when MEM0_TELEMETRY=True."""
        with patch.object(telemetry_module, "MEM0_TELEMETRY", True):
            with patch("mem0.memory.telemetry.Posthog") as mock_posthog:
                with patch("mem0.memory.telemetry.get_or_create_user_id", return_value="test-user"):
                    at = telemetry_module.AnonymousTelemetry()
                    mock_posthog.assert_called_once()
                    assert at.posthog is not None
                    assert at.user_id == "test-user"

    def test_capture_event_sends_when_enabled(self):
        """capture_event() should use singleton AnonymousTelemetry and call capture when enabled."""
        with patch.object(telemetry_module, "MEM0_TELEMETRY", True):
            mock_at = MagicMock()
            telemetry_module._oss_telemetry = mock_at
            mock_memory = MagicMock()
            mock_memory.config.graph_store.config = None
            mock_memory.api_version = "v1"
            telemetry_module.capture_event("test.event", mock_memory)
            mock_at.capture_event.assert_called_once()
            telemetry_module._oss_telemetry = None

    def test_capture_client_event_sends_when_enabled(self):
        """capture_client_event() should call client_telemetry.capture_event when enabled."""
        with patch.object(telemetry_module, "MEM0_TELEMETRY", True):
            mock_client_telemetry = MagicMock()
            telemetry_module._client_telemetry = mock_client_telemetry
            mock_instance = MagicMock()
            mock_instance.user_email = "test@example.com"
            telemetry_module.capture_client_event("test.event", mock_instance)
            mock_client_telemetry.capture_event.assert_called_once()
            telemetry_module._client_telemetry = None

    def test_capture_event_reuses_singleton(self):
        """capture_event() must reuse a single AnonymousTelemetry, not create one per call."""
        with patch.object(telemetry_module, "MEM0_TELEMETRY", True):
            with patch("mem0.memory.telemetry.AnonymousTelemetry") as mock_cls:
                mock_at = MagicMock()
                mock_cls.return_value = mock_at
                telemetry_module._oss_telemetry = None
                mock_memory = MagicMock()
                mock_memory.config.graph_store.config = None
                mock_memory.api_version = "v1"

                telemetry_module.capture_event("event1", mock_memory)
                telemetry_module.capture_event("event2", mock_memory)

                assert mock_cls.call_count == 1
                assert mock_at.capture_event.call_count == 2

                telemetry_module._oss_telemetry = None


class TestTelemetryShutdown:
    """Verify shutdown_telemetry() cleans up all singleton instances."""

    def test_shutdown_calls_close_on_singletons(self):
        mock_oss = MagicMock()
        mock_client = MagicMock()
        telemetry_module._oss_telemetry = mock_oss
        telemetry_module._client_telemetry = mock_client

        telemetry_module.shutdown_telemetry()

        mock_oss.close.assert_called_once()
        mock_client.close.assert_called_once()
        assert telemetry_module._oss_telemetry is None
        assert telemetry_module._client_telemetry is None

    def test_shutdown_is_safe_when_no_singletons(self):
        telemetry_module._oss_telemetry = None
        telemetry_module._client_telemetry = None
        telemetry_module.shutdown_telemetry()  # should not raise


class TestNoThreadLeak:
    """Verify capture_event does not leak threads."""

    def test_multiple_capture_events_single_posthog_instance(self):
        with patch.object(telemetry_module, "MEM0_TELEMETRY", True):
            with patch("mem0.memory.telemetry.Posthog") as mock_posthog:
                with patch("mem0.memory.telemetry.get_or_create_user_id", return_value="test"):
                    telemetry_module._oss_telemetry = None

                    mock_memory = MagicMock()
                    mock_memory.config.graph_store.config = None
                    mock_memory.api_version = "v1"

                    for _ in range(10):
                        telemetry_module.capture_event("test.event", mock_memory)

                    assert mock_posthog.call_count == 1

                    telemetry_module._oss_telemetry = None


class TestTelemetryEnvVar:
    """Verify the MEM0_TELEMETRY env var parsing logic."""

    @pytest.mark.parametrize(
        "value,expected",
        [
            ("true", True),
            ("True", True),
            ("TRUE", True),
            ("1", True),
            ("yes", True),
            ("false", False),
            ("False", False),
            ("0", False),
            ("no", False),
            ("anything_else", False),
        ],
    )
    def test_env_var_parsing(self, value, expected):
        result = value.lower() in ("true", "1", "yes")
        assert result == expected