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 ( 

11 DebugTrace, 

12 SuppressionReason, 

13 TargetRequired, 

14 TransportConfig, 

15 TransportFeature, 

16) 

17from custom_components.supernotify.target import TargetEntityCategory 

18from custom_components.supernotify.transport import ( 

19 Transport, 

20) 

21 

22if TYPE_CHECKING: 

23 from custom_components.supernotify.envelope import Envelope 

24 from custom_components.supernotify.hass_api import HomeAssistantAPI 

25 

26HA_MQTT_DOMAIN = "mqtt" 

27# what mqtt.publish accepts besides `topic` - its schema rejects any other key, and a notification's 

28# own `data`, meant for other deliveries, reaches every delivery 

29PUBLISH_DATA_KEYS = ("payload", "qos", "retain", "evaluate_payload", "message_expiry_interval") 

30 

31_LOGGER = logging.getLogger(__name__) 

32 

33 

34class MQTTTransport(Transport): 

35 name = TRANSPORT_MQTT 

36 integration_domain = HA_MQTT_DOMAIN 

37 

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

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

40 

41 @property 

42 def supported_features(self) -> TransportFeature: 

43 return TransportFeature.MESSAGE 

44 

45 @property 

46 def default_config(self) -> TransportConfig: 

47 config = super().default_config 

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

49 # a topic can come from a `data:` keyword instead of a target (see deliver() 

50 # below), so this can't be inferred ALWAYS like most transports with a 

51 # unique_target_categories entry - Notification._target_required() covers the gap 

52 # this leaves for an implicit-only, category-less-at-a-glance selection 

53 config.delivery_defaults.target_required = TargetRequired.OPTIONAL 

54 config.delivery_defaults.options = {} 

55 return config 

56 

57 @property 

58 def unique_target_categories(self) -> list[str | TargetEntityCategory]: 

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

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

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

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

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

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

65 # shape a validator recognises. 

66 return [ATTR_TOPIC] 

67 

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

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

70 return action == self.delivery_defaults.action 

71 

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

73 return hass_api.find_config_entry_data(HA_MQTT_DOMAIN) is not None 

74 

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

76 return {self.name: {}} 

77 

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

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

80 

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

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

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

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

85 if topics: 

86 data.pop(ATTR_TOPIC, None) 

87 else: 

88 topic = data.pop(ATTR_TOPIC, None) 

89 if topic: 

90 topics = [topic] 

91 

92 if not topics: 

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

94 envelope.skipped = 1 

95 envelope.skip_reason = SuppressionReason.NO_TARGET 

96 return False 

97 

98 dropped: list[str] = sorted(k for k in data if k not in PUBLISH_DATA_KEYS) 

99 if dropped: 

100 _LOGGER.debug("SUPERNOTIFY notify_mqtt: dropping data keys mqtt.publish doesn't accept: %s", dropped) 

101 data = {k: v for k, v in data.items() if k in PUBLISH_DATA_KEYS} 

102 

103 payload: Any = data.get("payload") 

104 if isinstance(payload, (dict, list)): 

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

106 elif payload is None: 

107 data["payload"] = envelope.message 

108 

109 success = True 

110 for topic in topics: 

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

112 action_data[ATTR_TOPIC] = topic 

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

114 return success