Coverage for tests/library/test_custom.py: 100%

1import logging 20 ctx1abcdefghijklmnopqrst

2from time import monotonic_ns 20 ctx1abcdefghijklmnopqrst

3from typing import Annotated, Any 20 ctx1abcdefghijklmnopqrst

4 

5import anyio 20 ctx1abcdefghijklmnopqrst

6import pytest 20 ctx1abcdefghijklmnopqrst

7 

8from fast_depends import Depends, inject 20 ctx1abcdefghijklmnopqrst

9from fast_depends.exceptions import ValidationError 20 ctx1abcdefghijklmnopqrst

10from fast_depends.library import CustomField 20 ctx1abcdefghijklmnopqrst

11from tests.marks import serializer 20 ctx1abcdefghijklmnopqrst

12 

13 

14class Header(CustomField): 20 ctx1abcdefghijklmnopqrst

15 def use(self, /, **kwargs: Any) -> dict[str, Any]: 20 ctx1abcdefghijklmnopqrst

16 kwargs = super().use(**kwargs) 20 ctx1abcdefghijklmnopqrst

17 if v := kwargs.get("headers", {}).get(self.param_name): 20 ctx1abcdefghijklmnopqrst

18 kwargs[self.param_name] = v 20 ctx1abcdefghijklmnopqrst

19 return kwargs 20 ctx1abcdefghijklmnopqrst

20 

21 

22class FieldHeader(Header): 20 ctx1abcdefghijklmnopqrst

23 def __init__(self, *, cast: bool = True, required: bool = True) -> None: 20 ctx1abcdefghijklmnopqrst

24 super().__init__(cast=cast, required=required) 18 ctx1abcdefghijklmnopqr

25 self.field = True 18 ctx1abcdefghijklmnopqr

26 

27 def use_field(self, kwargs: Any) -> None: 20 ctx1abcdefghijklmnopqrst

28 if v := kwargs.get("headers", {}).get(self.param_name): # pragma: no branch 18 ctx1abcdefghijklmnopqr

29 kwargs[self.param_name] = v 18 ctx1abcdefghijklmnopqr

30 

31 

32class AsyncHeader(Header): 20 ctx1abcdefghijklmnopqrst

33 async def use(self, /, **kwargs: Any) -> dict[str, Any]: 20 ctx1abcdefghijklmnopqrst

34 return super().use(**kwargs) 18 ctx1abcdefghijklmnopqr

35 

36 

37class AsyncFieldHeader(Header): 20 ctx1abcdefghijklmnopqrst

38 def __init__(self, *, cast: bool = True, required: bool = True) -> None: 20 ctx1abcdefghijklmnopqrst

39 super().__init__(cast=cast, required=required) 18 ctx1abcdefghijklmnopqr

40 self.field = True 18 ctx1abcdefghijklmnopqr

41 

42 async def use_field(self, kwargs: Any) -> None: 20 ctx1abcdefghijklmnopqrst

43 await anyio.sleep(0.1) 18 ctx1abcdefghijklmnopqr

44 if v := kwargs.get("headers", {}).get(self.param_name): # pragma: no branch 18 ctx1abcdefghijklmnopqr

45 kwargs[self.param_name] = v 18 ctx1abcdefghijklmnopqr

46 

47 

48class FailingAsyncFieldHeader(Header): 20 ctx1abcdefghijklmnopqrst

49 def __init__(self) -> None: 20 ctx1abcdefghijklmnopqrst

50 super().__init__() 20 ctx1abcdefghijklmnopqrst

51 self.field = True 20 ctx1abcdefghijklmnopqrst

52 

53 async def use_field(self, kwargs: Any) -> None: 20 ctx1abcdefghijklmnopqrst

54 raise ValueError("failed to resolve field") 20 ctx1abcdefghijklmnopqrst

55 

56 

57def test_header(): 20 ctx1abcdefghijklmnopqrst

58 @inject 20 ctx1abcdefghijklmnopqrst

59 def sync_catch(key: int = Header()): # noqa: B008 20 ctx1abcdefghijklmnopqrst

60 return key 20 ctx1abcdefghijklmnopqrst

61 

62 assert sync_catch(headers={"key": 1}) == 1 20 ctx1abcdefghijklmnopqrst

63 

64 

65@pytest.mark.anyio 20 ctx1abcdefghijklmnopqrst

66async def test_async_field_header_error_is_unwrapped(): 20 ctx1abcdefghijklmnopqrst

67 @inject 20 ctx1abcdefghijklmnopqrst

68 async def async_catch(key=FailingAsyncFieldHeader()): # noqa: B008 20 ctx1abcdefghijklmnopqrst

69 raise AssertionError("unreachable") 

70 

71 # `field=True` customs are resolved inside an `anyio` task group, which wraps 

72 # failures into an `ExceptionGroup` - the original error must be re-raised as is 

73 with pytest.raises(ValueError, match="failed to resolve field"): 20 ctx1abcdefghijklmnopqrst

74 await async_catch(headers={"key": 1}) 20 ctx1abcdefghijklmnopqrst

75 

76 

77def test_custom_with_class(): 20 ctx1abcdefghijklmnopqrst

78 class T: 20 ctx1abcdefghijklmnopqrst

79 @inject 20 ctx1abcdefghijklmnopqrst

80 def __init__(self, key: int = Header()): 20 ctx1abcdefghijklmnopqrst

81 self.key = key 20 ctx1abcdefghijklmnopqrst

82 

83 assert T(headers={"key": 1}).key == 1 20 ctx1abcdefghijklmnopqrst

84 

85 

86def test_reusable_annotated() -> None: 20 ctx1abcdefghijklmnopqrst

87 HeaderKey = Annotated[float, Header(cast=False)] 20 ctx1abcdefghijklmnopqrst

88 

89 @inject 20 ctx1abcdefghijklmnopqrst

90 def sync_catch(key: HeaderKey) -> float: 20 ctx1abcdefghijklmnopqrst

91 return key 20 ctx1abcdefghijklmnopqrst

92 

93 @inject 20 ctx1abcdefghijklmnopqrst

94 def sync_catch2(key2: HeaderKey) -> float: 20 ctx1abcdefghijklmnopqrst

95 return key2 20 ctx1abcdefghijklmnopqrst

96 

97 assert sync_catch(headers={"key": 1}) == 1 20 ctx1abcdefghijklmnopqrst

98 assert sync_catch2(headers={"key2": 1}) == 1 20 ctx1abcdefghijklmnopqrst

99 

100 

101def test_arguments_mapping(): 20 ctx1abcdefghijklmnopqrst

102 @inject 20 ctx1abcdefghijklmnopqrst

103 def func( 20 ctx1abcdefghijklmnopqrst

104 d: int = CustomField(cast=False), 

105 b: int = CustomField(cast=False), 

106 c: int = CustomField(cast=False), 

107 a: int = CustomField(cast=False), 

108 ): 

109 assert d == 4 20 ctx1abcdefghijklmnopqrst

110 assert b == 2 20 ctx1abcdefghijklmnopqrst

111 assert c == 3 20 ctx1abcdefghijklmnopqrst

112 assert a == 1 20 ctx1abcdefghijklmnopqrst

113 

114 for _ in range(50): 20 ctx1abcdefghijklmnopqrst

115 func(4, 2, 3, 1) 20 ctx1abcdefghijklmnopqrst

116 

117 

118@serializer 20 ctx1abcdefghijklmnopqrst

119class TestSerializer: 20 ctx1abcdefghijklmnopqrst

120 @pytest.mark.anyio 20 ctx1abcdefghijklmnopqrst

121 async def test_header_async(self): 20 ctx1abcdefghijklmnopqrst

122 @inject 18 ctx1abcdefghijklmnopqr

123 async def async_catch(key: int = Header()): # noqa: B008 18 ctx1abcdefghijklmnopqr

124 return key 18 ctx1abcdefghijklmnopqr

125 

126 assert (await async_catch(headers={"key": "1"})) == 1 18 ctx1abcdefghijklmnopqr

127 

128 def test_multiple_header(self): 20 ctx1abcdefghijklmnopqrst

129 @inject 18 ctx1abcdefghijklmnopqr

130 def sync_catch(key: str = Header(), key2: int = Header()): # noqa: B008 18 ctx1abcdefghijklmnopqr

131 assert key == "1" 18 ctx1abcdefghijklmnopqr

132 assert key2 == 2 18 ctx1abcdefghijklmnopqr

133 

134 sync_catch(headers={"key": "1", "key2": "2"}) 18 ctx1abcdefghijklmnopqr

135 

136 @pytest.mark.anyio 20 ctx1abcdefghijklmnopqrst

137 async def test_async_header_async(self): 20 ctx1abcdefghijklmnopqrst

138 @inject 18 ctx1abcdefghijklmnopqr

139 async def async_catch( # noqa: B008 18 ctx1abcdefghijklmnopqr

140 key: float = AsyncHeader(), key2: int = AsyncHeader() 

141 ): 

142 return key, key2 18 ctx1abcdefghijklmnopqr

143 

144 assert (await async_catch(headers={"key": "1", "key2": 1})) == (1.0, 1) 18 ctx1abcdefghijklmnopqr

145 

146 def test_sync_field_header(self): 20 ctx1abcdefghijklmnopqrst

147 @inject 18 ctx1abcdefghijklmnopqr

148 def sync_catch(key: float = FieldHeader(), key2: int = FieldHeader()): # noqa: B008 18 ctx1abcdefghijklmnopqr

149 return key, key2 18 ctx1abcdefghijklmnopqr

150 

151 assert sync_catch(headers={"key": "1", "key2": 1}) == (1.0, 1) 18 ctx1abcdefghijklmnopqr

152 

153 @pytest.mark.anyio 20 ctx1abcdefghijklmnopqrst

154 async def test_async_field_header(self): 20 ctx1abcdefghijklmnopqrst

155 @inject 18 ctx1abcdefghijklmnopqr

156 async def async_catch( # noqa: B008 18 ctx1abcdefghijklmnopqr

157 key: float = AsyncFieldHeader(), key2: int = AsyncFieldHeader() 

158 ): 

159 return key, key2 18 ctx1abcdefghijklmnopqr

160 

161 start = monotonic_ns() 18 ctx1abcdefghijklmnopqr

162 assert (await async_catch(headers={"key": "1", "key2": 1})) == (1.0, 1) 18 ctx1abcdefghijklmnopqr

163 assert (monotonic_ns() - start) / 10**9 < 0.2 18 ctx1abcdefghijklmnopqr

164 

165 def test_async_header_sync(self): 20 ctx1abcdefghijklmnopqrst

166 with pytest.raises(AssertionError): 18 ctx1abcdefghijklmnopqr

167 

168 @inject 18 ctx1abcdefghijklmnopqr

169 def sync_catch(key: str = AsyncHeader()): # pragma: no cover # noqa: B008 18 ctx1abcdefghijklmnopqr

170 return key 

171 

172 def test_header_annotated(self): 20 ctx1abcdefghijklmnopqrst

173 @inject 18 ctx1abcdefghijklmnopqr

174 def sync_catch(key: Annotated[int, Header()]): 18 ctx1abcdefghijklmnopqr

175 return key 18 ctx1abcdefghijklmnopqr

176 

177 assert sync_catch(headers={"key": 1}) == 1 18 ctx1abcdefghijklmnopqr

178 

179 def test_header_required(self): 20 ctx1abcdefghijklmnopqrst

180 @inject 18 ctx1abcdefghijklmnopqr

181 def sync_catch(key=Header()): # pragma: no cover # noqa: B008 18 ctx1abcdefghijklmnopqr

182 return key 

183 

184 with pytest.raises(ValidationError): 18 ctx1abcdefghijklmnopqr

185 sync_catch() 18 ctx1abcdefghijklmnopqr

186 

187 def test_header_not_required(self): 20 ctx1abcdefghijklmnopqrst

188 @inject 18 ctx1abcdefghijklmnopqr

189 def sync_catch(key2=Header(required=False)): # noqa: B008 18 ctx1abcdefghijklmnopqr

190 assert key2 is None 18 ctx1abcdefghijklmnopqr

191 

192 sync_catch() 18 ctx1abcdefghijklmnopqr

193 

194 def test_header_not_required_with_default(self): 20 ctx1abcdefghijklmnopqrst

195 @inject 18 ctx1abcdefghijklmnopqr

196 def sync_catch(key2: Annotated[str, Header(required=False)] = "1"): # noqa: B008 18 ctx1abcdefghijklmnopqr

197 return key2 == "1" 18 ctx1abcdefghijklmnopqr

198 

199 assert sync_catch() 18 ctx1abcdefghijklmnopqr

200 

201 def test_depends(self): 20 ctx1abcdefghijklmnopqrst

202 def dep(key: Annotated[int, Header()]): 18 ctx1abcdefghijklmnopqr

203 return key 18 ctx1abcdefghijklmnopqr

204 

205 @inject 18 ctx1abcdefghijklmnopqr

206 def sync_catch(k=Depends(dep)): 18 ctx1abcdefghijklmnopqr

207 return k 18 ctx1abcdefghijklmnopqr

208 

209 assert sync_catch(headers={"key": 1}) == 1 18 ctx1abcdefghijklmnopqr

210 

211 def test_not_cast(self): 20 ctx1abcdefghijklmnopqrst

212 @inject 18 ctx1abcdefghijklmnopqr

213 def sync_catch(key: Annotated[float, Header(cast=False)]): 18 ctx1abcdefghijklmnopqr

214 return key 18 ctx1abcdefghijklmnopqr

215 

216 assert sync_catch(headers={"key": 1}) == 1 18 ctx1abcdefghijklmnopqr

217 

218 @inject 18 ctx1abcdefghijklmnopqr

219 def sync_catch(key: logging.Logger = Header(cast=False)): # noqa: B008 18 ctx1abcdefghijklmnopqr

220 return key 18 ctx1abcdefghijklmnopqr

221 

222 assert sync_catch(headers={"key": 1}) == 1 18 ctx1abcdefghijklmnopqr