Coverage for custom_components/supernotify/transports/mqtt.py: 100%

1from __future__ import annotations 

2 

3import json 

4import logging 

5from typing import TYPE_CHECKING, Any 

6 

7from homeassistant.helpers.typing import ConfigType 

8 

9from custom_components.supernotify.const import ATTR_TOPIC, TRANSPORT_MQTT 

10from custom_components.supernotify.model import DebugTrace, TargetRequired, TransportConfig, TransportFeature 

11from custom_components.supernotify.target import Target, TargetEntityCategory 

12from custom_components.supernotify.transport import ( 

13 Transport, 

14) 

15 

16if TYPE_CHECKING: 

17 from custom_components.supernotify.envelope import Envelope 

18 from custom_components.supernotify.hass_api import HomeAssistantAPI 

19 

20HA_MQTT_DOMAIN = "mqtt" 

21 

22_LOGGER = logging.getLogger(__name__) 

23 

24 

25class MQTTTransport(Transport): 

26 name = TRANSPORT_MQTT 

27 

28 def __init__(self, *args: Any, **kwargs: Any) -> None: 

29 super().__init__(*args, **kwargs) 

30 

31 @property 

32 def supported_features(self) -> TransportFeature: 

33 return TransportFeature.MESSAGE 

34 

35 @property 

36 def default_config(self) -> TransportConfig: 

37 config = TransportConfig() 

38 config.delivery_defaults.action = "mqtt.publish" 

39 config.delivery_defaults.target_required = TargetRequired.OPTIONAL 

40 config.delivery_defaults.options = {} 

41 config.delivery_defaults.inclusion = self.inclusion_mode 

42 return config 

43 

44 @property 

45 def target_categories(self) -> list[str | TargetEntityCategory]: 

46 # `topic` is a clean, dedicated category name for the mapping form (`target: {topic: 

47 # ...}`), distinct from overloading the transport's own name (`target: {mqtt: ...}`, 

48 # still handled separately by Delivery.select_targets()). A bare, unqualified 

49 # `target: <topic>` set directly on the mqtt delivery block also reaches here - not 

50 # via this list, but because it's the sole plain-string entry `Delivery. 

51 # reclassify_unqualified_target()` falls back to for a delivery-scoped value with no 

52 # shape a validator recognises. 

53 return [ATTR_TOPIC] 

54 

55 def validate_action(self, action: str | None) -> bool: 

56 """Override in subclass if transport has fixed action or doesn't require one""" 

57 return action == self.delivery_defaults.action 

58 

59 def is_viable(self, hass_api: HomeAssistantAPI) -> bool: 

60 return hass_api.find_config_entry_data(HA_MQTT_DOMAIN) is not None 

61 

62 def build_standard_deliveries(self, hass_api: HomeAssistantAPI) -> dict[str, ConfigType]: 

63 return {self.name: {}} 

64 

65 def recipient_target(self, recipient: dict[str, Any]) -> Target | None: 

66 return None 

67 

68 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool: 

69 _LOGGER.debug("SUPERNOTIFY notify_mqtt: %s", envelope.delivery_name) 

70 

71 data: dict[str, Any] = dict(envelope.data) if envelope.data else {} 

72 # envelope.target is already scoped to this transport by delivery.select_targets() 

73 # (see target_categories above), so any resolved target here is a topic 

74 topics: list[str] = envelope.target.resolved_targets() if envelope.target else [] 

75 if topics: 

76 data.pop(ATTR_TOPIC, None) 

77 else: 

78 topic = data.pop(ATTR_TOPIC, None) 

79 if topic: 

80 topics = [topic] 

81 

82 if not topics: 

83 _LOGGER.warning("SUPERNOTIFY notify_mqtt: No topic for publication") 

84 return False 

85 

86 if isinstance(data.get("payload"), dict): 

87 data["payload"] = json.dumps(data["payload"]) 

88 else: 

89 data["payload"] = envelope.message 

90 

91 success = True 

92 for topic in topics: 

93 action_data: dict[str, Any] = dict(data) 

94 action_data[ATTR_TOPIC] = topic 

95 success = await self.call_action(envelope, action_data=action_data) and success 

96 return success