my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	import logging
#2	import os
#3
#4	from embedchain.telemetry.posthog import AnonymousTelemetry
#5
#6
#7	class TestAnonymousTelemetry:
#8	def test_init(self, mocker):
#9	# Enable telemetry specifically for this test
#10	os.environ["EC_TELEMETRY"] = "true"
#11	mock_posthog = mocker.patch("embedchain.telemetry.posthog.Posthog")
#12	telemetry = AnonymousTelemetry()
#13	assert telemetry.project_api_key == "phc_PHQDA5KwztijnSojsxJ2c1DuJd52QCzJzT2xnSGvjN2"
#14	assert telemetry.host == "https://app.posthog.com"
#15	assert telemetry.enabled is True
#16	assert telemetry.user_id
#17	mock_posthog.assert_called_once_with(project_api_key=telemetry.project_api_key, host=telemetry.host)
#18
#19	def test_init_with_disabled_telemetry(self, mocker):
#20	mocker.patch("embedchain.telemetry.posthog.Posthog")
#21	telemetry = AnonymousTelemetry()
#22	assert telemetry.enabled is False
#23	assert telemetry.posthog.disabled is True
#24
#25	def test_get_user_id(self, mocker, tmpdir):
#26	mock_uuid = mocker.patch("embedchain.telemetry.posthog.uuid.uuid4")
#27	mock_uuid.return_value = "unique_user_id"
#28	config_file = tmpdir.join("config.json")
#29	mocker.patch("embedchain.telemetry.posthog.CONFIG_FILE", str(config_file))
#30	telemetry = AnonymousTelemetry()
#31
#32	user_id = telemetry._get_user_id()
#33	assert user_id == "unique_user_id"
#34	assert config_file.read() == '{"user_id": "unique_user_id"}'
#35
#36	def test_capture(self, mocker):
#37	# Enable telemetry specifically for this test
#38	os.environ["EC_TELEMETRY"] = "true"
#39	mock_posthog = mocker.patch("embedchain.telemetry.posthog.Posthog")
#40	telemetry = AnonymousTelemetry()
#41	event_name = "test_event"
#42	properties = {"key": "value"}
#43	telemetry.capture(event_name, properties)
#44
#45	mock_posthog.assert_called_once_with(
#46	project_api_key=telemetry.project_api_key,
#47	host=telemetry.host,
#48	)
#49	mock_posthog.return_value.capture.assert_called_once_with(
#50	telemetry.user_id,
#51	event_name,
#52	properties,
#53	)
#54
#55	def test_capture_with_exception(self, mocker, caplog):
#56	os.environ["EC_TELEMETRY"] = "true"
#57	mock_posthog = mocker.patch("embedchain.telemetry.posthog.Posthog")
#58	mock_posthog.return_value.capture.side_effect = Exception("Test Exception")
#59	telemetry = AnonymousTelemetry()
#60	event_name = "test_event"
#61	properties = {"key": "value"}
#62	with caplog.at_level(logging.ERROR):
#63	telemetry.capture(event_name, properties)
#64	assert "Failed to send telemetry event" in caplog.text
#65	caplog.clear()
#66

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public