Coverage for tests/test_utils.py: 98.31%

1from __future__ import annotations as _annotations (empty)

2 

3import asyncio (empty)

4import os (empty)

5from collections.abc import AsyncIterator (empty)

6from importlib.metadata import distributions (empty)

7 

8import pytest (empty)

9from inline_snapshot import snapshot (empty)

10 

11from pydantic_ai import UserError (empty)

12from pydantic_ai._utils import UNSET, PeekableAsyncStream, check_object_json_schema, group_by_temporal (empty)

13 

14from .models.mock_async_stream import MockAsyncStream (empty)

15 

16pytestmark = pytest.mark.anyio (empty)

17 

18 

19@pytest.mark.parametrize( (empty)

20 'interval,expected', 

21 [ 

22 (None, snapshot([[1], [2], [3]])), 

23 (0, snapshot([[1], [2], [3]])), 

24 (0.02, snapshot([[1], [2], [3]])), 

25 (0.04, snapshot([[1, 2], [3]])), 

26 (0.1, snapshot([[1, 2, 3]])), 

27 ], 

28) 

29async def test_group_by_temporal(interval: float | None, expected: list[list[int]]): (empty)

30 async def yield_groups() -> AsyncIterator[int]: (empty)

31 yield 1 (empty)

32 await asyncio.sleep(0.02) (empty)

33 yield 2 (empty)

34 await asyncio.sleep(0.02) (empty)

35 yield 3 (empty)

36 await asyncio.sleep(0.02) (empty)

37 

38 async with group_by_temporal(yield_groups(), soft_max_interval=interval) as groups_iter: (empty)

39 groups: list[list[int]] = [g async for g in groups_iter] (empty)

40 assert groups == expected (empty)

41 

42 

43def test_check_object_json_schema(): (empty)

44 object_schema = {'type': 'object', 'properties': {'a': {'type': 'string'}}} (empty)

45 assert check_object_json_schema(object_schema) == object_schema (empty)

46 

47 array_schema = {'type': 'array', 'items': {'type': 'string'}} (empty)

48 with pytest.raises(UserError, match='^Schema must be an object$'): (empty)

49 check_object_json_schema(array_schema) (empty)

50 

51 

52@pytest.mark.parametrize('peek_first', [True, False]) (empty)

53@pytest.mark.anyio (empty)

54async def test_peekable_async_stream(peek_first: bool): (empty)

55 async_stream = MockAsyncStream(iter([1, 2, 3])) (empty)

56 peekable_async_stream = PeekableAsyncStream(async_stream) (empty)

57 

58 items: list[int] = [] (empty)

59 

60 # We need to both peek before starting the stream, and not, to achieve full coverage 

61 if peek_first: (empty)

62 assert not await peekable_async_stream.is_exhausted() (empty)

63 assert await peekable_async_stream.peek() == 1 (empty)

64 

65 async for item in peekable_async_stream: (empty)

66 items.append(item) (empty)

67 

68 # The next line is included mostly for the sake of achieving coverage 

69 assert await peekable_async_stream.peek() == (item + 1 if item < 3 else UNSET) (empty)

70 

71 assert await peekable_async_stream.is_exhausted() (empty)

72 assert await peekable_async_stream.peek() is UNSET (empty)

73 assert items == [1, 2, 3] (empty)

74 

75 

76def test_package_versions(capsys: pytest.CaptureFixture[str]): (empty)

77 if os.getenv('CI'): 77 ↛ exitline 77 didn't return from function 'test_package_versions' because the condition on line 77 was always true(empty)

78 with capsys.disabled(): (empty)

79 print('\npackage versions:') (empty)

80 packages = sorted((package.metadata['Name'], package.version) for package in distributions()) (empty)

81 for name, version in packages: (empty)

82 print(f'{name:30} {version}') (empty)