chore: 100% clean coverage (#1705)

Author: Kludex

.github/workflows/ci.yml

@@ -224,7 +224,7 @@ jobs:
           name: diff-coverage-html
           path: index.html
 
-      - run: uv run coverage report --fail-under 95
+      - run: uv run coverage report --fail-under 100
       - run: uv run diff-cover coverage.xml --fail-under 100
       - run: uv run strict-no-cover
 

pydantic_ai_slim/pydantic_ai/_agent_graph.py

@@ -196,7 +196,9 @@ async def _reevaluate_dynamic_prompts(
                     for i, part in enumerate(msg.parts):
                         if isinstance(part, _messages.SystemPromptPart) and part.dynamic_ref:
                             # Look up the runner by its ref
-                            if runner := self.system_prompt_dynamic_functions.get(part.dynamic_ref):
+                            if runner := self.system_prompt_dynamic_functions.get(  # pragma: lax no cover
+                                part.dynamic_ref
+                            ):
                                 updated_part_content = await runner.run(run_context)
                                 msg.parts[i] = _messages.SystemPromptPart(
                                     updated_part_content, dynamic_ref=part.dynamic_ref
@@ -265,7 +267,7 @@ async def run(
         if self._did_stream:
             # `self._result` gets set when exiting the `stream` contextmanager, so hitting this
             # means that the stream was started but not finished before `run()` was called
-            raise exceptions.AgentRunError('You must finish streaming before calling run()')
+            raise exceptions.AgentRunError('You must finish streaming before calling run()')  # pragma: no cover
 
         return await self._make_request(ctx)
 
@@ -316,7 +318,7 @@ async def _make_request(
         self, ctx: GraphRunContext[GraphAgentState, GraphAgentDeps[DepsT, NodeRunEndT]]
     ) -> CallToolsNode[DepsT, NodeRunEndT]:
         if self._result is not None:
-            return self._result
+            return self._result  # pragma: no cover
 
         model_settings, model_request_parameters = await self._prepare_request(ctx)
         model_request_parameters = ctx.deps.model.customize_request_parameters(model_request_parameters)
@@ -333,7 +335,7 @@ async def _prepare_request(
         ctx.state.message_history.append(self.request)
 
         # Check usage
-        if ctx.deps.usage_limits:
+        if ctx.deps.usage_limits:  # pragma: no branch
             ctx.deps.usage_limits.check_before_request(ctx.state.usage)
 
         # Increment run_step
@@ -350,7 +352,7 @@ def _finish_handling(
     ) -> CallToolsNode[DepsT, NodeRunEndT]:
         # Update usage
         ctx.state.usage.incr(response.usage)
-        if ctx.deps.usage_limits:
+        if ctx.deps.usage_limits:  # pragma: no branch
             ctx.deps.usage_limits.check_tokens(ctx.state.usage)
 
         # Append the model response to state.message_history
@@ -735,7 +737,7 @@ async def run_tool(ctx: RunContext[DepsT], **args: Any) -> Any:
 
     for server in ctx.deps.mcp_servers:
         tools = await server.list_tools()
-        if tool_name in {tool.name for tool in tools}:
+        if tool_name in {tool.name for tool in tools}:  # pragma: no branch
             return Tool(name=tool_name, function=run_tool, takes_ctx=True, max_retries=ctx.deps.default_retries)
     return None
 

pydantic_ai_slim/pydantic_ai/_cli.py

@@ -202,7 +202,7 @@ def cli(args_list: Sequence[str] | None = None, *, prog_name: str = 'pai') -> in
     elif args.code_theme == 'dark':
         code_theme = 'monokai'
     else:
-        code_theme = args.code_theme
+        code_theme = args.code_theme  # pragma: no cover
 
     if prompt := cast(str, args.prompt):
         try:

pydantic_ai_slim/pydantic_ai/_output.py

@@ -140,8 +140,8 @@ def find_named_tool(
         self, parts: Iterable[_messages.ModelResponsePart], tool_name: str
     ) -> tuple[_messages.ToolCallPart, OutputSchemaTool[OutputDataT]] | None:
         """Find a tool that matches one of the calls, with a specific name."""
-        for part in parts:
-            if isinstance(part, _messages.ToolCallPart):
+        for part in parts:  # pragma: no branch
+            if isinstance(part, _messages.ToolCallPart):  # pragma: no branch
                 if part.tool_name == tool_name:
                     return part, self.tools[tool_name]
 
@@ -151,7 +151,7 @@ def find_tool(
     ) -> Iterator[tuple[_messages.ToolCallPart, OutputSchemaTool[OutputDataT]]]:
         """Find a tool that matches one of the calls."""
         for part in parts:
-            if isinstance(part, _messages.ToolCallPart):
+            if isinstance(part, _messages.ToolCallPart):  # pragma: no branch
                 if result := self.tools.get(part.tool_name):
                     yield part, result
 
@@ -201,7 +201,7 @@ def __init__(
             if description is None:
                 tool_description = json_schema_description
             else:
-                tool_description = f'{description}. {json_schema_description}'
+                tool_description = f'{description}. {json_schema_description}'  # pragma: no cover
         else:
             tool_description = description or DEFAULT_DESCRIPTION
             if multiple:
@@ -243,7 +243,7 @@ def validate(
                 )
                 raise ToolRetryError(m) from e
             else:
-                raise
+                raise  # pragma: lax no cover
         else:
             if k := self.tool_def.outer_typed_dict_key:
                 output = output[k]
@@ -269,11 +269,11 @@ def extract_str_from_union(output_type: Any) -> _utils.Option[Any]:
                 includes_str = True
             else:
                 remain_args.append(arg)
-        if includes_str:
+        if includes_str:  # pragma: no branch
             if len(remain_args) == 1:
                 return _utils.Some(remain_args[0])
             else:
-                return _utils.Some(Union[tuple(remain_args)])
+                return _utils.Some(Union[tuple(remain_args)])  # pragma: no cover
 
 
 def get_union_args(tp: Any) -> tuple[Any, ...]:

pydantic_ai_slim/pydantic_ai/_parts_manager.py

@@ -164,7 +164,7 @@ def handle_tool_call_delta(
             if tool_name is None and self._parts:
                 part_index = len(self._parts) - 1
                 latest_part = self._parts[part_index]
-                if isinstance(latest_part, (ToolCallPart, ToolCallPartDelta)):
+                if isinstance(latest_part, (ToolCallPart, ToolCallPartDelta)):  # pragma: no branch
                     existing_matching_part_and_index = latest_part, part_index
         else:
             # vendor_part_id is provided, so look up the corresponding part or delta

pydantic_ai_slim/pydantic_ai/agent.py

@@ -1002,7 +1002,7 @@ async def stream_to_final(
                         final_result_details = await stream_to_final(streamed_response)
                         if final_result_details is not None:
                             if yielded:
-                                raise exceptions.AgentRunError('Agent run produced final results')
+                                raise exceptions.AgentRunError('Agent run produced final results')  # pragma: no cover
                             yielded = True
 
                             messages = graph_ctx.state.message_history.copy()
@@ -1049,11 +1049,13 @@ async def on_complete() -> None:
                             break
                 next_node = await agent_run.next(node)
                 if not isinstance(next_node, _agent_graph.AgentNode):
-                    raise exceptions.AgentRunError('Should have produced a StreamedRunResult before getting here')
+                    raise exceptions.AgentRunError(  # pragma: no cover
+                        'Should have produced a StreamedRunResult before getting here'
+                    )
                 node = cast(_agent_graph.AgentNode[Any, Any], next_node)
 
         if not yielded:
-            raise exceptions.AgentRunError('Agent run finished without producing a final result')
+            raise exceptions.AgentRunError('Agent run finished without producing a final result')  # pragma: no cover
 
     @contextmanager
     def override(
@@ -1227,7 +1229,7 @@ def decorator(
             ) -> _system_prompt.SystemPromptFunc[AgentDepsT]:
                 runner = _system_prompt.SystemPromptRunner[AgentDepsT](func_, dynamic=dynamic)
                 self._system_prompt_functions.append(runner)
-                if dynamic:
+                if dynamic:  # pragma: lax no cover
                     self._system_prompt_dynamic_functions[func_.__qualname__] = runner
                 return func_
 
@@ -1609,7 +1611,7 @@ def _infer_name(self, function_frame: FrameType | None) -> None:
                     if item is self:
                         self.name = name
                         return
-                if parent_frame.f_locals != parent_frame.f_globals:
+                if parent_frame.f_locals != parent_frame.f_globals:  # pragma: no branch
                     # if we couldn't find the agent in locals and globals are a different dict, try globals
                     for name, item in parent_frame.f_globals.items():
                         if item is self:
@@ -2025,7 +2027,7 @@ def usage(self) -> _usage.Usage:
         """Get usage statistics for the run so far, including token usage, model requests, and so on."""
         return self._graph_run.state.usage
 
-    def __repr__(self) -> str:
+    def __repr__(self) -> str:  # pragma: no cover
         result = self._graph_run.result
         result_repr = '<run not finished>' if result is None else repr(result.output)
         return f'<{type(self).__name__} result={result_repr} usage={self.usage()}>'

pydantic_ai_slim/pydantic_ai/exceptions.py

@@ -4,9 +4,9 @@
 import sys
 
 if sys.version_info < (3, 11):
-    from exceptiongroup import ExceptionGroup
+    from exceptiongroup import ExceptionGroup  # pragma: lax no cover
 else:
-    ExceptionGroup = ExceptionGroup
+    ExceptionGroup = ExceptionGroup  # pragma: lax no cover
 
 __all__ = (
     'ModelRetry',

pydantic_ai_slim/pydantic_ai/messages.py

@@ -326,7 +326,7 @@ def otel_event(self, settings: InstrumentationSettings) -> Event:
                         converted_part['binary_content'] = base64.b64encode(part.data).decode()
                     content.append(converted_part)
                 else:
-                    content.append({'kind': part.kind})
+                    content.append({'kind': part.kind})  # pragma: no cover
         return Event('gen_ai.user.message', body={'content': content, 'role': 'user'})
 
 
@@ -363,7 +363,7 @@ def model_response_object(self) -> dict[str, Any]:
         """Return a dictionary representation of the content, wrapping non-dict types appropriately."""
         # gemini supports JSON dict return values, but no other JSON types, hence we wrap anything else in a dict
         if isinstance(self.content, dict):
-            return tool_return_ta.dump_python(self.content, mode='json')  # pyright: ignore[reportUnknownMemberType]
+            return tool_return_ta.dump_python(self.content, mode='json')  # pyright: ignore[reportUnknownMemberType]  # pragma: no cover
         else:
             return {'return_value': tool_return_ta.dump_python(self.content, mode='json')}
 
@@ -625,7 +625,7 @@ def apply(self, part: ModelResponsePart) -> TextPart:
             ValueError: If `part` is not a `TextPart`.
         """
         if not isinstance(part, TextPart):
-            raise ValueError('Cannot apply TextPartDeltas to non-TextParts')
+            raise ValueError('Cannot apply TextPartDeltas to non-TextParts')  # pragma: no cover
         return replace(part, content=part.content + self.content_delta)
 
 
@@ -688,7 +688,9 @@ def apply(self, part: ModelResponsePart | ToolCallPartDelta) -> ToolCallPart | T
         if isinstance(part, ToolCallPartDelta):
             return self._apply_to_delta(part)
 
-        raise ValueError(f'Can only apply ToolCallPartDeltas to ToolCallParts or ToolCallPartDeltas, not {part}')
+        raise ValueError(  # pragma: no cover
+            f'Can only apply ToolCallPartDeltas to ToolCallParts or ToolCallPartDeltas, not {part}'
+        )
 
     def _apply_to_delta(self, delta: ToolCallPartDelta) -> ToolCallPart | ToolCallPartDelta:
         """Internal helper to apply this delta to another delta."""

pydantic_ai_slim/pydantic_ai/models/__init__.py

@@ -485,7 +485,7 @@ def infer_model(model: Model | KnownModelName | str) -> Model:
             raise UserError(f'Unknown model: {model}')
 
     if provider == 'vertexai':
-        provider = 'google-vertex'
+        provider = 'google-vertex'  # pragma: no cover
 
     if provider == 'cohere':
         from .cohere import CohereModel
@@ -516,7 +516,7 @@ def infer_model(model: Model | KnownModelName | str) -> Model:
 
         return BedrockConverseModel(model_name, provider=provider)
     else:
-        raise UserError(f'Unknown model: {model}')
+        raise UserError(f'Unknown model: {model}')  # pragma: no cover
 
 
 def cached_async_http_client(*, provider: str | None = None, timeout: int = 600, connect: int = 5) -> httpx.AsyncClient:

pydantic_ai_slim/pydantic_ai/models/anthropic.py

@@ -244,7 +244,7 @@ async def _messages_create(
         except APIStatusError as e:
             if (status_code := e.status_code) >= 400:
                 raise ModelHTTPError(status_code=status_code, model_name=self.model_name, body=e.body) from e
-            raise
+            raise  # pragma: lax no cover
 
     def _process_response(self, response: AnthropicMessage) -> ModelResponse:
         """Process a non-streamed response, and prepare a message to return."""
@@ -268,7 +268,7 @@ async def _process_streamed_response(self, response: AsyncStream[RawMessageStrea
         peekable_response = _utils.PeekableAsyncStream(response)
         first_chunk = await peekable_response.peek()
         if isinstance(first_chunk, _utils.Unset):
-            raise UnexpectedModelBehavior('Streamed response ended without content or tool calls')
+            raise UnexpectedModelBehavior('Streamed response ended without content or tool calls')  # pragma: no cover
 
         # Since Anthropic doesn't provide a timestamp in the message, we'll use the current time
         timestamp = datetime.now(tz=timezone.utc)
@@ -305,9 +305,10 @@ async def _map_message(self, messages: list[ModelMessage]) -> tuple[str, list[Me
                             is_error=False,
                         )
                         user_content_params.append(tool_result_block_param)
-                    elif isinstance(request_part, RetryPromptPart):
+                    elif isinstance(request_part, RetryPromptPart):  # pragma: no branch
                         if request_part.tool_name is None:
-                            retry_param = TextBlockParam(type='text', text=request_part.model_response())
+                            text = request_part.model_response()  # pragma: no cover
+                            retry_param = TextBlockParam(type='text', text=text)  # pragma: no cover
                         else:
                             retry_param = ToolResultBlockParam(
                                 tool_use_id=_guard_tool_call_id(t=request_part),
@@ -380,7 +381,7 @@ async def _map_user_prompt(
                     else:  # pragma: no cover
                         raise RuntimeError(f'Unsupported media type: {item.media_type}')
                 else:
-                    raise RuntimeError(f'Unsupported content type: {type(item)}')
+                    raise RuntimeError(f'Unsupported content type: {type(item)}')  # pragma: no cover
 
     @staticmethod
     def _map_tool_definition(f: ToolDefinition) -> ToolParam:
@@ -447,21 +448,25 @@ async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
             if isinstance(event, RawContentBlockStartEvent):
                 current_block = event.content_block
                 if isinstance(current_block, TextBlock) and current_block.text:
-                    yield self._parts_manager.handle_text_delta(vendor_part_id='content', content=current_block.text)
-                elif isinstance(current_block, ToolUseBlock):
+                    yield self._parts_manager.handle_text_delta(  # pragma: lax no cover
+                        vendor_part_id='content', content=current_block.text
+                    )
+                elif isinstance(current_block, ToolUseBlock):  # pragma: no branch
                     maybe_event = self._parts_manager.handle_tool_call_delta(
                         vendor_part_id=current_block.id,
                         tool_name=current_block.name,
                         args=cast(dict[str, Any], current_block.input),
                         tool_call_id=current_block.id,
                     )
-                    if maybe_event is not None:
+                    if maybe_event is not None:  # pragma: no branch
                         yield maybe_event
 
             elif isinstance(event, RawContentBlockDeltaEvent):
                 if isinstance(event.delta, TextDelta):
-                    yield self._parts_manager.handle_text_delta(vendor_part_id='content', content=event.delta.text)
-                elif (
+                    yield self._parts_manager.handle_text_delta(  # pragma: no cover
+                        vendor_part_id='content', content=event.delta.text
+                    )
+                elif (  # pragma: no branch
                     current_block and event.delta.type == 'input_json_delta' and isinstance(current_block, ToolUseBlock)
                 ):
                     # Try to parse the JSON immediately, otherwise cache the value for later. This handles
@@ -480,7 +485,7 @@ async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
                         args=parsed_args,
                         tool_call_id=current_block.id,
                     )
-                    if maybe_event is not None:
+                    if maybe_event is not None:  # pragma: no branch
                         yield maybe_event
 
             elif isinstance(event, (RawContentBlockStopEvent, RawMessageStopEvent)):