my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

my-project-public — gitlawb

#1	import threading
#2	from unittest.mock import MagicMock, patch
#3
#4	import pytest
#5
#6	import mem0.memory.telemetry as telemetry_module
#7
#8
#9	class TestTelemetryDisabled:
#10	"""Verify PostHog is never instantiated when telemetry is disabled."""
#11
#12	def test_posthog_not_created_when_disabled(self):
#13	"""Posthog() constructor should never be called when MEM0_TELEMETRY=False."""
#14	with patch.object(telemetry_module, "MEM0_TELEMETRY", False):
#15	with patch("mem0.memory.telemetry.Posthog") as mock_posthog:
#16	at = telemetry_module.AnonymousTelemetry()
#17	mock_posthog.assert_not_called()
#18	assert at.posthog is None
#19	assert at.user_id is None
#20
#21	def test_capture_event_noop_when_disabled(self):
#22	"""capture_event() should return immediately without creating AnonymousTelemetry."""
#23	with patch.object(telemetry_module, "MEM0_TELEMETRY", False):
#24	with patch("mem0.memory.telemetry.AnonymousTelemetry") as mock_cls:
#25	telemetry_module.capture_event("test.event", MagicMock())
#26	mock_cls.assert_not_called()
#27
#28	def test_capture_client_event_noop_when_disabled(self):
#29	"""capture_client_event() should return immediately without calling posthog."""
#30	with patch.object(telemetry_module, "MEM0_TELEMETRY", False):
#31	mock_instance = MagicMock()
#32	mock_client_telemetry = MagicMock()
#33	with patch.object(telemetry_module, "client_telemetry", mock_client_telemetry):
#34	telemetry_module.capture_client_event("test.event", mock_instance)
#35	mock_client_telemetry.capture_event.assert_not_called()
#36
#37	def test_instance_capture_event_noop_when_posthog_is_none(self):
#38	"""AnonymousTelemetry.capture_event() should be a no-op when posthog is None."""
#39	with patch.object(telemetry_module, "MEM0_TELEMETRY", False):
#40	at = telemetry_module.AnonymousTelemetry()
#41	at.capture_event("test.event", {"key": "value"}) # should not raise
#42
#43	def test_close_noop_when_posthog_is_none(self):
#44	"""close() should not raise when posthog is None."""
#45	with patch.object(telemetry_module, "MEM0_TELEMETRY", False):
#46	at = telemetry_module.AnonymousTelemetry()
#47	at.close() # should not raise
#48
#49	def test_no_threads_spawned_when_disabled(self):
#50	"""No consumer threads should be created when telemetry is disabled."""
#51	with patch.object(telemetry_module, "MEM0_TELEMETRY", False):
#52	threads_before = threading.active_count()
#53	telemetry_module.AnonymousTelemetry()
#54	threads_after = threading.active_count()
#55	assert threads_after == threads_before
#56
#57
#58	class TestTelemetryEnabled:
#59	"""Verify PostHog works normally when telemetry is enabled."""
#60
#61	def test_posthog_created_when_enabled(self):
#62	"""Posthog() should be instantiated when MEM0_TELEMETRY=True."""
#63	with patch.object(telemetry_module, "MEM0_TELEMETRY", True):
#64	with patch("mem0.memory.telemetry.Posthog") as mock_posthog:
#65	with patch("mem0.memory.telemetry.get_or_create_user_id", return_value="test-user"):
#66	at = telemetry_module.AnonymousTelemetry()
#67	mock_posthog.assert_called_once()
#68	assert at.posthog is not None
#69	assert at.user_id == "test-user"
#70
#71	def test_capture_event_sends_when_enabled(self):
#72	"""capture_event() should create AnonymousTelemetry and call capture when enabled."""
#73	with patch.object(telemetry_module, "MEM0_TELEMETRY", True):
#74	with patch("mem0.memory.telemetry.AnonymousTelemetry") as mock_cls:
#75	mock_at = MagicMock()
#76	mock_cls.return_value = mock_at
#77	mock_memory = MagicMock()
#78	mock_memory.config.graph_store.config = None
#79	mock_memory.api_version = "v1"
#80	telemetry_module.capture_event("test.event", mock_memory)
#81	mock_at.capture_event.assert_called_once()
#82
#83	def test_capture_client_event_sends_when_enabled(self):
#84	"""capture_client_event() should call client_telemetry.capture_event when enabled."""
#85	with patch.object(telemetry_module, "MEM0_TELEMETRY", True):
#86	mock_client_telemetry = MagicMock()
#87	with patch.object(telemetry_module, "client_telemetry", mock_client_telemetry):
#88	mock_instance = MagicMock()
#89	mock_instance.user_email = "test@example.com"
#90	telemetry_module.capture_client_event("test.event", mock_instance)
#91	mock_client_telemetry.capture_event.assert_called_once()
#92
#93
#94	class TestTelemetryEnvVar:
#95	"""Verify the MEM0_TELEMETRY env var parsing logic."""
#96
#97	@pytest.mark.parametrize(
#98	"value,expected",
#99	[
#100	("true", True),
#101	("True", True),
#102	("TRUE", True),
#103	("1", True),
#104	("yes", True),
#105	("false", False),
#106	("False", False),
#107	("0", False),
#108	("no", False),
#109	("anything_else", False),
#110	],
#111	)
#112	def test_env_var_parsing(self, value, expected):
#113	result = value.lower() in ("true", "1", "yes")
#114	assert result == expected
#115

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public