Coverage for tests / brokers / nats / test_test_client.py: 97%

1import asyncio (empty)

2 

3import pytest (empty)

4 

5from faststream import BaseMiddleware (empty)

6from faststream.nats import ( (empty)

7 ConsumerConfig, 

8 JStream, 

9 PullSub, 

10) 

11from faststream.nats.testing import FakeProducer (empty)

12from tests.brokers.base.testclient import BrokerTestclientTestcase (empty)

13 

14from .basic import NatsMemoryTestcaseConfig (empty)

15 

16 

17@pytest.mark.nats() (empty)

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

19class TestTestclient(NatsMemoryTestcaseConfig, BrokerTestclientTestcase): (empty)

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

21 async def test_stream_publish( (empty)

22 self, 

23 queue: str, 

24 ) -> None: 

25 pub_broker = self.get_broker(apply_types=False) (empty)

26 

27 @pub_broker.subscriber(queue, stream="test") (empty)

28 async def m(msg) -> None: ... (empty)

29 

30 async with self.patch_broker(pub_broker) as br: (empty)

31 await br.publish("Hi!", queue, stream="test") (empty)

32 m.mock.assert_called_once_with("Hi!") (empty)

33 

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

35 async def test_wrong_stream_publish( (empty)

36 self, 

37 queue: str, 

38 ) -> None: 

39 pub_broker = self.get_broker(apply_types=False) (empty)

40 

41 @pub_broker.subscriber(queue) (empty)

42 async def m(msg) -> None: ... (empty)

43 

44 async with self.patch_broker(pub_broker) as br: (empty)

45 await br.publish("Hi!", queue, stream="test") (empty)

46 assert not m.mock.called (empty)

47 

48 @pytest.mark.connected() (empty)

49 async def test_with_real_testclient( (empty)

50 self, 

51 queue: str, 

52 ) -> None: 

53 event = asyncio.Event() (empty)

54 

55 broker = self.get_broker() (empty)

56 

57 @broker.subscriber(queue) (empty)

58 def subscriber(m) -> None: (empty)

59 event.set() (empty)

60 

61 async with self.patch_broker(broker, with_real=True) as br: (empty)

62 await asyncio.wait( (empty)

63 ( 

64 asyncio.create_task(br.publish("hello", queue)), 

65 asyncio.create_task(event.wait()), 

66 ), 

67 timeout=3, 

68 ) 

69 

70 assert event.is_set() (empty)

71 

72 @pytest.mark.connected() (empty)

73 async def test_inbox_prefix_with_real( (empty)

74 self, 

75 queue: str, 

76 ) -> None: 

77 broker = self.get_broker(inbox_prefix="test") (empty)

78 

79 async with self.patch_broker(broker, with_real=True) as br: (empty)

80 assert br._connection._inbox_prefix == b"test" (empty)

81 assert "test" in str(br._connection.new_inbox()) (empty)

82 

83 async def test_respect_middleware(self, queue: str) -> None: (empty)

84 routes = [] (empty)

85 

86 class Middleware(BaseMiddleware): (empty)

87 async def on_receive(self) -> None: (empty)

88 routes.append(None) (empty)

89 return await super().on_receive() (empty)

90 

91 broker = self.get_broker(middlewares=(Middleware,)) (empty)

92 

93 @broker.subscriber(queue) (empty)

94 async def h1(m) -> None: ... (empty)

95 

96 @broker.subscriber(queue + "1") (empty)

97 async def h2(m) -> None: ... (empty)

98 

99 async with self.patch_broker(broker) as br: (empty)

100 await br.publish("", queue) (empty)

101 await br.publish("", queue + "1") (empty)

102 

103 assert len(routes) == 2 (empty)

104 

105 @pytest.mark.connected() (empty)

106 async def test_real_respect_middleware(self, queue: str) -> None: (empty)

107 routes = [] (empty)

108 

109 class Middleware(BaseMiddleware): (empty)

110 async def on_receive(self) -> None: (empty)

111 routes.append(None) (empty)

112 return await super().on_receive() (empty)

113 

114 broker = self.get_broker(middlewares=(Middleware,)) (empty)

115 

116 @broker.subscriber(queue) (empty)

117 async def h1(m) -> None: ... (empty)

118 

119 @broker.subscriber(queue + "1") (empty)

120 async def h2(m) -> None: ... (empty)

121 

122 async with self.patch_broker(broker, with_real=True) as br: (empty)

123 await br.publish("", queue) (empty)

124 await br.publish("", queue + "1") (empty)

125 await h1.wait_call(3) (empty)

126 await h2.wait_call(3) (empty)

127 

128 assert len(routes) == 2 (empty)

129 

130 async def test_js_subscriber_mock( (empty)

131 self, 

132 queue: str, 

133 stream: JStream, 

134 ) -> None: 

135 broker = self.get_broker() (empty)

136 

137 @broker.subscriber(queue, stream=stream) (empty)

138 async def m(msg) -> None: (empty)

139 pass (empty)

140 

141 async with self.patch_broker(broker) as br: (empty)

142 await br.publish("hello", queue, stream=stream.name) (empty)

143 m.mock.assert_called_once_with("hello") (empty)

144 

145 async def test_js_publisher_mock( (empty)

146 self, 

147 queue: str, 

148 stream: JStream, 

149 ) -> None: 

150 broker = self.get_broker() (empty)

151 

152 publisher = broker.publisher(queue + "resp") (empty)

153 

154 @publisher (empty)

155 @broker.subscriber(queue, stream=stream) (empty)

156 async def m(msg) -> str: (empty)

157 return "response" (empty)

158 

159 async with self.patch_broker(broker) as br: (empty)

160 await br.publish("hello", queue, stream=stream.name) (empty)

161 publisher.mock.assert_called_with("response") (empty)

162 

163 async def test_any_subject_routing(self) -> None: (empty)

164 broker = self.get_broker() (empty)

165 

166 @broker.subscriber("test.*.subj.*") (empty)

167 def subscriber(msg) -> None: ... (empty)

168 

169 async with self.patch_broker(broker) as br: (empty)

170 await br.publish("hello", "test.a.subj.b") (empty)

171 subscriber.mock.assert_called_once_with("hello") (empty)

172 

173 async def test_ending_subject_routing(self) -> None: (empty)

174 broker = self.get_broker() (empty)

175 

176 @broker.subscriber("test.>") (empty)

177 def subscriber(msg) -> None: ... (empty)

178 

179 async with self.patch_broker(broker) as br: (empty)

180 await br.publish("hello", "test.a.subj.b") (empty)

181 subscriber.mock.assert_called_once_with("hello") (empty)

182 

183 async def test_mixed_subject_routing(self) -> None: (empty)

184 broker = self.get_broker() (empty)

185 

186 @broker.subscriber("*.*.subj.>") (empty)

187 def subscriber(msg) -> None: ... (empty)

188 

189 async with self.patch_broker(broker) as br: (empty)

190 await br.publish("hello", "test.a.subj.b.c") (empty)

191 subscriber.mock.assert_called_once_with("hello") (empty)

192 

193 async def test_consume_pull( (empty)

194 self, 

195 queue: str, 

196 stream: JStream, 

197 ) -> None: 

198 broker = self.get_broker() (empty)

199 

200 @broker.subscriber(queue, stream=stream, pull_sub=PullSub(1)) (empty)

201 def subscriber(m) -> None: ... (empty)

202 

203 async with self.patch_broker(broker) as br: (empty)

204 await br.publish("hello", queue) (empty)

205 subscriber.mock.assert_called_once_with("hello") (empty)

206 

207 async def test_consume_batch( (empty)

208 self, 

209 queue: str, 

210 stream: JStream, 

211 ) -> None: 

212 broker = self.get_broker() (empty)

213 

214 @broker.subscriber( (empty)

215 queue, 

216 stream=stream, 

217 pull_sub=PullSub(1, batch=True), 

218 ) 

219 def subscriber(m) -> None: (empty)

220 pass (empty)

221 

222 async with self.patch_broker(broker) as br: (empty)

223 await br.publish("hello", queue) (empty)

224 subscriber.mock.assert_called_once_with(["hello"]) (empty)

225 

226 async def test_consume_with_subject_filter(self, queue: str) -> None: (empty)

227 broker = self.get_broker() (empty)

228 

229 @broker.subscriber( (empty)

230 config=ConsumerConfig(filter_subjects=[f"{queue}.a"]), 

231 stream=JStream(queue, subjects=[f"{queue}.*"]), 

232 ) 

233 def subscriber(m) -> None: (empty)

234 pass (empty)

235 

236 async with self.patch_broker(broker) as br: (empty)

237 await br.publish(1, f"{queue}.b") (empty)

238 await br.publish(2, f"{queue}.a") (empty)

239 subscriber.mock.assert_called_once_with(2) (empty)

240 

241 @pytest.mark.connected() (empty)

242 async def test_broker_gets_patched_attrs_within_cm(self) -> None: (empty)

243 await super().test_broker_gets_patched_attrs_within_cm(FakeProducer) (empty)

244 

245 @pytest.mark.connected() (empty)

246 async def test_broker_with_real_doesnt_get_patched(self) -> None: (empty)

247 await super().test_broker_with_real_doesnt_get_patched() (empty)

248 

249 @pytest.mark.connected() (empty)

250 async def test_broker_with_real_patches_publishers_and_subscribers( (empty)

251 self, 

252 queue: str, 

253 ) -> None: 

254 await super().test_broker_with_real_patches_publishers_and_subscribers(queue) (empty)

255 

256 @pytest.mark.xfail(reason="https://github.com/ag2ai/faststream/issues/2513") (empty)

257 async def test_publisher_without_destination(self) -> None: (empty)

258 """Fixes https://github.com/ag2ai/faststream/issues/2513.""" 

259 broker = self.get_broker() (empty)

260 

261 # use two publishers to check that we don't have conflicts 

262 publisher = broker.publisher(subject="") (empty)

263 another_publisher = broker.publisher(subject="") (empty)

264 

265 async with self.patch_broker(broker): (empty)

266 await publisher.publish(None, subject="new-key") 

267 publisher.mock.assert_called_once() 

268 

269 await another_publisher.publish(None, subject="new-key") 

270 another_publisher.mock.assert_called_once()