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.components.mqtt.const import ATTR_TOPIC 

8 

9from custom_components.supernotify.const import TRANSPORT_MQTT 

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

11from custom_components.supernotify.transport import ( 

12 Transport, 

13) 

14 

15if TYPE_CHECKING: 

16 from custom_components.supernotify.envelope import Envelope 

17 

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

19 

20_LOGGER = logging.getLogger(__name__) 

21 

22 

23class MQTTTransport(Transport): 

24 name = TRANSPORT_MQTT 

25 

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

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

28 

29 @property 

30 def supported_features(self) -> TransportFeature: 

31 return TransportFeature.MESSAGE | TransportFeature.TITLE 

32 

33 @property 

34 def default_config(self) -> TransportConfig: 

35 config = TransportConfig() 

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

37 config.delivery_defaults.target_required = TargetRequired.NEVER 

38 config.delivery_defaults.options = {} 

39 return config 

40 

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

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

43 return action == self.delivery_defaults.action 

44 

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

46 return None 

47 

48 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool: # noqa: ARG002 

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

50 

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

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

53 return False 

54 

55 action_data: dict[str, Any] = dict(envelope.data) 

56 if isinstance(action_data.get("payload"), dict): 

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

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