[openai, openai-agents] Use concrete invocation types in raw response and robust mock assertions

instrumentation/opentelemetry-instrumentation-genai-openai-agents/tests/test_processor.py

@@ -121,10 +121,14 @@ def test_function_span_creates_tool_invocation() -> None:
 
 def test_function_span_skips_content_when_capture_disabled() -> None:
     handler = _build_handler()
+    original_arguments = object()
+    original_result = object()
     handler.tool.return_value = MagicMock(
         spec=ToolInvocation,
         metric_attributes={},
         should_capture_content=False,
+        arguments=original_arguments,
+        tool_result=original_result,
     )
     processor = GenAITracingProcessor(handler, provider="openai")
     span = _Span(FunctionSpanData(name="get_weather", input=None, output=None))
@@ -135,12 +139,8 @@ def test_function_span_skips_content_when_capture_disabled() -> None:
     processor.on_span_end(span)
 
     tool_invocation = handler.tool.return_value
-    # A spec'd mock rejects reads of attributes nothing assigned, so these
-    # assert the processor skipped the serialization work entirely.
-    with pytest.raises(AttributeError):
-        _ = tool_invocation.arguments
-    with pytest.raises(AttributeError):
-        _ = tool_invocation.tool_result
+    assert tool_invocation.arguments is original_arguments
+    assert tool_invocation.tool_result is original_result
     tool_invocation.stop.assert_called_once_with()
 
 

instrumentation/opentelemetry-instrumentation-genai-openai/src/opentelemetry/instrumentation/genai/openai/_raw_response.py

@@ -27,7 +27,12 @@
 from .utils import get_served_model
 
 if TYPE_CHECKING:
-    from opentelemetry.util.genai.types import GenAIInvocation
+    from opentelemetry.util.genai.invocation import (
+        FetchResponseInvocation,
+        InferenceInvocation,
+    )
+
+    _StreamingInvocation = InferenceInvocation | FetchResponseInvocation
 
 _logger = logging.getLogger(__name__)
 
@@ -229,15 +234,15 @@ class StreamWrapperFactory(Protocol):
     def __call__(
         self,
         stream: AnyStream,
-        invocation: GenAIInvocation,
+        invocation: _StreamingInvocation,
         capture_content: bool,
     ) -> object: ...
 
 
 def wrap_stream_result(
     wrapper_cls: StreamWrapperFactory,
     result: RawResponseLike | AnyStream,
-    invocation: GenAIInvocation,
+    invocation: _StreamingInvocation,
     capture_content: bool,
 ) -> object:
     """Wrap a streaming call's result, deferring ``parse()`` on raw responses.