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

1#!/usr/bin/env python3 

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

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.database import MessageIndex 

11from ramses_rf.entity_base import _MessageDB 

12from ramses_rf.gateway import Gateway 

13from ramses_tx import Code, 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_entity_base: 

40 """Test _MessageDB 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 --- 01:145038 04:189078 --:------ 3220 005 00C0110000", # OTB 

58 ) 

59 ) 

60 

61 msg7: Message = Message._from_pkt( 

62 Packet( 

63 _NOW + td(seconds=60), 

64 "... I --- 04:189078 --:------ 01:145038 12B0 003 010000", # window_open 

65 ) 

66 ) 

67 

68 async def test_entity_base_dev(self, mock_gateway: MagicMock) -> None: 

69 # issues fetching results 

70 dev = _MessageDB(mock_gateway) 

71 dev.id = DeviceIdT("04:189078") 

72 dev._z_id = dev.id 

73 

74 # put messages in the msg_db 

75 dev._handle_msg(self.msg5) 

76 dev._handle_msg(self.msg6) 

77 dev._handle_msg(self.msg7) 

78 assert dev._gwy.msg_db 

79 assert len(dev._gwy.msg_db.all()) == 3, "len(msg_db.all) wrong" 

80 

81 # start tests 

82 assert dev.id == "04:189078" 

83 

84 sql = """ 

85 SELECT dtm from messages WHERE 

86 verb in (' I', 'RP') 

87 AND (src = ? OR dst = ?) 

88 AND ctx LIKE ? 

89 """ 

90 assert dev._gwy.msg_db.qry( 

91 sql, (dev.id[:9], dev.id[:9], f"%{dev.id[10:]}%") 

92 ) == ( 

93 self.msg5, 

94 self.msg7, 

95 self.msg6, 

96 ), "base qry wrong" 

97 

98 # create _msgs 

99 assert dev._msgs == {"12B0": self.msg7, "3150": self.msg5, "3220": self.msg6}, ( 

100 "base _msgs wrong" 

101 ) 

102 

103 # find our Codes 

104 assert dev._msg_dev_qry() == [ 

105 Code._3150, 

106 Code._12B0, 

107 Code._3220, 

108 ], "base _msg_dev_qry wrong" 

109 

110 # list our messages 

111 assert dev._msg_list == [self.msg5, self.msg7, self.msg6], "_msg_list wrong" 

112 

113 # create _msgz 

114 assert dev._msgz == { 

115 "12B0": {" I": {"01": self.msg7}}, 

116 "3150": {" I": {"01": self.msg5}}, 

117 "3220": {"RP": {"11": self.msg6}}, 

118 }, "base _msgz wrong" 

119 

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

121 

122 async def test_entity_base_zone(self, mock_gateway: MagicMock) -> None: 

123 # works as expected 

124 dev = _MessageDB(mock_gateway) 

125 dev.id = DeviceIdT("04:189078_01") 

126 dev._z_id = dev.id 

127 

128 # put messages in the msg_db 

129 dev._handle_msg(self.msg5) 

130 dev._handle_msg(self.msg6) 

131 dev._handle_msg(self.msg7) 

132 

133 # start tests 

134 assert dev.id == "04:189078_01" 

135 assert dev._gwy.msg_db 

136 

137 sql = """ 

138 SELECT dtm from messages WHERE 

139 verb in (' I', 'RP') 

140 AND (src = ? OR dst = ?) 

141 AND ctx LIKE ? 

142 """ 

143 assert dev._gwy.msg_db.qry( 

144 sql, (dev.id[:9], dev.id[:9], f"%{dev.id[10:]}%") 

145 ) == ( 

146 self.msg5, 

147 self.msg7, 

148 ), "zone qry wrong" 

149 

150 # create _msgs 

151 assert dev._msgs == {"12B0": self.msg7, "3150": self.msg5}, "zone _msgs wrong" 

152 

153 # find our Codes 

154 assert dev._msg_dev_qry() == [ 

155 Code._3150, 

156 Code._12B0, 

157 ], "zone _msg_dev_qry wrong" 

158 

159 # list our messages 

160 assert dev._msg_list == [self.msg5, self.msg7], "_msg_list wrong" 

161 

162 # create _msgz 

163 assert dev._msgz == { 

164 "12B0": {" I": {"01": self.msg7}}, 

165 "3150": {" I": {"01": self.msg5}}, 

166 }, "zone _msgz wrong" 

167 

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

169 

170 msg8: Message = Message._from_pkt( 

171 Packet( 

172 _NOW + td(seconds=70), 

173 "045 I --- 01:145038 --:------ 01:145038 3150 002 FC90", # heat_demand 

174 ) 

175 ) 

176 msg9: Message = Message._from_pkt( 

177 Packet( 

178 _NOW + td(seconds=80), 

179 "045 RP --- 01:145038 18:006402 --:------ 1260 003 00182B", # setpoint 

180 ) 

181 ) 

182 

183 async def test_entity_base_dhw(self, mock_gateway: MagicMock) -> None: 

184 # works as expected 

185 dev = _MessageDB(mock_gateway) 

186 dev.id = DeviceIdT("01:145038_HW") 

187 dev._z_id = dev.id 

188 

189 # put messages in the msg_db 

190 dev._handle_msg(self.msg8) 

191 dev._handle_msg(self.msg9) 

192 

193 # start tests 

194 assert dev.id == "01:145038_HW" 

195 assert dev._gwy.msg_db 

196 assert dev._gwy.msg_db.all() == (self.msg8, self.msg9), "wrong dhw all" 

197 

198 sql = """ 

199 SELECT dtm from messages WHERE 

200 verb in (' I', 'RP') 

201 AND (src = ? OR dst = ?) 

202 AND (ctx IN ('FC', 'FA', 'F9', 'FA') OR plk LIKE ?) 

203 """ 

204 _ctx_qry = "%dhw_idx%" 

205 # SELECT just fields 

206 # assert dev._gwy.msg_db.qry_field( 

207 # sql, (dev.id[:9], dev.id[:9], _ctx_qry) 

208 # ) == [('FC',), ('00',)] 

209 

210 # fetch Messages 

211 assert dev._gwy.msg_db.qry(sql, (dev.id[:9], dev.id[:9], _ctx_qry)) == ( 

212 self.msg8, 

213 self.msg9, 

214 ), "dhw qry wrong" 

215 

216 # create _msgs 

217 assert dev._msgs == {"1260": self.msg9, "3150": self.msg8}, "dhw _msgs wrong" 

218 

219 # find our Codes 

220 assert dev._msg_dev_qry() == [ 

221 Code._3150, 

222 Code._1260, 

223 ], "dhw _msg_dev_qry wrong" 

224 

225 # list our messages 

226 assert dev._msg_list == [self.msg8, self.msg9], "dhw _msg_list wrong" 

227 

228 # create _msgz 

229 assert dev._msgz == { 

230 "1260": {"RP": {"00": self.msg9}}, 

231 "3150": {" I": {"FC": self.msg8}}, 

232 }, "dhw _msgz wrong" 

233 

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