Coverage for tests / utils / context / test_main.py: 99%

1import pytest (empty)

2from fast_depends import ValidationError (empty)

3 

4from faststream import Context, ContextRepo (empty)

5from faststream._internal.utils import apply_types (empty)

6 

7 

8def test_context_getattr(context: ContextRepo) -> None: (empty)

9 a = 1000 (empty)

10 context.set_global("key", a) (empty)

11 

12 assert context.key is a (empty)

13 assert context.key2 is None (empty)

14 

15 

16@pytest.mark.asyncio() (empty)

17async def test_context_apply(context: ContextRepo) -> None: (empty)

18 a = 1000 (empty)

19 context.set_global("key", a) (empty)

20 

21 @apply_types(context__=context) (empty)

22 async def use(key=Context()): (empty)

23 return key is a (empty)

24 

25 assert await use() (empty)

26 

27 

28@pytest.mark.asyncio() (empty)

29async def test_context_ignore(context: ContextRepo) -> None: (empty)

30 a = 3 (empty)

31 context.set_global("key", a) (empty)

32 

33 @apply_types(context__=context) (empty)

34 async def use() -> None: (empty)

35 return None (empty)

36 

37 assert await use() is None (empty)

38 

39 

40@pytest.mark.asyncio() (empty)

41async def test_context_apply_multi(context: ContextRepo) -> None: (empty)

42 a = 1001 (empty)

43 context.set_global("key_a", a) (empty)

44 

45 b = 1000 (empty)

46 context.set_global("key_b", b) (empty)

47 

48 @apply_types(context__=context) (empty)

49 async def use1(key_a=Context()): (empty)

50 return key_a is a (empty)

51 

52 assert await use1() (empty)

53 

54 @apply_types(context__=context) (empty)

55 async def use2(key_b=Context()): (empty)

56 return key_b is b (empty)

57 

58 assert await use2() (empty)

59 

60 @apply_types(context__=context) (empty)

61 async def use3(key_a=Context(), key_b=Context()): (empty)

62 return key_a is a and key_b is b (empty)

63 

64 assert await use3() (empty)

65 

66 

67@pytest.mark.asyncio() (empty)

68async def test_context_overrides(context: ContextRepo) -> None: (empty)

69 a = 1001 (empty)

70 context.set_global("test", a) (empty)

71 

72 b = 1000 (empty)

73 context.set_global("test", b) (empty)

74 

75 @apply_types(context__=context) (empty)

76 async def use(test=Context()): (empty)

77 return test is b (empty)

78 

79 assert await use() (empty)

80 

81 

82@pytest.mark.asyncio() (empty)

83async def test_context_nested_apply(context: ContextRepo) -> None: (empty)

84 a = 1000 (empty)

85 context.set_global("key", a) (empty)

86 

87 @apply_types(context__=context) (empty)

88 def use_nested(key=Context()): (empty)

89 return key (empty)

90 

91 @apply_types(context__=context) (empty)

92 async def use(key=Context()): (empty)

93 return key is use_nested() is a (empty)

94 

95 assert await use() (empty)

96 

97 

98@pytest.mark.asyncio() (empty)

99async def test_reset_global(context: ContextRepo) -> None: (empty)

100 a = 1000 (empty)

101 context.set_global("key", a) (empty)

102 context.reset_global("key") (empty)

103 

104 @apply_types(context__=context) (empty)

105 async def use(key=Context()) -> None: ... (empty)

106 

107 with pytest.raises(ValidationError): (empty)

108 await use() (empty)

109 

110 

111@pytest.mark.asyncio() (empty)

112async def test_clear_context(context: ContextRepo) -> None: (empty)

113 a = 1000 (empty)

114 context.set_global("key", a) (empty)

115 context.clear() (empty)

116 

117 @apply_types(context__=context) (empty)

118 async def use(key=Context(default=None)): (empty)

119 return key is None (empty)

120 

121 assert await use() (empty)

122 

123 

124def test_scope(context: ContextRepo) -> None: (empty)

125 @apply_types(context__=context) (empty)

126 def use(key=Context(), key2=Context()) -> None: (empty)

127 assert key == 1 (empty)

128 assert key2 == 1 (empty)

129 

130 with context.scope("key", 1), context.scope("key2", 1): (empty)

131 use() (empty)

132 

133 assert context.get("key") is None (empty)

134 assert context.get("key2") is None (empty)

135 

136 

137def test_default(context: ContextRepo) -> None: (empty)

138 @apply_types(context__=context) (empty)

139 def use( (empty)

140 key=Context(), 

141 key2=Context(), 

142 key3=Context(default=1), 

143 key4=Context("key.key4", default=1), 

144 key5=Context("key5.key6"), 

145 ) -> None: 

146 assert key == 0 (empty)

147 assert key2 is True (empty)

148 assert key3 == 1 (empty)

149 assert key4 == 1 (empty)

150 assert key5 is False (empty)

151 

152 with ( (empty)

153 context.scope("key", 0), 

154 context.scope("key2", True), 

155 context.scope( 

156 "key5", 

157 {"key6": False}, 

158 ), 

159 ): 

160 use() (empty)

161 

162 

163def test_local_default(context: ContextRepo) -> None: (empty)

164 key = "some-key" (empty)

165 

166 tag = context.set_local(key, "useless") (empty)

167 context.reset_local(key, tag) (empty)

168 

169 assert context.get_local(key, 1) == 1 (empty)

170 

171 

172def test_initial(context: ContextRepo) -> None: (empty)

173 @apply_types(context__=context) (empty)

174 def use( (empty)

175 a, 

176 key=Context(initial=list), 

177 ): 

178 key.append(a) (empty)

179 return key (empty)

180 

181 assert use(1) == [1] (empty)

182 assert use(2) == [1, 2] (empty)

183 

184 

185@pytest.mark.asyncio() (empty)

186async def test_context_with_custom_object_implementing_comparison( (empty)

187 context: ContextRepo, 

188) -> None: 

189 class User: (empty)

190 def __init__(self, user_id: int) -> None: (empty)

191 self.user_id = user_id (empty)

192 

193 def __eq__(self, other): (empty)

194 if not isinstance(other, User): (empty)

195 return NotImplemented 

196 return self.user_id == other.user_id (empty)

197 

198 def __ne__(self, other): (empty)

199 return not self.__eq__(other) 

200 

201 user2 = User(user_id=2) (empty)

202 user3 = User(user_id=3) (empty)

203 

204 @apply_types(context__=context) (empty)

205 async def use( (empty)

206 key1=Context("user1"), 

207 key2=Context("user2", default=user2), 

208 key3=Context("user3", default=user3), 

209 ): 

210 return ( (empty)

211 key1 == User(user_id=1) 

212 and key2 == User(user_id=2) 

213 and key3 == User(user_id=4) 

214 ) 

215 

216 with ( (empty)

217 context.scope("user1", User(user_id=1)), 

218 context.scope("user3", User(user_id=4)), 

219 ): 

220 assert await use() (empty)