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

1import json 

2import logging 

3from typing import Any 

4 

5from homeassistant.components.mqtt.const import ATTR_TOPIC 

6 

7from custom_components.supernotify import TRANSPORT_MQTT 

8from custom_components.supernotify.envelope import Envelope 

9from custom_components.supernotify.model import Target, TargetRequired, TransportConfig 

10from custom_components.supernotify.transport import ( 

11 Transport, 

12) 

13 

14RE_VALID_PHONE = r"^(\+\d{1,3})?\s?\(?\d{1,4}\)?[\s.-]?\d{3}[\s.-]?\d{4}$" 

15 

16_LOGGER = logging.getLogger(__name__) 

17 

18 

19class MQTTTransport(Transport): 

20 name = TRANSPORT_MQTT 

21 

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

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

24 

25 @property 

26 def default_config(self) -> TransportConfig: 

27 config = TransportConfig() 

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

29 config.delivery_defaults.target_required = TargetRequired.NEVER 

30 config.delivery_defaults.options = {} 

31 return config 

32 

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

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

35 return action is self.delivery_defaults.action 

36 

37 def recipient_target(self, recipient: dict[str, Any]) -> Target | None: # noqa: ARG002 

38 return None 

39 

40 async def deliver(self, envelope: Envelope) -> bool: 

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

42 

43 if not envelope.data or ATTR_TOPIC not in envelope.data: 

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

45 action_data: dict[str, Any] = envelope.data 

46 if isinstance(action_data["payload"], dict): 

47 action_data["payload"] = json.dumps(action_data["payload"]) 

48 return await self.call_action(envelope, action_data=action_data)