diff --git a/.github/scripts/test_check_public_api.py b/.github/scripts/test_check_public_api.py index 6fade7b44..c1bf97f25 100644 --- a/.github/scripts/test_check_public_api.py +++ b/.github/scripts/test_check_public_api.py @@ -24,9 +24,22 @@ def load_check_public_api(): def test_attribute_details_uses_placeholder_values() -> None: check_public_api = load_check_public_api() + obj = SimpleNamespace( + path="posthog.version.VERSION", annotation=None, value='"7.19.1"' + ) + assert ( + check_public_api._attribute_details(obj) + == "posthog.version.VERSION = " + ) + for path, placeholder in check_public_api.ATTRIBUTE_VALUE_PLACEHOLDERS.items(): - obj = SimpleNamespace(path=path, annotation=None, value='"7.19.1"') - assert check_public_api._attribute_details(obj) == f"{path} = {placeholder}" + entry = SimpleNamespace(path=path, annotation=None, value='"7.19.1"') + assert check_public_api._attribute_details(entry) == f"{path} = {placeholder}" + + obj.path = "posthog.other.CONSTANT" + assert ( + check_public_api._attribute_details(obj) == 'posthog.other.CONSTANT = "7.19.1"' + ) def main() -> int: diff --git a/integration_tests/django5/test_exception_capture.py b/integration_tests/django5/test_exception_capture.py index 4a495e3ec..b29f6e322 100644 --- a/integration_tests/django5/test_exception_capture.py +++ b/integration_tests/django5/test_exception_capture.py @@ -61,9 +61,8 @@ def mock_capture(exception, **kwargs): assert response.status_code == 500 # CRITICAL: Verify PostHog captured the exception - assert len(captured) > 0, "Exception was NOT captured to PostHog!" + assert len(captured) == 1 - # Verify it's the right exception exception_data = captured[0] assert exception_data["type"] == "ValueError" assert "Test exception from Django 5 async view" in exception_data["message"] @@ -103,9 +102,8 @@ def mock_capture(exception, **kwargs): assert response.status_code == 500 # CRITICAL: Verify PostHog captured the exception - assert len(captured) > 0, "Exception was NOT captured to PostHog!" + assert len(captured) == 1 - # Verify it's the right exception exception_data = captured[0] assert exception_data["type"] == "ValueError" assert "Test exception from Django 5 sync view" in exception_data["message"] diff --git a/posthog/test/ai/langchain/test_callbacks.py b/posthog/test/ai/langchain/test_callbacks.py index 4faca7b0b..2556294ae 100644 --- a/posthog/test/ai/langchain/test_callbacks.py +++ b/posthog/test/ai/langchain/test_callbacks.py @@ -571,11 +571,29 @@ def test_personless_mode(mock_client): assert trace_args["distinct_id"] == id -def test_personless_mode_exception(mock_client): +@pytest.fixture +def unauthorized_http_client(): + import httpx + + def unauthorized(request): + return httpx.Response(401, json={"error": {"message": "Invalid API key"}}) + + with httpx.Client(transport=httpx.MockTransport(unauthorized)) as client: + yield client + + +def test_personless_mode_exception(mock_client, unauthorized_http_client): + from openai import AuthenticationError + prompt = ChatPromptTemplate.from_messages([("user", "Foo")]) - chain = prompt | ChatOpenAI(api_key="test", model="gpt-4o-mini") + chain = prompt | ChatOpenAI( + api_key="test", + model="gpt-4o-mini", + http_client=unauthorized_http_client, + max_retries=0, + ) callbacks = CallbackHandler(mock_client) - with pytest.raises(Exception): + with pytest.raises(AuthenticationError): chain.invoke({}, config={"callbacks": [callbacks]}) assert mock_client.capture.call_count == 3 span_args = mock_client.capture.call_args_list[0][1] @@ -593,7 +611,7 @@ def test_personless_mode_exception(mock_client): assert trace_args["properties"]["$process_person_profile"] is False id = uuid.uuid4() - with pytest.raises(Exception): + with pytest.raises(AuthenticationError): chain.invoke( {}, config={"callbacks": [CallbackHandler(mock_client, distinct_id=id)]} ) @@ -856,13 +874,24 @@ def runnable(_): def test_openai_error(mock_client): + import httpx + from openai import AuthenticationError + + requests = [] + + def unauthorized(request): + requests.append(request) + return httpx.Response(401, json={"error": {"message": "Invalid API key"}}) + prompt = ChatPromptTemplate.from_messages([("user", "Foo")]) - chain = prompt | ChatOpenAI(api_key="test", model="gpt-4o-mini") callbacks = CallbackHandler(mock_client) - - # 401 - with pytest.raises(Exception): - chain.invoke({}, config={"callbacks": [callbacks]}) + with httpx.Client(transport=httpx.MockTransport(unauthorized)) as http_client: + chain = prompt | ChatOpenAI( + api_key="test", model="gpt-4o-mini", http_client=http_client, max_retries=0 + ) + with pytest.raises(AuthenticationError): + chain.invoke({}, config={"callbacks": [callbacks]}) + assert len(requests) == 1 assert callbacks._runs == {} assert callbacks._parent_tree == {} @@ -1091,15 +1120,19 @@ async def test_async_openai_streaming(mock_client): assert isinstance(trace_props["$ai_output_state"], AIMessage) -def test_base_url_retrieval(mock_client): +def test_base_url_retrieval(mock_client, unauthorized_http_client): + from openai import AuthenticationError + prompt = ChatPromptTemplate.from_messages([("user", "Foo")]) chain = prompt | ChatOpenAI( api_key="test", model="posthog-mini", base_url="https://test.posthog.com", + http_client=unauthorized_http_client, + max_retries=0, ) callbacks = CallbackHandler(mock_client) - with pytest.raises(Exception): + with pytest.raises(AuthenticationError): chain.invoke({}, config={"callbacks": [callbacks]}) assert mock_client.capture.call_count == 3 diff --git a/posthog/test/ai/openai/test_resource_wrapping.py b/posthog/test/ai/openai/test_resource_wrapping.py index 02bf7b094..adda447f3 100644 --- a/posthog/test/ai/openai/test_resource_wrapping.py +++ b/posthog/test/ai/openai/test_resource_wrapping.py @@ -56,9 +56,11 @@ def test_client_resources_are_discovered_and_wrapped( client_type, client_kwargs, wrappers, resource_types ): + assert wrappers.keys() == resource_types.keys() client = client_type(posthog_client=MagicMock(), **client_kwargs) - for resource_name, wrapper_type in wrappers.items(): + for resource_name in resource_types: + wrapper_type = wrappers[resource_name] wrapped = getattr(client, resource_name) original = getattr(client, f"_original_{resource_name}") diff --git a/posthog/test/ai/otel/test_exporter.py b/posthog/test/ai/otel/test_exporter.py index 0b3f52be7..d1e36a4c4 100644 --- a/posthog/test/ai/otel/test_exporter.py +++ b/posthog/test/ai/otel/test_exporter.py @@ -94,5 +94,9 @@ def test_force_flush_delegates(self, mock_otlp_cls): exporter = PostHogTraceExporter(api_key="phc_test") inner = mock_otlp_cls.return_value - exporter.force_flush(timeout_millis=5000) - inner.force_flush.assert_called_once_with(5000) + for result in (True, False): + with self.subTest(result=result): + inner.force_flush.reset_mock() + inner.force_flush.return_value = result + self.assertIs(exporter.force_flush(timeout_millis=5000), result) + inner.force_flush.assert_called_once_with(5000) diff --git a/posthog/test/ai/otel/test_processor.py b/posthog/test/ai/otel/test_processor.py index 0bcbc34dc..c2e2c2694 100644 --- a/posthog/test/ai/otel/test_processor.py +++ b/posthog/test/ai/otel/test_processor.py @@ -89,8 +89,12 @@ def test_force_flush_delegates(self, mock_batch_cls, mock_otlp_cls): processor = PostHogSpanProcessor(api_key="phc_test") inner = mock_batch_cls.return_value - processor.force_flush(timeout_millis=5000) - inner.force_flush.assert_called_once_with(5000) + for result in (True, False): + with self.subTest(result=result): + inner.force_flush.reset_mock() + inner.force_flush.return_value = result + self.assertIs(processor.force_flush(timeout_millis=5000), result) + inner.force_flush.assert_called_once_with(5000) @patch("posthog.ai.otel.processor.OTLPSpanExporter") @patch("posthog.ai.otel.processor.BatchSpanProcessor") @@ -98,5 +102,9 @@ def test_force_flush_without_timeout(self, mock_batch_cls, mock_otlp_cls): processor = PostHogSpanProcessor(api_key="phc_test") inner = mock_batch_cls.return_value - processor.force_flush() - inner.force_flush.assert_called_once_with() + for result in (True, False): + with self.subTest(result=result): + inner.force_flush.reset_mock() + inner.force_flush.return_value = result + self.assertIs(processor.force_flush(), result) + inner.force_flush.assert_called_once_with() diff --git a/posthog/test/ai/test_async_stream_wrapper.py b/posthog/test/ai/test_async_stream_wrapper.py index 7cdba8298..cad184533 100644 --- a/posthog/test/ai/test_async_stream_wrapper.py +++ b/posthog/test/ai/test_async_stream_wrapper.py @@ -140,4 +140,6 @@ async def gen(): yield wrapper = AsyncStreamWrapper(gen(), source) - assert not hasattr(wrapper, "_nonexistent_private") + source._private = object() + with pytest.raises(AttributeError): + _ = wrapper._private diff --git a/posthog/test/ai/test_sanitization.py b/posthog/test/ai/test_sanitization.py index de386023a..34be067b4 100644 --- a/posthog/test/ai/test_sanitization.py +++ b/posthog/test/ai/test_sanitization.py @@ -373,7 +373,7 @@ def test_flag_preserves_media_across_entry_points(self): "source": { "type": "base64", "media_type": "image/jpeg", - "data": "A" * 64, + "data": LONG_RAW_BASE64, }, } ] @@ -388,7 +388,7 @@ def test_flag_preserves_media_across_entry_points(self): { "inline_data": { "mime_type": "image/jpeg", - "data": "A" * 64, + "data": LONG_RAW_BASE64, } } ] @@ -406,7 +406,9 @@ def test_flag_preserves_media_across_entry_points(self): ], ), ]: - self.assertEqual(fn(data, ph_client=client), data) + with self.subTest(entry_point=fn.__name__): + self.assertNotEqual(fn(data, ph_client=self._client(False)), data) + self.assertEqual(fn(data, ph_client=client), data) def test_flag_off_still_redacts(self): result = sanitize_openai(self.openai_input, ph_client=self._client(False)) @@ -452,13 +454,13 @@ def test_openai_audio_preserved_with_flag(self): { "role": "assistant", "content": [ - {"type": "audio", "data": "base64audiodata", "id": "audio_123"} + {"type": "audio", "data": LONG_RAW_BASE64, "id": "audio_123"} ], } ] result = sanitize_openai(input_data, ph_client=self._client(True)) - self.assertEqual(result[0]["content"][0]["data"], "base64audiodata") + self.assertEqual(result[0]["content"][0]["data"], LONG_RAW_BASE64) def test_gemini_audio_redacted_by_default(self): input_data = [ @@ -487,7 +489,7 @@ def test_gemini_audio_preserved_with_flag(self): { "inline_data": { "mime_type": "audio/L16;codec=pcm;rate=24000", - "data": "base64audiodata", + "data": LONG_RAW_BASE64, } } ] @@ -495,9 +497,7 @@ def test_gemini_audio_preserved_with_flag(self): ] result = sanitize_gemini(input_data, ph_client=self._client(True)) - self.assertEqual( - result[0]["parts"][0]["inline_data"]["data"], "base64audiodata" - ) + self.assertEqual(result[0]["parts"][0]["inline_data"]["data"], LONG_RAW_BASE64) PNG_B64 = base64.b64encode(b"\x89PNG\r\n\x1a\n" + b"\x00" * 400).decode() diff --git a/posthog/test/integrations/test_celery_integration.py b/posthog/test/integrations/test_celery_integration.py index 3ef0b4160..565800cac 100644 --- a/posthog/test/integrations/test_celery_integration.py +++ b/posthog/test/integrations/test_celery_integration.py @@ -363,8 +363,11 @@ def test_postrun_includes_duration(self): ) task = SimpleNamespace(name="app.tasks.timed", request=request) - integration._on_task_prerun(sender=task, task_id="task-t") - integration._on_task_success(sender=task) + with patch( + "posthog.integrations.celery.time.monotonic", side_effect=[100, 100.125] + ): + integration._on_task_prerun(sender=task, task_id="task-t") + integration._on_task_success(sender=task) completed_call = [ c @@ -372,7 +375,9 @@ def test_postrun_includes_duration(self): if c.args[0] == "celery task success" ] self.assertEqual(len(completed_call), 1) - self.assertIn("celery_task_duration_ms", completed_call[0].kwargs["properties"]) + self.assertEqual( + completed_call[0].kwargs["properties"]["celery_task_duration_ms"], 125.0 + ) def test_failure_includes_duration(self): mock_client = Mock() @@ -386,10 +391,13 @@ def test_failure_includes_duration(self): ) task = SimpleNamespace(name="app.tasks.failing_timed", request=request) - integration._on_task_prerun(sender=task, task_id="task-f") - integration._on_task_failure( - sender=task, task_id="task-f", exception=ValueError("boom") - ) + with patch( + "posthog.integrations.celery.time.monotonic", side_effect=[100, 100.125] + ): + integration._on_task_prerun(sender=task, task_id="task-f") + integration._on_task_failure( + sender=task, task_id="task-f", exception=ValueError("boom") + ) failed_call = [ c @@ -397,7 +405,9 @@ def test_failure_includes_duration(self): if c.args[0] == "celery task failure" ] self.assertEqual(len(failed_call), 1) - self.assertIn("celery_task_duration_ms", failed_call[0].kwargs["properties"]) + self.assertEqual( + failed_call[0].kwargs["properties"]["celery_task_duration_ms"], 125.0 + ) def test_task_failure_captures_exception_and_failure_event(self): mock_client = Mock() diff --git a/posthog/test/integrations/test_middleware.py b/posthog/test/integrations/test_middleware.py index 22ce948c9..1d4e26089 100644 --- a/posthog/test/integrations/test_middleware.py +++ b/posthog/test/integrations/test_middleware.py @@ -2,6 +2,7 @@ new_context, get_context_session_id, get_context_distinct_id, + get_tags, ) import unittest from unittest.mock import Mock, patch @@ -509,12 +510,13 @@ async def async_get_response(request): # Override request filter after initialization middleware.request_filter = lambda req: False - request = MockRequest() + request = MockRequest(headers={"X-POSTHOG-SESSION-ID": "filtered"}) - # Should skip context creation and return response directly - result = middleware(request) - response = await result - self.assertEqual(response, mock_response) + with patch.object(middleware, "aextract_tags") as extract: + result = middleware(request) + response = await result + self.assertEqual(response, mock_response) + extract.assert_not_awaited() asyncio.run(run_test()) @@ -690,6 +692,7 @@ def extra_tags_callback(request): return {"custom_tag": "custom_value"} async def async_get_response(request): + self.assertEqual(get_tags()["custom_tag"], "custom_value") return mock_response middleware = PosthogContextMiddleware(async_get_response) @@ -727,6 +730,7 @@ def tag_map_callback(tags): return tags async def async_get_response(request): + self.assertEqual(get_tags()["mapped"], "yes") return mock_response middleware = PosthogContextMiddleware(async_get_response) diff --git a/posthog/test/mcp/test_conversation_session.py b/posthog/test/mcp/test_conversation_session.py index 0e1c04dad..fef61256b 100644 --- a/posthog/test/mcp/test_conversation_session.py +++ b/posthog/test/mcp/test_conversation_session.py @@ -246,9 +246,11 @@ def echo(msg: str) -> str: return msg client = FakeClient() - instrument(server, client) # enable_conversation_id defaults off + instrument(server, client, MCPAnalyticsOptions(enable_conversation_id=False)) - await server._tool_manager.call_tool("echo", {"msg": "a", "context": "x"}) + await server._tool_manager.call_tool( + "echo", {"msg": "a", "context": "x"}, convert_result=True + ) await _flush() props = _events(client, "$mcp_tool_call")[0]["properties"] diff --git a/posthog/test/mcp/test_feedback.py b/posthog/test/mcp/test_feedback.py index b7583d3f9..0ce4275e9 100644 --- a/posthog/test/mcp/test_feedback.py +++ b/posthog/test/mcp/test_feedback.py @@ -132,6 +132,7 @@ def test_descriptor_deep_copies_host_fragments(): options = CollectFeedbackOptions(extra_properties={"area": fragment}) descriptor = get_feedback_tool_descriptor(options) fragment["enum"].append("mutated") + assert descriptor["inputSchema"]["properties"]["area"]["enum"] == ["a", "b"] descriptor["inputSchema"]["properties"]["feedback_type"]["enum"].append("bogus") assert get_feedback_tool_descriptor(options)["inputSchema"]["properties"]["area"][ "enum" diff --git a/posthog/test/mcp/test_pending_tasks.py b/posthog/test/mcp/test_pending_tasks.py index a9abf09e3..f1da78659 100644 --- a/posthog/test/mcp/test_pending_tasks.py +++ b/posthog/test/mcp/test_pending_tasks.py @@ -9,6 +9,7 @@ async def test_async_drain_is_scoped_to_owner(): first_owner = object() second_owner = object() first_done = [] + second_done = [] second_started = asyncio.Event() release_second = asyncio.Event() @@ -19,18 +20,20 @@ async def first_capture(): async def second_capture(): second_started.set() await release_second.wait() + second_done.append(True) instrumentation.fire_and_forget(first_capture(), first_owner) instrumentation.fire_and_forget(second_capture(), second_owner) await second_started.wait() - await asyncio.wait_for(instrumentation.drain_pending(first_owner), timeout=1) - - assert first_done == [True] - assert not release_second.is_set() - - release_second.set() - await instrumentation.drain_pending(second_owner) + try: + await asyncio.wait_for(instrumentation.drain_pending(first_owner), timeout=1) + assert first_done == [True] + assert second_done == [] + finally: + release_second.set() + await asyncio.wait_for(instrumentation.drain_pending(second_owner), timeout=1) + assert second_done == [True] async def test_async_drain_ignores_same_owner_tasks_on_another_loop(): diff --git a/posthog/test/mcp/test_pipeline.py b/posthog/test/mcp/test_pipeline.py index cab7cb9c1..81b6db644 100644 --- a/posthog/test/mcp/test_pipeline.py +++ b/posthog/test/mcp/test_pipeline.py @@ -839,10 +839,7 @@ def test_normalize_detects_cycles(): def test_normalize_limits_depth(): deep = {"a": {"b": {"c": {"d": {"e": {"f": "deep"}}}}}} out = normalize(deep, depth=2) - # at depth 2 the nested object should be collapsed to a marker - assert out["a"]["b"] in ("[Object]", {"c": "[Object]"}) or isinstance( - out["a"]["b"], (dict, str) - ) + assert out == {"a": {"b": "[Object]"}} def test_normalize_handles_nan_and_infinity(): diff --git a/posthog/test/mcp/test_review_fixes.py b/posthog/test/mcp/test_review_fixes.py index d8be0b4e3..8c2886ff1 100644 --- a/posthog/test/mcp/test_review_fixes.py +++ b/posthog/test/mcp/test_review_fixes.py @@ -325,6 +325,7 @@ def test_posthogmcp_can_disable_exception_fanout(): def test_version_warning_uses_supplied_logger(monkeypatch): # #6: the mcp-version advisory goes to the supplied logger, not the no-op default. logs = [] + monkeypatch.setattr("posthog.mcp.logger._active_logger", None) monkeypatch.setattr("importlib.metadata.version", lambda name: "1.20.0") server = Server("ver-warn") diff --git a/posthog/test/mcp/test_truncation.py b/posthog/test/mcp/test_truncation.py index c05af44da..cc00c3fef 100644 --- a/posthog/test/mcp/test_truncation.py +++ b/posthog/test/mcp/test_truncation.py @@ -120,7 +120,7 @@ def test_truncate_event_caps_exception_value_and_frames(): ) exc = out["error"]["$exception_list"][0] assert len(exc["value"]) == 2048 + 3 - assert len(exc["stacktrace"]["frames"]) == 50 # head + tail + assert exc["stacktrace"]["frames"] == frames[:25] + frames[-25:] def test_truncate_event_caps_response_text_block(): diff --git a/posthog/test/test_code_variables.py b/posthog/test/test_code_variables.py index ad197a9c0..cb3d3cbe1 100644 --- a/posthog/test/test_code_variables.py +++ b/posthog/test/test_code_variables.py @@ -77,6 +77,17 @@ def extract( import os import posthog from posthog import Posthog +from requests import Response +import posthog.request + + +def offline_post(url, **kwargs): + response = Response() + response.status_code = 200 + return response + + +posthog.request._session.post = offline_post def make_client(**options): @@ -104,7 +115,10 @@ def run_app(tmpdir, body, *, env=None): run_env = {**os.environ, **(env or {})} with pytest.raises(subprocess.CalledProcessError) as excinfo: subprocess.check_output( - [sys.executable, str(app)], stderr=subprocess.STDOUT, env=run_env + [sys.executable, str(app)], + stderr=subprocess.STDOUT, + env=run_env, + timeout=15, ) return excinfo.value.output.decode("utf-8") diff --git a/posthog/test/test_consumer.py b/posthog/test/test_consumer.py index 3782ad37f..c2c54cf52 100644 --- a/posthog/test/test_consumer.py +++ b/posthog/test/test_consumer.py @@ -209,10 +209,16 @@ def test_max_msg_size_param_raises_per_event_ceiling(self) -> None: def test_upload(self) -> None: q = Queue() - consumer = Consumer(q, TEST_API_KEY) - q.put(_track_event()) - success = consumer.upload() + consumer = Consumer(q, TEST_API_KEY, flush_at=1) + event = _track_event() + q.put(event) + with mock.patch("posthog.consumer.batch_post") as post: + success = consumer.upload() self.assertTrue(success) + post.assert_called_once() + self.assertEqual(post.call_args.kwargs["batch"], [event]) + self.assertEqual(q.unfinished_tasks, 0) + self.assertTrue(q.empty()) def test_message_only_error_logs_include_posthog_prefix(self) -> None: q = Queue() @@ -237,41 +243,112 @@ def test_message_only_error_logs_include_posthog_prefix(self) -> None: ) def test_flush_interval(self) -> None: - # Put _n_ items in the queue, pausing a little bit more than - # _flush_interval_ after each one. - # The consumer should upload _n_ times. q = Queue() flush_interval = 0.3 consumer = Consumer(q, TEST_API_KEY, flush_at=10, flush_interval=flush_interval) - with mock.patch.object(consumer, "request") as mock_request: + delivered = threading.Event() + with mock.patch.object( + consumer, "request", side_effect=lambda batch: delivered.set() + ) as mock_request: consumer.start() - for i in range(3): - q.put(_track_event("python event %d" % i)) - time.sleep(flush_interval * 1.1) - self.assertEqual(mock_request.call_count, 3) + try: + for i in range(3): + delivered.clear() + event = _track_event("python event %d" % i) + q.put(event) + self.assertTrue(delivered.wait(5)) + self.assertEqual(mock_request.call_args.args[0], [event]) + self.assertEqual(mock_request.call_count, 3) + finally: + consumer.pause() + consumer.join(5) + self.assertFalse(consumer.is_alive()) + + def test_partial_batch_waits_for_remaining_flush_interval(self) -> None: + from queue import Empty + + q = Queue() + event = _track_event() + q.put(event) + consumer = Consumer(q, TEST_API_KEY, flush_at=10, flush_interval=0.3) + now = [100.0] + waits = [] + original_get = q.get + + def timed_get(*, block, timeout): + waits.append(timeout) + if len(waits) == 1: + now[0] += 0.125 + return original_get(block=False) + now[0] += timeout + raise Empty + + with ( + mock.patch("posthog.consumer.time.monotonic", side_effect=lambda: now[0]), + mock.patch.object(q, "get", side_effect=timed_get), + mock.patch.object(consumer, "request") as request, + ): + self.assertTrue(consumer.upload()) + + self.assertEqual(len(waits), 2) + self.assertAlmostEqual(waits[0], 0.3) + self.assertAlmostEqual(waits[1], 0.175) + self.assertAlmostEqual(now[0], 100.3) + request.assert_called_once_with([event]) + self.assertEqual(q.unfinished_tasks, 0) def test_multiple_uploads_per_interval(self) -> None: - # Put _flush_at*2_ items in the queue at once, then pause for - # _flush_interval_. The consumer should upload 2 times. q = Queue() - flush_interval = 0.5 + flush_interval = 10 flush_at = 10 consumer = Consumer( q, TEST_API_KEY, flush_at=flush_at, flush_interval=flush_interval ) - with mock.patch("posthog.consumer.batch_post") as mock_post: + delivered = threading.Event() + batches = [] + + def record_batch(*args, **kwargs): + batches.append(kwargs["batch"]) + if len(batches) == 2: + delivered.set() + + with mock.patch("posthog.consumer.batch_post", side_effect=record_batch): consumer.start() - for i in range(flush_at * 2): - q.put(_track_event("python event %d" % i)) - time.sleep(flush_interval * 1.1) - self.assertEqual(mock_post.call_count, 2) + try: + events = [ + _track_event("python event %d" % i) for i in range(flush_at * 2) + ] + for event in events: + q.put(event) + self.assertTrue(delivered.wait(5)) + self.assertEqual(batches, [events[:10], events[10:]]) + finally: + consumer.pause() + consumer.join(15) + self.assertFalse(consumer.is_alive()) def test_request(self) -> None: consumer = Consumer(None, TEST_API_KEY) - consumer.request([_track_event()]) + batch = [_track_event()] + with mock.patch("posthog.consumer.batch_post") as post: + consumer.request(batch) + post.assert_called_once_with( + TEST_API_KEY, + None, + gzip=False, + timeout=15, + batch=batch, + historical_migration=False, + path="/batch/", + ) def _run_retry_test( - self, exception: Exception, exception_count: int, retries: int = 10 + self, + exception: Exception, + exception_count: int, + retries: int = 10, + expected_attempts: int = 3, + raises: bool = False, ) -> None: call_count = [0] @@ -281,14 +358,20 @@ def mock_post(*args: Any, **kwargs: Any) -> None: raise exception consumer = Consumer(None, TEST_API_KEY, retries=retries) - with mock.patch( - "posthog.consumer.batch_post", mock.Mock(side_effect=mock_post) + batch = [_track_event()] + with ( + mock.patch("posthog.consumer.batch_post", side_effect=mock_post) as post, + mock.patch("posthog.consumer.time.sleep"), ): - if exception_count <= retries: - consumer.request([_track_event()]) + if raises: + with self.assertRaises(type(exception)) as raised: + consumer.request(batch) + self.assertIs(raised.exception, exception) else: - with self.assertRaises(type(exception)): - consumer.request([_track_event()]) + consumer.request(batch) + self.assertEqual(post.call_count, expected_attempts) + for call in post.call_args_list: + self.assertEqual(call.kwargs["batch"], batch) @parameterized.expand( [ @@ -303,11 +386,18 @@ def test_request_retries_on_retriable_errors( self._run_retry_test(exception, exception_count) def test_request_does_not_retry_client_errors(self) -> None: - with self.assertRaises(APIError): - self._run_retry_test(APIError(400, "Client Errors"), 1) + self._run_retry_test( + APIError(400, "Client Errors"), 1, expected_attempts=1, raises=True + ) def test_request_fails_when_exceptions_exceed_retries(self) -> None: - self._run_retry_test(APIError(500, "Internal Server Error"), 4, retries=3) + self._run_retry_test( + APIError(500, "Internal Server Error"), + 4, + retries=3, + expected_attempts=4, + raises=True, + ) def test_negative_retries_still_attempts_delivery_once(self) -> None: consumer = Consumer(None, TEST_API_KEY, retries=-1) @@ -512,22 +602,23 @@ def test_max_batch_size(self) -> None: # Let's capture 8MB of data to trigger two batches n_msgs = int(8_000_000 / msg_size) - def mock_send_fn(batch: list[dict[str, Any]], _path: str) -> None: - request_size = len(json.dumps({"batch": batch}).encode()) - # Batches close after the first message bringing it bigger than BATCH_SIZE_LIMIT, let's add 10% of margin - self.assertTrue( - request_size < (5 * 1024 * 1024) * 1.1, - "batch size (%d) higher than limit" % request_size, - ) - - with mock.patch.object( - consumer, "_send", side_effect=mock_send_fn - ) as mock_send: + with mock.patch.object(consumer, "_send") as mock_send: consumer.start() - for _ in range(0, n_msgs + 2): - q.put(track) - q.join() - self.assertEqual(mock_send.call_count, 2) + try: + for _ in range(0, n_msgs + 2): + q.put(track) + q.join() + self.assertEqual(mock_send.call_count, 2) + batches = [call.args[0] for call in mock_send.call_args_list] + self.assertEqual(sum(map(len, batches)), n_msgs + 2) + for batch in batches: + request_size = len(json.dumps({"batch": batch}).encode()) + # The event crossing the byte limit is included in the batch. + self.assertLess(request_size, (5 * 1024 * 1024) * 1.1) + finally: + consumer.pause() + consumer.join(5) + self.assertFalse(consumer.is_alive()) def test_request_sleeps_with_retry_after(self) -> None: error = APIError(429, "Too Many Requests", retry_after=5.0) diff --git a/posthog/test/test_exception_capture.py b/posthog/test/test_exception_capture.py index 85b50ac16..ef83c433f 100644 --- a/posthog/test/test_exception_capture.py +++ b/posthog/test/test_exception_capture.py @@ -278,6 +278,15 @@ def test_excepthook(tmpdir): dedent( """ from posthog import Posthog + from requests import Response + import posthog.request + + def offline_post(url, **kwargs): + response = Response() + response.status_code = 200 + return response + + posthog.request._session.post = offline_post posthog = Posthog('phc_x', host='https://eu.i.posthog.com', enable_exception_autocapture=True, debug=True, on_error=lambda e, batch: print('error handling batch: ', e, batch)) # frame_value = "LOL" @@ -288,7 +297,9 @@ def test_excepthook(tmpdir): ) with pytest.raises(subprocess.CalledProcessError) as excinfo: - subprocess.check_output([sys.executable, str(app)], stderr=subprocess.STDOUT) + subprocess.check_output( + [sys.executable, str(app)], stderr=subprocess.STDOUT, timeout=15 + ) output = excinfo.value.output diff --git a/posthog/test/test_flag_definition_cache.py b/posthog/test/test_flag_definition_cache.py index b66ee7c0f..2d05997d4 100644 --- a/posthog/test/test_flag_definition_cache.py +++ b/posthog/test/test_flag_definition_cache.py @@ -622,9 +622,7 @@ def test_multiple_join_calls_only_shutdown_once(self, mock_get): client.join() client.join() - # Shutdown should be called each time (current behavior - no guard) - # This test documents the current behavior - self.assertGreaterEqual(self.cache_provider.shutdown_call_count, 1) + self.assertEqual(self.cache_provider.shutdown_call_count, 1) class TestBackwardCompatibility(TestFlagDefinitionCacheProvider): diff --git a/posthog/test/test_module.py b/posthog/test/test_module.py index 9e3db5a47..dc875e674 100644 --- a/posthog/test/test_module.py +++ b/posthog/test/test_module.py @@ -13,26 +13,40 @@ class TestModule(unittest.TestCase): posthog = None def _assert_enqueue_result(self, result): - self.assertEqual(type(result[0]), str) - - def failed(self): - self.failed = True + self.assertIsInstance(result, str) + self.assertTrue(result) def setUp(self): - self.failed = False - self.posthog = Posthog( - "testsecret", host="http://localhost:8000", on_error=self.failed - ) + patcher = mock.patch("posthog.consumer.batch_post") + self.transport = patcher.start() + self.addCleanup(patcher.stop) + self.on_error = mock.Mock() + self.posthog = Posthog("testsecret", on_error=self.on_error) + self.addCleanup(self.posthog.shutdown) + + def tearDown(self): + self.on_error.assert_not_called() def test_track(self): res = self.posthog.capture("python module event", distinct_id="distinct_id") self._assert_enqueue_result(res) self.posthog.flush() + self.transport.assert_called_once() + event = self.transport.call_args.kwargs["batch"][0] + self.assertEqual(event["event"], "python module event") + self.assertEqual(event["distinct_id"], "distinct_id") + self.assertEqual(event["uuid"], res) def test_alias(self): res = self.posthog.alias("previousId", "distinct_id") self._assert_enqueue_result(res) self.posthog.flush() + self.transport.assert_called_once() + event = self.transport.call_args.kwargs["batch"][0] + self.assertEqual(event["event"], "$create_alias") + self.assertEqual(event["distinct_id"], "previousId") + self.assertEqual(event["properties"]["alias"], "distinct_id") + self.assertEqual(event["uuid"], res) def test_flush(self): self.posthog.flush() diff --git a/posthog/test/test_request.py b/posthog/test/test_request.py index 6a1226cc4..d4ad7c9da 100644 --- a/posthog/test/test_request.py +++ b/posthog/test/test_request.py @@ -1,3 +1,4 @@ +import gzip import json import unittest import zlib @@ -132,22 +133,44 @@ def test_message_only_debug_logs_include_posthog_prefix(): class TestRequests(unittest.TestCase): def test_valid_request(self): - res = batch_post( - TEST_API_KEY, - batch=[ - {"distinct_id": "distinct_id", "event": "python event", "type": "track"} - ], - ) - self.assertEqual(res.status_code, 200) + response = requests.Response() + response.status_code = 200 + session = mock.Mock() + session.post.return_value = response + batch = [ + {"distinct_id": "distinct_id", "event": "python event", "type": "track"} + ] + + res = batch_post(TEST_API_KEY, batch=batch, session=session) + + self.assertIs(res, response) + session.post.assert_called_once() + self.assertTrue(session.post.call_args.args[0].endswith("/batch/")) + body = json.loads(session.post.call_args.kwargs["data"]) + self.assertEqual(body["batch"], batch) + self.assertEqual(body["api_key"], TEST_API_KEY) def test_invalid_request_error(self): - self.assertRaises( - Exception, batch_post, "testsecret", "https://t.posthog.com", False, "[{]" - ) + response = requests.Response() + response.status_code = 400 + response._content = b'{"detail": "Invalid batch"}' + session = mock.Mock() + session.post.return_value = response + + with self.assertRaises(APIError) as raised: + batch_post("testsecret", batch=[], session=session) + + self.assertEqual(raised.exception.status, 400) + self.assertEqual(raised.exception.message, "Invalid batch") + session.post.assert_called_once() def test_invalid_host(self): self.assertRaises( - Exception, batch_post, "testsecret", "t.posthog.com/", batch=[] + requests.exceptions.MissingSchema, + batch_post, + "testsecret", + "t.posthog.com/", + batch=[], ) def test_post_without_path_preserves_type_error(self): @@ -201,6 +224,9 @@ def test_post_sends_bytes_payload_with_gzip(self): headers = mock_session.post.call_args.kwargs["headers"] self.assertIsInstance(data, bytes) self.assertEqual(headers["Content-Encoding"], "gzip") + body = json.loads(gzip.decompress(data)) + self.assertEqual(body["batch"], []) + self.assertEqual(body["api_key"], TEST_API_KEY) def test_post_falls_back_to_uncompressed_payload_when_gzip_fails(self): for compression_error in [OSError("boom"), zlib.error("boom")]: @@ -240,28 +266,28 @@ def test_date_serialization(self): self.assertEqual(result, expected) def test_should_not_timeout(self): - res = batch_post( - TEST_API_KEY, - batch=[ - {"distinct_id": "distinct_id", "event": "python event", "type": "track"} - ], - timeout=15, + response = requests.Response() + response.status_code = 200 + session = mock.Mock() + session.post.return_value = response + + self.assertIs( + batch_post(TEST_API_KEY, batch=[], timeout=7, session=session), response ) - self.assertEqual(res.status_code, 200) + session.post.assert_called_once() + self.assertEqual(session.post.call_args.kwargs["timeout"], 7) def test_should_timeout(self): - with self.assertRaises(requests.ReadTimeout): - batch_post( - "key", - batch=[ - { - "distinct_id": "distinct_id", - "event": "python event", - "type": "track", - } - ], - timeout=0.0001, - ) + error = requests.ReadTimeout("response deadline exceeded") + session = mock.Mock() + session.post.side_effect = error + + with self.assertRaises(requests.ReadTimeout) as raised: + batch_post("key", batch=[], timeout=1, session=session) + + self.assertIs(raised.exception, error) + session.post.assert_called_once() + self.assertEqual(session.post.call_args.kwargs["timeout"], 1) def test_quota_limited_flags_response(self): mock_response = requests.Response() diff --git a/posthog/test/tracing/test_client_traces.py b/posthog/test/tracing/test_client_traces.py index ae7072b88..ab9d0ee0e 100644 --- a/posthog/test/tracing/test_client_traces.py +++ b/posthog/test/tracing/test_client_traces.py @@ -349,11 +349,10 @@ def test_continues_a_trace_from_an_inbound_traceparent_and_propagates_the_flag( def test_records_a_raised_error_and_rethrows_it_unchanged(self): client = make_client(traces={}) error = RuntimeError("boom") - try: + with pytest.raises(RuntimeError) as raised: with client.start_span("x"): raise error - except RuntimeError as raised: - assert raised is error + assert raised.value is error payload, _, _ = flush_and_capture(client) (record,) = spans_from(payload) assert record["status"] == {"code": 2, "message": "boom"} diff --git a/posthog/test/tracing/test_config.py b/posthog/test/tracing/test_config.py index 2bc1af5e3..0c0812f6d 100644 --- a/posthog/test/tracing/test_config.py +++ b/posthog/test/tracing/test_config.py @@ -7,7 +7,6 @@ DEFAULT_FLUSH_INTERVAL_SECONDS, DEFAULT_MAX_EXPORT_BATCH_SIZE, DEFAULT_MAX_LIVE_SPANS, - DEFAULT_MAX_QUEUE_SIZE, DEFAULT_MAX_SPAN_AGE_SECONDS, ResolvedTracesConfig, resolve_traces_config, @@ -17,11 +16,11 @@ class TestDefaults: def test_applies_the_documented_defaults(self): assert resolve_traces_config({}) == ResolvedTracesConfig( - flush_interval=DEFAULT_FLUSH_INTERVAL_SECONDS, - max_export_batch_size=DEFAULT_MAX_EXPORT_BATCH_SIZE, - max_queue_size=DEFAULT_MAX_QUEUE_SIZE, - max_live_spans=DEFAULT_MAX_LIVE_SPANS, - max_span_age=DEFAULT_MAX_SPAN_AGE_SECONDS, + flush_interval=5.0, + max_export_batch_size=512, + max_queue_size=2048, + max_live_spans=10000, + max_span_age=3600.0, ) def test_leaves_service_name_unset_so_the_encoder_supplies_unknown_service(self): diff --git a/posthog/test/tracing/test_export.py b/posthog/test/tracing/test_export.py index 3d1c532bb..52cff1515 100644 --- a/posthog/test/tracing/test_export.py +++ b/posthog/test/tracing/test_export.py @@ -378,12 +378,11 @@ def test_drops_a_poison_batch_rather_than_wedging_the_queue(self): assert queued(pipeline) == [] def test_does_not_surface_a_transport_failure_through_span_end(self): - def explode(client, payload): - raise RuntimeError("transport broke") - - pipeline, _, _ = make_traces(sender=explode, max_export_batch_size=1) + sender = mock.Mock(side_effect=RuntimeError("transport broke")) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=1) pipeline.start_span("a").end() FakeTimer.instances[-1].fire() + sender.assert_called_once() def test_never_returns_a_span_it_failed_to_encode(self): pipeline, sender, _ = make_traces() diff --git a/posthog/test/tracing/test_transport.py b/posthog/test/tracing/test_transport.py index 29e9ad4b8..1073bbda6 100644 --- a/posthog/test/tracing/test_transport.py +++ b/posthog/test/tracing/test_transport.py @@ -150,10 +150,12 @@ def test_reads_retry_after_delta_seconds(self): assert outcome == SendOutcome("retry-later", 120.0) def test_reads_retry_after_http_date(self): - when = format_datetime(datetime.now(timezone.utc) + timedelta(seconds=120)) - outcome, _ = send(session=mock_session(503, {"Retry-After": when})) - assert outcome.kind == "retry-later" - assert outcome.retry_after is not None and 100 < outcome.retry_after <= 120 + from freezegun import freeze_time + + with freeze_time("2026-09-10T12:00:00Z"): + when = format_datetime(datetime.now(timezone.utc) + timedelta(seconds=120)) + outcome, _ = send(session=mock_session(503, {"Retry-After": when})) + assert outcome == SendOutcome("retry-later", 120.0) NOW = datetime(2026, 9, 10, 12, 0, 0, tzinfo=timezone.utc)