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

1#!/usr/bin/env python3 

2"""Fixtures for testing.""" 

3 

4import logging 

5import os 

6from collections.abc import AsyncGenerator 

7from pathlib import Path 

8from typing import Final, NoReturn, TypeAlias, TypedDict 

9from unittest.mock import patch 

10 

11import pytest 

12import serial as ser # type: ignore[import-untyped] 

13from serial.tools.list_ports import comports # type: ignore[import-untyped] 

14 

15from ramses_rf import Gateway 

16from ramses_rf.device import HgiGateway 

17from ramses_tx import exceptions as exc 

18from ramses_tx.address import HGI_DEVICE_ID 

19from ramses_tx.schemas import DeviceIdT 

20from tests_rf.virtual_rf import HgiFwTypes, VirtualRf 

21 

22# 

23PortStrT: TypeAlias = str 

24 

25TEST_DIR = Path(__file__).resolve().parent # TEST_DIR = f"{os.path.dirname(__file__)}" 

26 

27SZ_GWY_CONFIG: Final = "gwy_config" 

28SZ_GWY_DEV_ID: Final = "gwy_dev_id" 

29 

30 

31class _ConfigDictT(TypedDict): 

32 disable_discovery: bool 

33 disable_qos: bool 

34 enforce_known_list: bool 

35 

36 

37class _GwyConfigDictT(TypedDict): 

38 config: _ConfigDictT 

39 known_list: dict[DeviceIdT, dict[str, bool]] 

40 

41 

42_LOGGER = logging.getLogger(__name__) 

43 

44 

45IN_GITHUB_ACTIONS = os.getenv("GITHUB_ACTIONS") == "true" 

46# set -x GITHUB_ACTIONS true 

47# set -u GITHUB_ACTIONS 

48 

49_global_failed_ports: list[str] = [] 

50 

51 

52####################################################################################### 

53 

54# pytestmark = pytest.mark.asyncio(scope="function") # needed? 

55 

56 

57@pytest.fixture(autouse=True) 

58def patches_for_tests(monkeypatch: pytest.MonkeyPatch) -> None: 

59 monkeypatch.setattr("ramses_tx.protocol._DBG_DISABLE_IMPERSONATION_ALERTS", True) 

60 monkeypatch.setattr("ramses_tx.transport._DBG_DISABLE_DUTY_CYCLE_LIMIT", True) 

61 monkeypatch.setattr("ramses_tx.transport.MIN_INTER_WRITE_GAP", 0) 

62 monkeypatch.setattr("ramses_tx.transport._DEFAULT_TIMEOUT_MQTT", 2) 

63 monkeypatch.setattr("ramses_tx.transport._DEFAULT_TIMEOUT_PORT", 0.5) 

64 

65 

66# TODO: add teardown to cleanup orphan MessageIndex thread 

67# @pytest.fixture(scope="session", autouse=True) 

68# def close_timer_threads(request: pytest.FixtureRequest) -> None: 

69# import threading 

70 

71# def finalize() -> None: 

72# running_timer_threads = [ 

73# thread 

74# for thread in threading.enumerate() 

75# if isinstance(thread, threading.Timer) 

76# ] 

77# for timer in running_timer_threads: 

78# timer.cancel() 

79 

80# request.addfinalizer(finalize) 

81 

82 

83####################################################################################### 

84 

85 

86@pytest.fixture() 

87async def rf() -> AsyncGenerator[VirtualRf, None]: 

88 """Utilize a virtual evofw3-compatible gateway.""" 

89 

90 rf = VirtualRf(2) 

91 

92 try: 

93 yield rf 

94 finally: 

95 await rf.stop() 

96 

97 

98@pytest.fixture() 

99def fake_evofw3_port(request: pytest.FixtureRequest, rf: VirtualRf) -> PortStrT | None: 

100 """Utilize a virtual evofw3-compatible gateway. 

101 

102 Requires test to supply the gwy_dev_id fixture. 

103 """ 

104 

105 gwy_dev_id: DeviceIdT = request.getfixturevalue(SZ_GWY_DEV_ID) 

106 

107 rf.set_gateway(rf.ports[0], gwy_dev_id, fw_type=HgiFwTypes.EVOFW3) 

108 

109 # with patch("ramses_tx.transport.comports", rf.comports): 

110 return rf.ports[0] 

111 

112 

113@pytest.fixture() 

114def fake_ti3410_port(request: pytest.FixtureRequest, rf: VirtualRf) -> PortStrT | None: 

115 """Utilize a virtual HGI80-compatible gateway. 

116 

117 Requires test to supply the gwy_dev_id fixture. 

118 """ 

119 

120 gwy_dev_id: DeviceIdT = request.getfixturevalue(SZ_GWY_DEV_ID) 

121 

122 rf.set_gateway(rf.ports[0], gwy_dev_id, fw_type=HgiFwTypes.HGI_80) 

123 

124 # with patch("ramses_tx.transport.comports", rf.comports): 

125 return rf.ports[0] 

126 

127 

128@pytest.fixture() # TODO: remove HACK, below 

129async def mqtt_evofw3_port() -> PortStrT: 

130 """Utilize an actual evofw3-compatible gateway.""" 

131 

132 # We could mock the MQTT client at: patch("ramses_tx.transport.MqttTransport" 

133 pytest.skip("This test fixture requires an MQTT broker") 

134 

135 return "mqtt://mqtt_username:mqtt_passw0rd@127.0.0.1" # type: ignore[unreachable] 

136 

137 

138@pytest.fixture() # TODO: remove HACK, below 

139async def real_evofw3_port() -> PortStrT | NoReturn: 

140 """Utilize an actual evofw3-compatible gateway.""" 

141 

142 if IN_GITHUB_ACTIONS: # replace with your condition 

143 pytest.skip("This test fixture requires physical hardware") 

144 

145 port_names: list[PortStrT] = [ 

146 p.device for p in comports() if p.product and "evofw3" in p.product 

147 ] 

148 

149 if port_names: 

150 return port_names[0] 

151 

152 if port_names := [ 

153 p.device for p in comports() if p.name[:6] == "ttyACM" 

154 ]: # HACK: evofw3-esp 

155 _LOGGER.warning(f"Assuming {port_names[0]} is evofw3-compatible") 

156 return port_names[0] 

157 

158 pytest.skip("No evofw3-based gateway device found") 

159 

160 

161@pytest.fixture() 

162async def real_ti3410_port() -> PortStrT | NoReturn: 

163 """Utilize an actual HGI80-compatible gateway.""" 

164 

165 if IN_GITHUB_ACTIONS: # replace with your condition 

166 pytest.skip("This test fixture requires physical hardware") 

167 

168 port_names: list[PortStrT] = [ 

169 p.device for p in comports() if p.product and "TUSB3410" in p.product 

170 ] 

171 

172 if port_names: 

173 return port_names[0] 

174 

175 pytest.skip("No ti3410-based gateway device found") 

176 

177 

178####################################################################################### 

179 

180 

181async def _gateway(gwy_port: PortStrT, gwy_config: _GwyConfigDictT) -> Gateway: 

182 """Instantiate a gateway.""" 

183 

184 gwy = Gateway(gwy_port, **gwy_config) 

185 

186 assert gwy.hgi is None and gwy.devices == [] 

187 

188 await gwy.start() 

189 return gwy 

190 

191 

192async def _fake_gateway( 

193 gwy_port: PortStrT, gwy_config: _GwyConfigDictT, rf: VirtualRf 

194) -> Gateway: 

195 """Wrapper to instantiate a virtual gateway.""" 

196 

197 with patch("ramses_tx.transport.comports", rf.comports): 

198 gwy = await _gateway(gwy_port, gwy_config) 

199 

200 assert gwy._transport # mypy 

201 gwy._transport._extra["virtual_rf"] = rf 

202 return gwy 

203 

204 

205async def _real_gateway(gwy_port: PortStrT, gwy_config: _GwyConfigDictT) -> Gateway: 

206 """Wrapper to instantiate a physical gateway.""" 

207 

208 global _global_failed_ports 

209 

210 if gwy_port in _global_failed_ports: 

211 pytest.skip(f"Port {gwy_port} previously failed") 

212 

213 try: 

214 return await _gateway(gwy_port, gwy_config) 

215 except (ser.SerialException, exc.TransportError) as err: 

216 _global_failed_ports.append(gwy_port) 

217 pytest.xfail(str(err)) # not skip, as we had determined port exists elsewhere 

218 

219 

220@pytest.fixture() 

221async def fake_evofw3( 

222 fake_evofw3_port: PortStrT, request: pytest.FixtureRequest, rf: VirtualRf 

223) -> AsyncGenerator[Gateway, None]: 

224 """Utilize a virtual evofw3-compatible gateway (discovered by fake_evofw3_port). 

225 

226 Requires test to supply gwy_config & gwy_dev_id (used by fake_evofw3_port) fixtures. 

227 """ 

228 

229 gwy_config: _GwyConfigDictT = request.getfixturevalue(SZ_GWY_CONFIG) 

230 gwy_dev_id: DeviceIdT = request.getfixturevalue(SZ_GWY_DEV_ID) 

231 

232 gwy = await _fake_gateway(fake_evofw3_port, gwy_config, rf) 

233 

234 assert isinstance(gwy.hgi, HgiGateway) and gwy.hgi.id == gwy_dev_id 

235 assert gwy._protocol._is_evofw3 is True 

236 

237 try: 

238 yield gwy 

239 finally: 

240 await gwy.stop() 

241 

242 

243@pytest.fixture() 

244async def fake_ti3410( 

245 fake_ti3410_port: PortStrT, request: pytest.FixtureRequest, rf: VirtualRf 

246) -> AsyncGenerator[Gateway, None]: 

247 """Utilize a virtual HGI80-compatible gateway (discovered by fake_ti3410_port). 

248 

249 Requires test to supply gwy_config & gwy_dev_id (used by fake_ti3410_port) fixtures. 

250 """ 

251 

252 gwy_config: _GwyConfigDictT = request.getfixturevalue(SZ_GWY_CONFIG) 

253 gwy_dev_id: DeviceIdT = request.getfixturevalue(SZ_GWY_DEV_ID) 

254 

255 gwy = await _fake_gateway(fake_ti3410_port, gwy_config, rf) 

256 

257 assert isinstance(gwy.hgi, HgiGateway) and gwy.hgi.id == gwy_dev_id 

258 assert gwy._protocol._is_evofw3 is False 

259 

260 try: 

261 yield gwy 

262 finally: 

263 await gwy.stop() 

264 

265 

266@pytest.fixture() 

267async def mqtt_evofw3( 

268 mqtt_evofw3_port: PortStrT, request: pytest.FixtureRequest 

269) -> AsyncGenerator[Gateway, None]: 

270 """Utilize an actual evofw3-compatible gateway (discovered by mqtt_evofw3_port). 

271 

272 Requires test to supply corresponding gwy_config fixture. 

273 """ 

274 

275 # TODO: pytest.skip() if the MQTT broker is available 

276 gwy_config: _GwyConfigDictT = request.getfixturevalue(SZ_GWY_CONFIG) 

277 gwy = await _gateway(mqtt_evofw3_port, gwy_config) 

278 

279 gwy.get_device(gwy._protocol.hgi_id) # HACK: not instantiated: no puzzle pkts sent 

280 

281 assert isinstance(gwy.hgi, HgiGateway) and gwy.hgi.id not in (None, HGI_DEVICE_ID) 

282 assert gwy._protocol._is_evofw3 is True 

283 

284 try: 

285 yield gwy 

286 finally: 

287 await gwy.stop() 

288 

289 

290@pytest.fixture() 

291async def real_evofw3( 

292 real_evofw3_port: PortStrT, request: pytest.FixtureRequest 

293) -> AsyncGenerator[Gateway, None]: 

294 """Utilize an actual evofw3-compatible gateway (discovered by real_evofw3_port). 

295 

296 Requires test to supply corresponding gwy_config fixture. 

297 """ 

298 

299 gwy_config: _GwyConfigDictT = request.getfixturevalue(SZ_GWY_CONFIG) 

300 

301 gwy = await _real_gateway(real_evofw3_port, gwy_config) 

302 

303 assert isinstance(gwy.hgi, HgiGateway) and gwy.hgi.id not in (None, HGI_DEVICE_ID) 

304 assert gwy._protocol._is_evofw3 is True 

305 

306 try: 

307 yield gwy 

308 finally: 

309 await gwy.stop() 

310 

311 

312@pytest.fixture() 

313async def real_ti3410( 

314 real_ti3410_port: PortStrT, request: pytest.FixtureRequest 

315) -> AsyncGenerator[Gateway, None]: 

316 """Utilize an actual HGI80-compatible gateway (discovered by real_ti3410_port). 

317 

318 Requires test to supply corresponding gwy_config fixture. 

319 """ 

320 

321 gwy_config: _GwyConfigDictT = request.getfixturevalue(SZ_GWY_CONFIG) 

322 

323 gwy = await _real_gateway(real_ti3410_port, gwy_config) 

324 

325 assert isinstance(gwy.hgi, HgiGateway) and gwy.hgi.id not in (None, HGI_DEVICE_ID) 

326 assert gwy._protocol._is_evofw3 is False 

327 

328 try: 

329 yield gwy 

330 finally: 

331 await gwy.stop()