Coverage for tests/tests/test_api_faultlog.py: 0%

1#!/usr/bin/env python3 

2"""RAMSES RF - Test the Faultlog functions.""" 

3 

4import random 

5from datetime import datetime as dt 

6from typing import Any 

7 

8from ramses_rf import Address, Command, Message, Packet 

9from ramses_rf.system.faultlog import FaultLog, FaultLogEntry 

10from ramses_tx.address import HGI_DEVICE_ID 

11from ramses_tx.const import SZ_LOG_ENTRY, FaultDeviceClass, FaultState, FaultType 

12from ramses_tx.schemas import DeviceIdT 

13from ramses_tx.typed_dicts import LogIdxT 

14 

15from ramses_tx.const import ( # noqa: F401, isort: skip, pylint: disable=unused-import 

16 I_, 

17 RP, 

18 RQ, 

19 W_, 

20 Code, 

21) 

22 

23from .helpers import TEST_DIR 

24 

25WORK_DIR = f"{TEST_DIR}/parsers" 

26 

27 

28# ### TEST DATA ####################################################################### 

29 

30CTL_ID = Address("01:145038").id 

31HGI_ID = HGI_DEVICE_ID 

32 

33 

34def _fault_log_entry( 

35 *args: Any, timestamp: str | None = None, **kwargs: Any 

36) -> FaultLogEntry: 

37 if timestamp is None: 

38 timestamp = dt.strftime(dt.now(), "%y-%m-%dT%H:%M:%S") 

39 

40 return FaultLogEntry( 

41 timestamp=timestamp, 

42 fault_state=args[0], 

43 fault_type=args[1], 

44 domain_idx=kwargs.get("domain_idx", "00"), 

45 device_class=args[2], 

46 device_id=kwargs.get("device_id"), 

47 ) 

48 

49 

50# Keys corresponding to order in the faultlog 

51TEST_FAULTS: dict[LogIdxT, FaultLogEntry] = {} 

52 

53TEST_FAULTS["00"] = _fault_log_entry( 

54 FaultState.RESTORE, 

55 FaultType.BATTERY_ERROR, 

56 FaultDeviceClass.CONTROLLER, 

57 timestamp="21-12-23T00:59:00", 

58) 

59TEST_FAULTS["01"] = _fault_log_entry( 

60 FaultState.FAULT, 

61 FaultType.BATTERY_ERROR, 

62 FaultDeviceClass.CONTROLLER, 

63 timestamp="21-12-23T00:58:01", 

64) 

65TEST_FAULTS["02"] = _fault_log_entry( 

66 FaultState.RESTORE, 

67 FaultType.BATTERY_LOW, 

68 FaultDeviceClass.ACTUATOR, 

69 timestamp="21-12-23T00:57:02", 

70 device_id="04:111111", 

71 domain_idx="03", 

72) 

73TEST_FAULTS["03"] = _fault_log_entry( 

74 FaultState.FAULT, 

75 FaultType.BATTERY_LOW, 

76 FaultDeviceClass.ACTUATOR, 

77 timestamp="21-12-23T00:56:03", 

78 device_id="04:111111", 

79 domain_idx="03", 

80) 

81TEST_FAULTS["04"] = _fault_log_entry( 

82 FaultState.RESTORE, 

83 FaultType.COMMS_FAULT, 

84 FaultDeviceClass.SETPOINT, 

85 timestamp="21-12-23T00:55:04", 

86 device_id="03:123456", 

87 domain_idx="06", 

88) 

89TEST_FAULTS["05"] = _fault_log_entry( 

90 FaultState.FAULT, 

91 FaultType.COMMS_FAULT, 

92 FaultDeviceClass.SETPOINT, 

93 timestamp="21-12-23T00:54:05", 

94 device_id="03:123456", 

95 domain_idx="06", 

96) 

97 

98EXPECTED_MAP = { 

99 0: "21-12-23T00:59:00", 

100 1: "21-12-23T00:58:01", 

101 2: "21-12-23T00:57:02", 

102 3: "21-12-23T00:56:03", 

103 4: "21-12-23T00:55:04", 

104 5: "21-12-23T00:54:05", 

105} 

106 

107 

108# ### FIXTURES, ETC ################################################################### 

109 

110 

111class EvohomeStub: 

112 def __init__(self, ctl_id: DeviceIdT) -> None: 

113 self.id = ctl_id 

114 self._gwy = None 

115 

116 

117def _proc_log_line(log_line: str) -> None: 

118 try: 

119 pkt = Packet.from_file(log_line[:26], log_line[27:]) 

120 except ValueError: 

121 return 

122 

123 msg = Message(pkt) 

124 if msg.code != Code._0418 or not msg.payload.get(SZ_LOG_ENTRY): 

125 return 

126 

127 entry = FaultLogEntry.from_msg(msg) 

128 assert entry 

129 

130 

131def _proc_null_fault_entry(fault_log: FaultLog, _log_idx: LogIdxT = "00") -> None: 

132 """Return a 0418 packet with no entry.""" 

133 cmd = Command.from_attrs( 

134 I_, CTL_ID, Code._0418, f"0000{_log_idx}B0000000000000000000007FFFFF7000000000" 

135 ) 

136 fault_log.handle_msg(Message(Packet._from_cmd(cmd))) 

137 

138 

139def _proc_test_fault_entry( 

140 fault_log: FaultLog, text_idx: LogIdxT, _log_idx: LogIdxT = "00" 

141) -> None: 

142 entry: FaultLogEntry = TEST_FAULTS[text_idx] 

143 

144 cmd = Command._put_system_log_entry( 

145 CTL_ID, 

146 entry.fault_state, 

147 entry.fault_type, 

148 entry.device_class, 

149 device_id=entry.device_id, 

150 domain_idx=entry.domain_idx, 

151 _log_idx=_log_idx, 

152 timestamp=entry.timestamp, 

153 ) 

154 fault_log.handle_msg(Message(Packet._from_cmd(cmd))) 

155 

156 

157# ### TESTS ########################################################################### 

158 

159 

160def test_faultlog_entries() -> None: 

161 """Test instantiation of faultlog entries.""" 

162 

163 with open(f"{WORK_DIR}/code_0418.log") as f: 

164 while line := (f.readline()): 

165 _proc_log_line(line) 

166 

167 

168def test_faultlog_instantiation_0() -> None: 

169 """Log entries arrive in order of timestamp (i.e. as they'd occur).""" 

170 

171 fault_log = FaultLog(EvohomeStub(CTL_ID)) # type: ignore[type-var] 

172 

173 # log entries arrive in order of timestamp (i.e. as they'd occur) 

174 for i in reversed(range(len(TEST_FAULTS))): 

175 _proc_test_fault_entry(fault_log, f"{i:02}") # _log_idx="00") 

176 

177 # assert sorted(fault_log._log.keys(), reverse=True) == list(EXPECTED_MAP.values()) 

178 assert fault_log._map == EXPECTED_MAP 

179 

180 

181def test_faultlog_instantiation_1() -> None: 

182 """Log entries arrive in order of log_idx (e.g. enumerating the log via RQs).""" 

183 

184 fault_log = FaultLog(EvohomeStub(CTL_ID)) # type: ignore[type-var] 

185 

186 # log entries arrive in order of log_idx (e.g. enumerating the log via RQs) 

187 for i in reversed(range(len(TEST_FAULTS))): 

188 _proc_test_fault_entry(fault_log, f"{i:02}", _log_idx=f"{i:02}") 

189 

190 assert sorted(fault_log._log.keys(), reverse=True) == list(EXPECTED_MAP.values()) 

191 assert fault_log._map == EXPECTED_MAP 

192 

193 

194def test_faultlog_instantiation_2() -> None: 

195 """Log entries arrive in random order albeit with their correct log_idx.""" 

196 

197 fault_log = FaultLog(EvohomeStub(CTL_ID)) # type: ignore[type-var] 

198 

199 # log entries arrive in random order albeit with their correct log_idx 

200 numbers = list(range(len(TEST_FAULTS))) 

201 random.shuffle(numbers) 

202 

203 for i in numbers: 

204 _proc_test_fault_entry(fault_log, f"{i:02}", _log_idx=f"{i:02}") 

205 

206 assert sorted(fault_log._log.keys(), reverse=True) == list(EXPECTED_MAP.values()) 

207 assert fault_log._map == EXPECTED_MAP 

208 

209 

210def test_faultlog_instantiation_3() -> None: 

211 """Log entries arrive in an order set to confuse.""" 

212 

213 fault_log = FaultLog(EvohomeStub(CTL_ID)) # type: ignore[type-var] 

214 

215 # a log with two entries arrives in order 

216 _proc_test_fault_entry(fault_log, "05") 

217 _proc_test_fault_entry(fault_log, "04") 

218 

219 assert fault_log._map == { 

220 0: "21-12-23T00:55:04", 

221 1: "21-12-23T00:54:05", 

222 } 

223 

224 # the two entries arrives out of order 

225 _proc_test_fault_entry(fault_log, "05", _log_idx="01") 

226 _proc_test_fault_entry(fault_log, "04", _log_idx="00") 

227 

228 assert fault_log._map == { 

229 0: "21-12-23T00:55:04", 

230 1: "21-12-23T00:54:05", 

231 } 

232 

233 # the log is cleared 

234 _proc_null_fault_entry(fault_log) 

235 

236 assert fault_log._map == {} 

237 

238 # a log with three entries is enumerated, kinda 

239 _proc_test_fault_entry(fault_log, "03", _log_idx="00") 

240 # roc_fault_entry(fault_log, "04", _log_idx="01") # went missing 

241 _proc_test_fault_entry(fault_log, "05", _log_idx="02") 

242 

243 assert fault_log._map == { 

244 0: "21-12-23T00:56:03", 

245 2: "21-12-23T00:54:05", 

246 } 

247 

248 # the missing entry arrives, only after a new entry 

249 _proc_test_fault_entry(fault_log, "02") # pushes others down 

250 _proc_test_fault_entry(fault_log, "01") # pushes others down 

251 _proc_test_fault_entry(fault_log, "04", _log_idx="03") # _log_idx was 01, above 

252 

253 assert fault_log._map == { 

254 0: "21-12-23T00:58:01", 

255 1: "21-12-23T00:57:02", 

256 2: "21-12-23T00:56:03", 

257 3: "21-12-23T00:55:04", 

258 4: "21-12-23T00:54:05", 

259 } 

260 

261 # a new entry 

262 _proc_test_fault_entry(fault_log, "00") # pushes others down 

263 

264 assert fault_log._map == { 

265 0: "21-12-23T00:59:00", 

266 1: "21-12-23T00:58:01", 

267 2: "21-12-23T00:57:02", 

268 3: "21-12-23T00:56:03", 

269 4: "21-12-23T00:55:04", 

270 5: "21-12-23T00:54:05", 

271 } 

272 

273 

274def test_faultlog_instantiation_4() -> None: 

275 """Log entries arrive in an order set to confuse.""" 

276 

277 fault_log = FaultLog(EvohomeStub(CTL_ID)) # type: ignore[type-var] 

278 

279 # a log with three entries is enumerated, kinda 

280 _proc_test_fault_entry(fault_log, "03", _log_idx="00") 

281 # roc_fault_entry(fault_log, "04", _log_idx="01") # went missing 

282 _proc_test_fault_entry(fault_log, "05", _log_idx="02") 

283 

284 assert fault_log._map == { 

285 0: "21-12-23T00:56:03", 

286 2: "21-12-23T00:54:05", 

287 } 

288 

289 _proc_test_fault_entry(fault_log, "02", _log_idx="02") # pushes others down 

290 

291 assert fault_log._map == { 

292 2: "21-12-23T00:57:02", 

293 3: "21-12-23T00:56:03", 

294 5: "21-12-23T00:54:05", 

295 } 

296 

297 _proc_test_fault_entry(fault_log, "01") 

298 

299 assert fault_log._map == { 

300 0: "21-12-23T00:58:01", 

301 2: "21-12-23T00:57:02", 

302 3: "21-12-23T00:56:03", 

303 5: "21-12-23T00:54:05", 

304 } 

305 

306 _proc_test_fault_entry(fault_log, "01", _log_idx="01") 

307 

308 assert fault_log._map == { 

309 1: "21-12-23T00:58:01", 

310 2: "21-12-23T00:57:02", 

311 3: "21-12-23T00:56:03", 

312 5: "21-12-23T00:54:05", 

313 } 

314 

315 _proc_test_fault_entry(fault_log, "04", _log_idx="04") # _log_idx was 01, above 

316 

317 assert fault_log._map == { 

318 1: "21-12-23T00:58:01", 

319 2: "21-12-23T00:57:02", 

320 3: "21-12-23T00:56:03", 

321 4: "21-12-23T00:55:04", 

322 5: "21-12-23T00:54:05", 

323 }