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

26 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2026-04-01 15:06 +0000

1import logging 

2from typing import TYPE_CHECKING, Any 

3 

4from custom_components.supernotify.const import ( 

5 ATTR_NOTIFICATION_ID, 

6 TRANSPORT_PERSISTENT, 

7) 

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

9from custom_components.supernotify.transport import Transport 

10 

11if TYPE_CHECKING: 

12 from custom_components.supernotify.envelope import Envelope 

13 

14_LOGGER = logging.getLogger(__name__) 

15 

16 

17class PersistentTransport(Transport): 

18 name = TRANSPORT_PERSISTENT 

19 

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

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

22 

23 @property 

24 def supported_features(self) -> TransportFeature: 

25 return TransportFeature.MESSAGE | TransportFeature.TITLE 

26 

27 @property 

28 def default_config(self) -> TransportConfig: 

29 config = TransportConfig() 

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

31 config.delivery_defaults.target_required = TargetRequired.NEVER 

32 return config 

33 

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

35 data = envelope.data or {} 

36 

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

38 action_data = envelope.core_action_data() 

39 if notification_id is not None: 

40 action_data["notification_id"] = notification_id 

41 

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