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

1#!/usr/bin/env python3 

2"""RAMSES RF - Check get of TCS fault logs.""" 

3 

4import asyncio 

5import re 

6 

7import pytest 

8 

9from ramses_rf import Gateway 

10from ramses_rf.device import Controller 

11from ramses_rf.system import Evohome 

12from ramses_tx.address import HGI_DEVICE_ID, Address 

13from ramses_tx.protocol import PortProtocol 

14from ramses_tx.schemas import DeviceIdT 

15from tests_rf.virtual_rf import VirtualRf 

16 

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

18 I_, 

19 RP, 

20 RQ, 

21 W_, 

22 Code, 

23) 

24 

25from .conftest import TEST_DIR, _GwyConfigDictT 

26 

27LOGS_DIR = f"{TEST_DIR}/logs" 

28 

29TST_ID_ = Address("18:123456").id # the id of the test HGI80-compatible device 

30 

31TST_FAULTLOG_LIMIT = 3 # max is 63 (0x3F) entries, 3 is enough for testing 

32 

33 

34# ### FIXTURES ######################################################################## 

35 

36pytestmark = pytest.mark.asyncio() 

37 

38 

39@pytest.fixture() 

40def gwy_config() -> _GwyConfigDictT: 

41 return { 

42 "config": { 

43 "disable_discovery": True, 

44 "disable_qos": False, # QoS is required for this test 

45 "enforce_known_list": False, 

46 }, 

47 "known_list": {HGI_DEVICE_ID: {}}, # req'd to thwart foreign HGI blacklisting 

48 } 

49 

50 

51@pytest.fixture() 

52def gwy_dev_id() -> DeviceIdT: 

53 return TST_ID_ 

54 

55 

56# ### TESTS ########################################################################### 

57 

58 

59async def _test_get_faultlog(gwy: Gateway, ctl_id: DeviceIdT) -> None: 

60 """Test obtaining the fault log.""" 

61 

62 assert gwy._loop is asyncio.get_running_loop() # scope BUG is here 

63 assert isinstance(gwy._protocol, PortProtocol) # mypy 

64 assert gwy._protocol._disable_qos is False # QoS is required for this test 

65 

66 _: Controller = gwy.get_device(ctl_id) # type: ignore[assignment] 

67 

68 tcs: Evohome | None = gwy.tcs 

69 assert isinstance(tcs, Evohome) # mypy 

70 

71 faultlog = await tcs.get_faultlog(limit=TST_FAULTLOG_LIMIT) 

72 assert faultlog 

73 

74 

75####################################################################################### 

76 

77 

78def _create_test_suite(log_file_name: str) -> dict[str, str]: 

79 def proc_log_line_pair(rq: str, rp: str) -> dict[str, str]: 

80 if RQ not in rq and RP not in rp: 

81 # RQ --- 18:006402 01:145038 --:------ 0418 003 000000 

82 # RP --- 01:145038 18:006402 --:------ 0418 022 004000B00400000000004A18... 

83 raise ValueError(f"Bad log file RQ/RP pair at line {rq}") 

84 

85 rq_ = re.sub(r" 18:...... ", f" {TST_ID_} ", rq[31:].strip()) 

86 rp_ = re.sub(r" 18:...... ", f" {TST_ID_} ", rp[31:].strip()) 

87 return {rq_: rp_} 

88 

89 result: dict[str, str] = {} 

90 

91 with open(log_file_name) as file: 

92 lines = [line for line in file if line.strip()] # Skip blank lines 

93 for i in range(0, len(lines), 2): 

94 result |= proc_log_line_pair(lines[i], lines[i + 1]) 

95 

96 return result 

97 

98 

99# TEST_SUITE = { 

100# r"RQ.* 18:.* 01:.* 0418 003 000000": "RP --- 01:145038 18:006402 --:------ 0418 022 004000B00400000000004A18659A7FFFFF7000000001", 

101# r"RQ.* 18:.* 01:.* 0418 003 000001": "RP --- 01:145038 18:006402 --:------ 0418 022 000001B00400000000004A184B58FFFFFF7000000001", 

102# r"RQ.* 18:.* 01:.* 0418 003 000002": "RP --- 01:145038 18:006402 --:------ 0418 022 004002B0060401000000431888F87FFFFF70005A23FD", 

103# r"RQ.* 18:.* 01:.* 0418 003 000003": "RP --- 01:145038 18:006402 --:------ 0418 022 000003B006040100000043187D63FFFFFF70005A23FD", 

104# r"RQ.* 18:.* 01:.* 0418 003 000004": "RP --- 01:145038 18:006402 --:------ 0418 022 000000B0000000000000000000007FFFFF7000000000", 

105# } 

106TEST_SUITE = _create_test_suite(f"{LOGS_DIR}/test_api_faultlog.log") 

107 

108 

109async def test_get_faultlog_fake(fake_evofw3: Gateway) -> None: 

110 """Test obtaining the schedule from a faked controller via Virtual RF.""" 

111 

112 assert fake_evofw3._transport # mypy 

113 

114 # Prime the virtual RF with the expected replies... 

115 rf: VirtualRf = fake_evofw3._transport.get_extra_info("virtual_rf") 

116 for k, v in TEST_SUITE.items(): 

117 rf.add_reply_for_cmd(k, v) 

118 

119 # we'll need null replies for all the other fault log slots (max 64?) 

120 for idx in range(len(TEST_SUITE), 64): 

121 rf.add_reply_for_cmd( 

122 list(TEST_SUITE.keys())[-1][:-2] + f"{idx:02X}", 

123 list(TEST_SUITE.values())[-1], 

124 ) 

125 

126 await _test_get_faultlog(fake_evofw3, "01:145038") 

127 

128 tcs = fake_evofw3.tcs 

129 assert tcs # mypy 

130 

131 _ = await tcs.get_faultlog(limit=64) # TODO: multiple TEST_SUITEs 

132 

133 assert len(tcs._faultlog._log) == 49 

134 assert ( 

135 str(tcs._faultlog.latest_event) 

136 == "24-04-20T12:44:52, restore, battery_low, 00:000001, 00, controller" 

137 ) 

138 assert ( 

139 str(tcs._faultlog.latest_fault) 

140 == "24-04-20T09:26:49, fault, battery_low, 00:000001, 00, controller" 

141 ) 

142 assert ( 

143 tcs._faultlog.active_faults is not None 

144 and len(tcs._faultlog.active_faults) == 1 

145 and ( 

146 str(tcs._faultlog.active_faults[0]) 

147 == "24-03-20T20:11:13, fault, comms_fault, 07:123456, FA, dhw_sensor" 

148 ) 

149 ) 

150 

151 # assert tcs.latest_event 

152 # assert tcs.latest_fault 

153 # assert tcs.active_fault 

154 

155 

156@pytest.mark.xdist_group(name="real_serial") 

157async def test_get_faultlog_mqtt(mqtt_evofw3: Gateway) -> None: 

158 """Test obtaining the fault log from a real controller via MQTT.""" 

159 

160 await _test_get_faultlog(mqtt_evofw3, "01:145038") 

161 

162 

163@pytest.mark.xdist_group(name="real_serial") 

164async def test_get_faultlog_real(real_evofw3: Gateway) -> None: 

165 """Test obtaining the fault log from a real controller via RF.""" 

166 

167 await _test_get_faultlog(real_evofw3, "01:145038")