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

1#!/usr/bin/env python3 

2 

3# TODO: Remove unittest.mock.patch (use monkeypatch instead of unittest patch) 

4# TODO: Test with strict address checking 

5 

6"""RAMSES RF - Check GWY address/type detection and its treatment of addr0.""" 

7 

8import asyncio 

9from unittest.mock import patch 

10 

11import pytest 

12 

13from ramses_rf import Command, Gateway 

14from ramses_tx import exceptions as exc 

15from ramses_tx.address import HGI_DEVICE_ID, Address 

16from ramses_tx.protocol import PortProtocol 

17from ramses_tx.schemas import DeviceIdT 

18from ramses_tx.transport import MqttTransport 

19from ramses_tx.typing import QosParams 

20 

21from .conftest import _GwyConfigDictT 

22 

23# patched constants 

24_DBG_DISABLE_STRICT_CHECKING = True # # ramses_tx.address 

25 

26# other constants 

27ASSERT_CYCLE_TIME = 0.001 # max_cycles_per_assert = max_sleep / ASSERT_CYCLE_TIME 

28DEFAULT_MAX_SLEEP = 0.05 # 0.01/0.05 minimum for mocked (virtual RF)/actual 

29 

30 

31HGI_ID_ = HGI_DEVICE_ID # the sentinel value 

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

33 

34TEST_CMDS = { # test command strings (no impersonation) 

35 10: f"RQ --- {TST_ID_} 63:262142 --:------ 10E0 001 00", 

36 11: r"RQ --- 18:000730 63:262142 --:------ 10E0 001 00", 

37 20: f" I --- {TST_ID_} {TST_ID_} --:------ 30C9 003 000222", 

38 21: f" I --- 18:000730 {TST_ID_} --:------ 30C9 003 000333", 

39 30: f"RP --- {TST_ID_} 18:000730 --:------ 30C9 003 000444", 

40 31: r"RP --- 18:000730 18:000730 --:------ 30C9 003 000555", 

41 40: f" I --- {TST_ID_} --:------ {TST_ID_} 30C9 003 000666", 

42 41: f" I --- 18:000730 --:------ {TST_ID_} 30C9 003 000777", 

43 50: f" I --- {TST_ID_} --:------ 18:000730 30C9 003 000888", 

44 51: r" I --- 18:000730 --:------ 18:000730 30C9 003 000999", 

45 60: f" I --- --:------ --:------ {TST_ID_} 0008 002 00AA", 

46 61: r" I --- --:------ --:------ 18:000730 0008 002 00BB", 

47} 

48# NOTE: HGI80 will silently discard all frames that have addr0 != 18:000730 

49TEST_CMDS_FAIL_ON_HGI80 = [k for k, v in TEST_CMDS.items() if v[7:16] == TST_ID_] 

50 

51 

52# ### FIXTURES ######################################################################### 

53 

54pytestmark = pytest.mark.asyncio() # scope="module") 

55 

56 

57@pytest.fixture() 

58def gwy_config() -> _GwyConfigDictT: 

59 return { 

60 "config": { 

61 "disable_discovery": True, 

62 "disable_qos": False, # this is required for this test 

63 "enforce_known_list": False, 

64 }, 

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

66 } 

67 

68 

69@pytest.fixture() 

70def gwy_dev_id() -> DeviceIdT: 

71 return TST_ID_ 

72 

73 

74def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: 

75 metafunc.parametrize("test_idx", TEST_CMDS) 

76 

77 

78# ### TESTS ############################################################################ 

79 

80 

81@patch( # DISABLE_STRICT_CHECKING 

82 "ramses_tx.address._DBG_DISABLE_STRICT_CHECKING", 

83 _DBG_DISABLE_STRICT_CHECKING, 

84) 

85async def _test_gwy_device(gwy: Gateway, test_idx: int) -> None: 

86 """Check GWY address/type detection, and behaviour of its treatment of addr0.""" 

87 

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

89 

90 if not isinstance(gwy._protocol, PortProtocol) or not gwy._protocol._context: 

91 assert False, "QoS protocol not enabled" # use assert, not skip 

92 

93 assert gwy.hgi # mypy 

94 

95 # we replace the (non-sentinel) gwy_id with the real gwy's actual dev_id 

96 cmd_str = TEST_CMDS[test_idx].replace(TST_ID_, gwy.hgi.id) 

97 # this is irrevelent for fake (virtual) gwys, as they been assigned this id 

98 

99 cmd = Command(cmd_str) 

100 assert str(cmd) == cmd_str # sanity check 

101 

102 # a HGI80 (ti4310) will silently discard all frames that have addr0 != 18:000730 

103 is_hgi80 = not gwy._protocol._is_evofw3 # TODO: is_hgi80? 

104 

105 assert gwy._transport # mypy 

106 

107 # NOTE: timeout values are empirical, and may need to be adjusted 

108 if isinstance(gwy._transport, MqttTransport): # MQTT 

109 timeout = 0.375 * 2 # intesting, fail: 0.370 work: 0.375: 0.75 margin of safety 

110 elif gwy._transport.get_extra_info("virtual_rf"): # # fake 

111 timeout = 0.010 * 2 # was 0.003 * 2: increased to 20ms for CI stability 

112 else: # # real 

113 timeout = 0.355 * 2 # intesting, fail: 0.350 work: 0.355: 0.71 margin of safety 

114 

115 try: 

116 # using gwy._protocol.send_cmd() instead of gwy.async_send_cmd() as the 

117 # latter may swallow the exception we wish to capture (ProtocolSendFailed) 

118 pkt = await gwy._protocol.send_cmd( 

119 cmd, qos=QosParams(wait_for_reply=False, timeout=timeout) 

120 ) # for this test, we only need the cmd echo 

121 except exc.ProtocolSendFailed: 

122 if is_hgi80 and cmd_str[7:16] != HGI_DEVICE_ID: 

123 return # should have failed, and has 

124 raise # should not have failed, but has! 

125 

126 assert pkt is not None 

127 

128 if is_hgi80 and cmd_str[7:16] != HGI_DEVICE_ID: 

129 assert False, pkt # should have failed, but has not! 

130 

131 # NOTE: both HGI80/evofw3 will swap out addr0 (only) for its own device_id 

132 

133 if cmd_str[7:16] == HGI_DEVICE_ID: 

134 pkt_str = cmd_str[:7] + gwy.hgi.id + cmd_str[16:] 

135 else: 

136 pkt_str = cmd_str 

137 

138 assert pkt._frame == pkt_str 

139 

140 

141# ### TESTS ############################################################################ 

142 

143 

144@pytest.mark.xdist_group(name="virt_serial") 

145async def test_fake_evofw3(fake_evofw3: Gateway, test_idx: int) -> None: 

146 """Check the behaviour of the fake (virtual) evofw3 against the GWY test.""" 

147 

148 await _test_gwy_device(fake_evofw3, test_idx) 

149 

150 

151@pytest.mark.xdist_group(name="virt_serial") 

152async def test_fake_ti3410(fake_ti3410: Gateway, test_idx: int) -> None: 

153 """Check the behaviour of the fake (virtual) HGI80 against the GWY test.""" 

154 

155 await _test_gwy_device(fake_ti3410, test_idx) 

156 

157 

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

159async def test_mqtt_evofw3(mqtt_evofw3: Gateway, test_idx: int) -> None: 

160 """Validate the GWY test against an MQTT server.""" 

161 

162 await _test_gwy_device(mqtt_evofw3, test_idx) 

163 

164 

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

166async def test_real_evofw3(real_evofw3: Gateway, test_idx: int) -> None: 

167 """Validate the GWY test against a real (physical) evofw3.""" 

168 

169 await _test_gwy_device(real_evofw3, test_idx) 

170 

171 

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

173async def test_real_ti3410(real_ti3410: Gateway, test_idx: int) -> None: 

174 """Validate the GWY test against a real (physical) HGI80.""" 

175 

176 await _test_gwy_device(real_ti3410, test_idx)