Coverage for tests/tests_rf/test_create_stack.py: 0%
75 statements
« prev ^ index » next coverage.py v7.11.3, created at 2026-01-05 21:46 +0100
1#!/usr/bin/env python3
2"""RAMSES RF - Test the binding protocol with a virtual RF
4NB: This test will likely fail with pytest -n x, because of the protocol's throttle
5limits.
6"""
8import asyncio
9from collections.abc import Callable
10from typing import Any
12import pytest
13import serial # type: ignore[import-untyped]
15from ramses_rf import Message
16from ramses_tx.const import SZ_ACTIVE_HGI, SZ_IS_EVOFW3, Code
17from ramses_tx.protocol import RamsesProtocolT, create_stack, protocol_factory
18from ramses_tx.schemas import DeviceIdT
19from ramses_tx.transport import RamsesTransportT, transport_factory
21from .virtual_rf import HgiFwTypes, VirtualRf
23# ######################################################################################
26async def assert_stack_state(
27 protocol: RamsesProtocolT, transport: RamsesTransportT
28) -> None:
29 assert transport._this_pkt and transport._this_pkt.code == Code._PUZZ
30 assert transport._this_pkt and transport._this_pkt.src.id == GWY_ID
31 assert transport._prev_pkt is None
33 assert transport.get_extra_info(SZ_ACTIVE_HGI) == GWY_ID
34 assert transport.get_extra_info(SZ_IS_EVOFW3) is True
36 assert protocol._active_hgi == GWY_ID
37 assert protocol._is_evofw3 is True
40def _msg_handler(msg: Message) -> None:
41 pass
44# ### TESTS ############################################################################
47async def _test_create_stack(
48 rf: VirtualRf,
49 /,
50 *,
51 protocol_factory_: Callable | None = None,
52 transport_factory_: Callable | None = None,
53 disable_qos: bool | None = False,
54 disable_sending: bool | None = False,
55 enforce_include_list: bool = False,
56 exclude_list: dict[DeviceIdT, dict] | None = None,
57 include_list: dict[DeviceIdT, dict] | None = None,
58 **kwargs: Any, # TODO: these are for the transport_factory
59) -> None:
60 protocol: RamsesProtocolT
61 transport: RamsesTransportT
63 protocol, transport = await create_stack(
64 _msg_handler,
65 disable_qos=disable_qos,
66 disable_sending=disable_sending,
67 enforce_include_list=enforce_include_list,
68 exclude_list=exclude_list,
69 include_list=include_list,
70 **kwargs,
71 )
73 try:
74 await assert_stack_state(protocol, transport)
75 except serial.SerialException as err:
76 transport._close(exc=err)
77 raise
78 except (AssertionError, asyncio.InvalidStateError, TimeoutError):
79 transport.close()
80 raise
81 else:
82 transport.close()
85async def _test_factories(
86 rf: VirtualRf,
87 /,
88 *,
89 protocol_factory_: Callable | None = None,
90 transport_factory_: Callable | None = None,
91 disable_qos: bool | None = False,
92 disable_sending: bool | None = False,
93 enforce_include_list: bool = False,
94 exclude_list: dict[DeviceIdT, dict] | None = None,
95 include_list: dict[DeviceIdT, dict] | None = None,
96 **kwargs: Any, # TODO: these are for the transport_factory
97) -> None:
98 protocol: RamsesProtocolT = protocol_factory(
99 _msg_handler,
100 disable_qos=disable_qos,
101 disable_sending=disable_sending,
102 enforce_include_list=enforce_include_list,
103 exclude_list=exclude_list,
104 include_list=include_list,
105 )
106 transport: RamsesTransportT = await transport_factory(
107 protocol,
108 disable_sending=disable_sending,
109 **kwargs,
110 )
112 try:
113 await assert_stack_state(protocol, transport)
114 except serial.SerialException as err:
115 transport._close(exc=err)
116 raise
117 except (AssertionError, asyncio.InvalidStateError, TimeoutError):
118 transport.close()
119 raise
120 else:
121 transport.close()
124# ######################################################################################
127GWY_ID = "18:111111"
128OTH_ID = "18:123456"
130TEST_SUITE_GOOD = {
131 "00": {
132 "include_list": {},
133 },
134 "01": {
135 "enforce_include_list": True,
136 "include_list": {GWY_ID: {"class": "HGI"}},
137 },
138 "02": {
139 "enforce_include_list": True,
140 "include_list": {GWY_ID: {"class": "HGI"}, OTH_ID: {"class": "HGI"}},
141 },
142 "03": {
143 "enforce_include_list": True,
144 "include_list": {OTH_ID: {"class": "HGI"}, GWY_ID: {"class": "HGI"}},
145 },
146 "04": {
147 "enforce_include_list": True,
148 "include_list": {OTH_ID: {"class": "HGI"}},
149 },
150}
151TEST_SUITE_FAIL = { # fails because exclude_list is checked before active_hgi
152 "10": {
153 "exclude_list": {GWY_ID: {"class": "HGI"}},
154 },
155 "11": {
156 "enforce_include_list": True,
157 "exclude_list": {GWY_ID: {"class": "HGI"}},
158 "include_list": {GWY_ID: {"class": "HGI"}},
159 },
160}
163@pytest.mark.xdist_group(name="virt_serial")
164@pytest.mark.parametrize("idx", TEST_SUITE_GOOD)
165async def test_create_stack(idx: str) -> None:
166 """Check that Transport calls Protocol.connection_made() correctly."""
168 rf = VirtualRf(2, start=True)
169 rf.set_gateway(rf.ports[0], GWY_ID, fw_type=HgiFwTypes.EVOFW3)
171 kwargs = {
172 "port_name": rf.ports[0],
173 "port_config": {},
174 "extra": {"virtual_rf": rf},
175 }
177 try:
178 await _test_create_stack(rf, **TEST_SUITE_GOOD[idx], **kwargs)
179 finally:
180 await rf.stop()
183@pytest.mark.xdist_group(name="virt_serial")
184@pytest.mark.parametrize("idx", TEST_SUITE_GOOD)
185async def test_create_s_alt(idx: str) -> None:
186 """Check that Transport calls Protocol.connection_made() correctly."""
188 rf = VirtualRf(2, start=True)
189 rf.set_gateway(rf.ports[0], GWY_ID, fw_type=HgiFwTypes.EVOFW3)
191 kwargs = {
192 "port_name": rf.ports[0],
193 "port_config": {},
194 "protocol_factory_": protocol_factory,
195 "transport_factory_": transport_factory,
196 "extra": {"virtual_rf": rf},
197 }
199 try:
200 await _test_create_stack(rf, **TEST_SUITE_GOOD[idx], **kwargs)
201 finally:
202 await rf.stop()
205@pytest.mark.xdist_group(name="virt_serial")
206@pytest.mark.parametrize("idx", TEST_SUITE_GOOD)
207async def test_factories_01(idx: str) -> None:
208 """Check that Transport calls Protocol.connection_made() correctly.
210 This is the method used by ramses_tf.gateway.py.
211 """
213 rf = VirtualRf(2, start=True)
214 rf.set_gateway(rf.ports[0], GWY_ID, fw_type=HgiFwTypes.EVOFW3)
216 kwargs = {
217 "port_name": rf.ports[0],
218 "port_config": {},
219 "extra": {"virtual_rf": rf},
220 }
222 try:
223 await _test_factories(rf, **TEST_SUITE_GOOD[idx], **kwargs)
224 finally:
225 await rf.stop()