test: Strengthen assertions and isolate external HTTP calls

.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 = <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:

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"]

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

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

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)

posthog/test/ai/otel/test_processor.py

@@ -89,14 +89,22 @@ 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")
     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()

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

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,17 +489,15 @@ def test_gemini_audio_preserved_with_flag(self):
                     {
                         "inline_data": {
                             "mime_type": "audio/L16;codec=pcm;rate=24000",
-                            "data": "base64audiodata",
+                            "data": LONG_RAW_BASE64,
                         }
                     }
                 ]
             }
         ]
 
         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()

posthog/test/integrations/test_celery_integration.py

@@ -363,16 +363,21 @@ 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
             for c in mock_client.capture.call_args_list
             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,18 +391,23 @@ 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
             for c in mock_client.capture.call_args_list
             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()

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)

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"]

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"