Coverage for tests/test_utils.py: 98.36%

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 ref_schema = { (empty)

48 '$defs': { 

49 'JsonModel': { 

50 'properties': { 

51 'type': {'title': 'Type', 'type': 'string'}, 

52 'items': {'anyOf': [{'$ref': '#/$defs/JsonModel'}, {'type': 'null'}]}, 

53 }, 

54 'required': ['type', 'items'], 

55 'title': 'JsonModel', 

56 'type': 'object', 

57 } 

58 }, 

59 '$ref': '#/$defs/JsonModel', 

60 } 

61 assert check_object_json_schema(ref_schema) == { (empty)

62 'properties': { 

63 'type': {'title': 'Type', 'type': 'string'}, 

64 'items': {'anyOf': [{'$ref': '#/$defs/JsonModel'}, {'type': 'null'}]}, 

65 }, 

66 'required': ['type', 'items'], 

67 'title': 'JsonModel', 

68 'type': 'object', 

69 } 

70 

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

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

73 check_object_json_schema(array_schema) (empty)

74 

75 

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

77@pytest.mark.anyio (empty)

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

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

80 peekable_async_stream = PeekableAsyncStream(async_stream) (empty)

81 

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

83 

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

85 if peek_first: (empty)

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

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

88 

89 async for item in peekable_async_stream: (empty)

90 items.append(item) (empty)

91 

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

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

94 

95 assert await peekable_async_stream.is_exhausted() (empty)

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

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

98 

99 

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

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

102 with capsys.disabled(): (empty)

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

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

105 for name, version in packages: (empty)

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