Coverage for tests / brokers / mqtt / test_testclient.py: 99%

1import pytest (empty)

2 

3from faststream.mqtt.broker.broker import MQTTBroker (empty)

4from faststream.mqtt.testing import FakeProducer, TestMQTTBroker, mqtt_topic_matches (empty)

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

6 

7from .basic import MQTTMemoryTestcaseConfig (empty)

8 

9_SKIP_V311 = "not supported in MQTT 3.1.1" (empty)

10 

11 

12@pytest.mark.mqtt() (empty)

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

14class TestTestclient(MQTTMemoryTestcaseConfig, BrokerTestclientTestcase): (empty)

15 def get_fake_producer_class(self) -> type: (empty)

16 return FakeProducer 

17 

18 async def test_consume_with_filter(self, queue, mock): (empty)

19 if self.version == "3.1.1": (empty)

20 pytest.skip(_SKIP_V311) (empty)

21 await super().test_consume_with_filter(queue, mock) (empty)

22 

23 async def test_response(self, queue, mock): (empty)

24 if self.version == "3.1.1": (empty)

25 pytest.skip(_SKIP_V311) (empty)

26 await super().test_response(queue, mock) (empty)

27 

28 async def test_reply_to(self, queue, mock): (empty)

29 if self.version == "3.1.1": (empty)

30 pytest.skip(_SKIP_V311) (empty)

31 await super().test_reply_to(queue, mock) (empty)

32 

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

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

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

36 

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

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

39 await super().test_broker_with_real_doesnt_get_patched() (empty)

40 

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

42 async def test_broker_with_real_patches_publishers_and_subscribers( (empty)

43 self, 

44 queue: str, 

45 ) -> None: 

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

47 

48 

49class TestTopicMatching: (empty)

50 """Unit tests for the MQTT wildcard matching helper.""" 

51 

52 def test_exact_match(self) -> None: (empty)

53 assert mqtt_topic_matches("sensors/temp", "sensors/temp") (empty)

54 

55 def test_exact_no_match(self) -> None: (empty)

56 assert not mqtt_topic_matches("sensors/temp", "sensors/humidity") (empty)

57 

58 def test_single_level_wildcard(self) -> None: (empty)

59 assert mqtt_topic_matches("sensors/+/temp", "sensors/room1/temp") (empty)

60 assert not mqtt_topic_matches("sensors/+/temp", "sensors/room1/floor2/temp") (empty)

61 

62 def test_multi_level_wildcard(self) -> None: (empty)

63 assert mqtt_topic_matches("sensors/#", "sensors/room1/temp") (empty)

64 assert mqtt_topic_matches("sensors/#", "sensors/room1/floor/temp") (empty)

65 assert mqtt_topic_matches("sensors/#", "sensors") (empty)

66 

67 def test_root_hash(self) -> None: (empty)

68 assert mqtt_topic_matches("#", "anything/at/all") (empty)

69 

70 def test_shared_subscription(self) -> None: (empty)

71 assert mqtt_topic_matches("$share/workers/sensors/#", "sensors/temp") (empty)

72 assert not mqtt_topic_matches("$share/workers/sensors/+", "sensors/a/b") (empty)

73 

74 

75@pytest.mark.mqtt() (empty)

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

77class TestWildcardRouting: (empty)

78 """Verify that TestMQTTBroker routes wildcard topics correctly.""" 

79 

80 async def test_plus_wildcard_routes(self, queue: str) -> None: (empty)

81 broker = MQTTBroker() (empty)

82 

83 @broker.subscriber(f"{queue}/+/temp") (empty)

84 async def handler(temperature: float) -> None: (empty)

85 pass (empty)

86 

87 async with TestMQTTBroker(broker) as br: (empty)

88 await br.start() (empty)

89 await br.publish("22.5", f"{queue}/room1/temp") (empty)

90 handler.mock.assert_called_once_with("22.5") (empty)

91 

92 async def test_hash_wildcard_routes(self, queue: str) -> None: (empty)

93 broker = MQTTBroker() (empty)

94 

95 @broker.subscriber(f"{queue}/#") (empty)

96 async def handler(msg: str) -> None: (empty)

97 pass (empty)

98 

99 async with TestMQTTBroker(broker) as br: (empty)

100 await br.start() (empty)

101 await br.publish("data", f"{queue}/a/b/c") (empty)

102 handler.mock.assert_called_once_with("data") (empty)

103 

104 async def test_no_match_skips_handler(self, queue: str) -> None: (empty)

105 broker = MQTTBroker() (empty)

106 

107 @broker.subscriber(f"{queue}/exact") (empty)

108 async def handler(msg: str) -> None: (empty)

109 pass 

110 

111 async with TestMQTTBroker(broker) as br: (empty)

112 await br.start() (empty)

113 await br.publish("data", f"{queue}/other") (empty)

114 handler.mock.assert_not_called() (empty)

115 

116 async def test_shared_subscription_routing(self, queue: str) -> None: (empty)

117 """Messages to a shared group topic should reach exactly one subscriber.""" 

118 broker = MQTTBroker() (empty)

119 

120 call_count = 0 (empty)

121 

122 @broker.subscriber(f"{queue}/data", shared="workers") (empty)

123 async def handler(msg: str) -> None: (empty)

124 nonlocal call_count 

125 call_count += 1 (empty)

126 

127 async with TestMQTTBroker(broker) as br: (empty)

128 await br.start() (empty)

129 await br.publish("hello", f"{queue}/data") (empty)

130 assert call_count == 1 (empty)