Coverage for tests/tests_rf/test_dispatcher.py: 0%

1#!/usr/bin/env python3 

2"""RAMSES RF - Unittests for dispatcher.""" 

3 

4from collections.abc import Generator 

5from datetime import datetime as dt, timedelta as td 

6from unittest.mock import AsyncMock, MagicMock 

7 

8import pytest 

9 

10from ramses_rf import Device, dispatcher 

11from ramses_rf.database import MessageIndex 

12from ramses_rf.gateway import Gateway 

13from ramses_tx import Address, DeviceIdT, Message, Packet 

14 

15 

16@pytest.fixture 

17def mock_gateway() -> Generator[MagicMock, None, None]: 

18 """Create a mock Gateway instance for testing.""" 

19 gateway = MagicMock(spec=Gateway) 

20 gateway.send_cmd = AsyncMock() 

21 gateway.dispatcher = MagicMock() 

22 gateway.dispatcher.send = MagicMock() 

23 

24 # Add required attributes 

25 gateway.config = MagicMock() 

26 gateway.config.disable_discovery = False 

27 gateway.config.enable_eavesdrop = False 

28 gateway._loop = MagicMock() 

29 gateway._loop.call_soon = MagicMock() 

30 gateway._loop.call_later = MagicMock() 

31 gateway._loop.time = MagicMock(return_value=0.0) 

32 gateway._include = {} 

33 # activate the SQLite MessageIndex 

34 gateway.msg_db = MessageIndex(maintain=False) 

35 

36 yield gateway 

37 

38 

39class Test_dispatcher_gateway: 

40 """Test Dispatcher class.""" 

41 

42 _SRC1 = "32:166025" 

43 _SRC2 = "01:087939" # (CTR) 

44 _NONA = "--:------" 

45 _NOW = dt.now().replace(microsecond=0) 

46 

47 msg5: Message = Message._from_pkt( 

48 Packet( 

49 _NOW + td(seconds=40), 

50 "... I --- 04:189078 --:------ 01:145038 3150 002 0100", # heat_demand 

51 ) 

52 ) 

53 

54 msg6: Message = Message._from_pkt( 

55 Packet( 

56 _NOW + td(seconds=50), 

57 "061 RP --- 10:078099 01:087939 --:------ 3220 005 00C0110000", # OTB 

58 ) 

59 ) 

60 

61 @pytest.mark.skip(reason="requires gwy") 

62 async def test_create_devices_from_addrs(self, mock_gateway: MagicMock) -> None: 

63 # device_by_id = { 

64 dev1 = Device(mock_gateway, Address(DeviceIdT("04:189078"))) 

65 # dev2 = Device(mock_gateway, Address(DeviceIdT("01:145038"))) 

66 # } 

67 mock_gateway.device_by_id = MagicMock(return_value=dev1) 

68 mock_gateway._check_dst_slug = MagicMock(return_value="CTL") 

69 dispatcher._create_devices_from_addrs(mock_gateway, self.msg5) 

70 

71 mock_gateway.msg_db.stop() # close sqlite3 connection 

72 

73 async def test_check_msg_addrs(self) -> None: 

74 dispatcher._check_msg_addrs(self.msg5) 

75 dispatcher._check_msg_addrs(self.msg6) 

76 

77 async def test_check_dst_slug(self) -> None: 

78 dispatcher._check_dst_slug(self.msg5) 

79 

80 async def test_detect_array_fragment(self) -> None: 

81 msg1: Message = Message._from_pkt( 

82 Packet( 

83 self._NOW, 

84 "... I --- 01:158182 --:------ 01:158182 000A 048 001001F40BB8011101F40BB8021101F40BB8031001F40BB8041101F40BB8051101F40BB8061101F40BB8071001F40BB8", 

85 ) 

86 ) 

87 msg2: Message = Message._from_pkt( 

88 Packet( 

89 self._NOW + td(seconds=1), # delta dtm < 3 secs 

90 "... I --- 01:158182 --:------ 01:158182 000A 006 081001F409C4", 

91 ) 

92 ) 

93 msg3: Message = Message._from_pkt( 

94 Packet( 

95 self._NOW + td(seconds=10), # delta dtm > 3 secs 

96 "... I --- 01:158182 --:------ 01:158182 000A 006 081001F409C4", 

97 ) 

98 ) 

99 assert msg1._has_array 

100 assert dispatcher.detect_array_fragment(msg2, msg1) 

101 assert not dispatcher.detect_array_fragment(msg3, msg1)