Coverage for tests/conftest.py: 98.98%

1from __future__ import annotations as _annotations (empty)

2 

3import asyncio (empty)

4import importlib.util (empty)

5import os (empty)

6import re (empty)

7import secrets (empty)

8import sys (empty)

9from collections.abc import AsyncIterator, Iterator (empty)

10from contextlib import contextmanager (empty)

11from datetime import datetime (empty)

12from pathlib import Path (empty)

13from types import ModuleType (empty)

14from typing import TYPE_CHECKING, Any, Callable (empty)

15 

16import httpx (empty)

17import pytest (empty)

18from _pytest.assertion.rewrite import AssertionRewritingHook (empty)

19from typing_extensions import TypeAlias (empty)

20 

21import pydantic_ai.models (empty)

22 

23__all__ = 'IsNow', 'IsFloat', 'TestEnv', 'ClientWithHandler', 'try_import' (empty)

24 

25 

26pydantic_ai.models.ALLOW_MODEL_REQUESTS = False (empty)

27 

28if TYPE_CHECKING: (empty)

29 

30 def IsNow(*args: Any, **kwargs: Any) -> datetime: ... 

31 def IsFloat(*args: Any, **kwargs: Any) -> float: ... 

32else: 

33 from dirty_equals import IsFloat, IsNow as _IsNow (empty)

34 

35 def IsNow(*args: Any, **kwargs: Any): (empty)

36 # Increase the default value of `delta` to 10 to reduce test flakiness on overburdened machines 

37 if 'delta' not in kwargs: 37 ↛ 39line 37 didn't jump to line 39 because the condition on line 37 was always true(empty)

38 kwargs['delta'] = 10 (empty)

39 return _IsNow(*args, **kwargs) (empty)

40 

41 

42try: (empty)

43 from logfire.testing import CaptureLogfire (empty)

44except ImportError: (empty)

45 pass (empty)

46else: 

47 

48 @pytest.fixture(autouse=True) (empty)

49 def logfire_disable(capfire: CaptureLogfire): (empty)

50 pass (empty)

51 

52 

53class TestEnv: (empty)

54 __test__ = False (empty)

55 

56 def __init__(self): (empty)

57 self.envars: dict[str, str | None] = {} (empty)

58 

59 def set(self, name: str, value: str) -> None: (empty)

60 self.envars[name] = os.getenv(name) (empty)

61 os.environ[name] = value (empty)

62 

63 def remove(self, name: str) -> None: (empty)

64 self.envars[name] = os.environ.pop(name, None) (empty)

65 

66 def reset(self) -> None: # pragma: no cover (empty)

67 for name, value in self.envars.items(): (empty)

68 if value is None: (empty)

69 os.environ.pop(name, None) (empty)

70 else: 

71 os.environ[name] = value 

72 

73 

74@pytest.fixture (empty)

75def env() -> Iterator[TestEnv]: (empty)

76 test_env = TestEnv() (empty)

77 

78 yield test_env (empty)

79 

80 test_env.reset() (empty)

81 

82 

83@pytest.fixture (empty)

84def anyio_backend(): (empty)

85 return 'asyncio' (empty)

86 

87 

88@pytest.fixture (empty)

89def allow_model_requests(): (empty)

90 with pydantic_ai.models.override_allow_model_requests(True): (empty)

91 yield (empty)

92 

93 

94@pytest.fixture (empty)

95async def client_with_handler() -> AsyncIterator[ClientWithHandler]: (empty)

96 client: httpx.AsyncClient | None = None (empty)

97 

98 def create_client(handler: Callable[[httpx.Request], httpx.Response]) -> httpx.AsyncClient: (empty)

99 nonlocal client 

100 assert client is None, 'client_with_handler can only be called once' (empty)

101 client = httpx.AsyncClient(mounts={'all://': httpx.MockTransport(handler)}) (empty)

102 return client (empty)

103 

104 try: (empty)

105 yield create_client (empty)

106 finally: 

107 if client: # pragma: no cover (empty)

108 await client.aclose() (empty)

109 

110 

111ClientWithHandler: TypeAlias = Callable[[Callable[[httpx.Request], httpx.Response]], httpx.AsyncClient] (empty)

112 

113 

114# pyright: reportUnknownMemberType=false, reportUnknownArgumentType=false 

115@pytest.fixture (empty)

116def create_module(tmp_path: Path, request: pytest.FixtureRequest) -> Callable[[str], Any]: (empty)

117 """Taken from `pydantic/tests/conftest.py`, create module object, execute and return it.""" 

118 

119 def run( (empty)

120 source_code: str, 

121 rewrite_assertions: bool = True, 

122 module_name_prefix: str | None = None, 

123 ) -> ModuleType: 

124 """Create module object, execute and return it. 

125 

126 Can be used as a decorator of the function from the source code of which the module will be constructed. 

127 

128 Args: 

129 source_code: Python source code of the module 

130 rewrite_assertions: whether to rewrite assertions in module or not 

131 module_name_prefix: string prefix to use in the name of the module, does not affect the name of the file. 

132 

133 """ 

134 

135 # Max path length in Windows is 260. Leaving some buffer here 

136 max_name_len = 240 - len(str(tmp_path)) (empty)

137 # Windows does not allow these characters in paths. Linux bans slashes only. 

138 sanitized_name = re.sub('[' + re.escape('<>:"/\\|?*') + ']', '-', request.node.name)[:max_name_len] (empty)

139 module_name = f'{sanitized_name}_{secrets.token_hex(5)}' (empty)

140 path = tmp_path / f'{module_name}.py' (empty)

141 path.write_text(source_code) (empty)

142 filename = str(path) (empty)

143 

144 if module_name_prefix: # pragma: no cover (empty)

145 module_name = module_name_prefix + module_name 

146 

147 if rewrite_assertions: (empty)

148 loader = AssertionRewritingHook(config=request.config) (empty)

149 loader.mark_rewrite(module_name) (empty)

150 else: # pragma: no cover 

151 loader = None 

152 

153 spec = importlib.util.spec_from_file_location(module_name, filename, loader=loader) (empty)

154 sys.modules[module_name] = module = importlib.util.module_from_spec(spec) # pyright: ignore[reportArgumentType] (empty)

155 spec.loader.exec_module(module) # pyright: ignore[reportOptionalMemberAccess] (empty)

156 return module (empty)

157 

158 return run (empty)

159 

160 

161@contextmanager (empty)

162def try_import() -> Iterator[Callable[[], bool]]: (empty)

163 import_success = False (empty)

164 

165 def check_import() -> bool: (empty)

166 return import_success (empty)

167 

168 try: (empty)

169 yield check_import (empty)

170 except ImportError: (empty)

171 pass (empty)

172 else: 

173 import_success = True (empty)

174 

175 

176@pytest.fixture(autouse=True) (empty)

177def set_event_loop() -> Iterator[None]: (empty)

178 new_loop = asyncio.new_event_loop() (empty)

179 asyncio.set_event_loop(new_loop) (empty)

180 yield (empty)

181 new_loop.close() (empty)