Coverage for custom_components / supernotify / transports / persistent.py: 96%

1from __future__ import annotations 

2 

3import logging 

4from typing import TYPE_CHECKING, Any 

5 

6from custom_components.supernotify.const import ( 

7 ATTR_NOTIFICATION_ID, 

8 TRANSPORT_PERSISTENT, 

9) 

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

11from custom_components.supernotify.transport import Transport 

12 

13if TYPE_CHECKING: 

14 from custom_components.supernotify.envelope import Envelope 

15 

16_LOGGER = logging.getLogger(__name__) 

17 

18 

19class PersistentTransport(Transport): 

20 name = TRANSPORT_PERSISTENT 

21 

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

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

24 

25 @property 

26 def supported_features(self) -> TransportFeature: 

27 return TransportFeature.MESSAGE | TransportFeature.TITLE 

28 

29 @property 

30 def default_config(self) -> TransportConfig: 

31 config = TransportConfig() 

32 config.delivery_defaults.action = "persistent_notification.create" 

33 config.delivery_defaults.target_required = TargetRequired.NEVER 

34 return config 

35 

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

37 data = envelope.data or {} 

38 

39 notification_id = data.get(ATTR_NOTIFICATION_ID) or envelope.delivery.data.get(ATTR_NOTIFICATION_ID) 

40 action_data = envelope.core_action_data() 

41 if notification_id is not None: 

42 action_data["notification_id"] = notification_id 

43 

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