From c37898afe08eefe2656060447ff6bab4c1f429b9 Mon Sep 17 00:00:00 2001 From: Olivier Pieters Date: Tue, 5 Nov 2024 15:38:33 +0100 Subject: [PATCH] Avoid timeout and speed-up test. --- test/management_tests/management_test.py | 47 +++++++++++------------- 1 file changed, 22 insertions(+), 25 deletions(-) diff --git a/test/management_tests/management_test.py b/test/management_tests/management_test.py index 3de3dfeda..087a7f883 100644 --- a/test/management_tests/management_test.py +++ b/test/management_tests/management_test.py @@ -276,29 +276,26 @@ def send_responses(index): ) await asyncio.sleep(0) - if rate_limit > 0: - # this should raise a timeout, since the data was never confirmed - with pytest.raises(ManagementConnectionTimeout): - # the request is queued - await time_travel(0.5 / rate_limit) - assert xknx.cemi_handler.send_telegram.call_args_list == [] - - # the request should now be sent - await time_travel(0.6 / rate_limit) - - assert xknx.cemi_handler.send_telegram.call_args_list == [ - call( - Telegram( - destination_address=ia, - tpci=tpci.TDataConnected(1), - payload=apci.DeviceDescriptorRead(descriptor=0), - ) - ), - ] - - await task - else: - # should work as expected - send_responses(1) + if rate_limit: + await time_travel(0.5 / rate_limit) - await task + # the request is still queued + assert not xknx.cemi_handler.send_telegram.call_args_list + + await time_travel(0.5 / rate_limit) + + # the requests should be sent now, the behaviour should match no rate limit + + assert xknx.cemi_handler.send_telegram.call_args_list == [ + call( + Telegram( + destination_address=ia, + tpci=tpci.TDataConnected(1), + payload=apci.DeviceDescriptorRead(descriptor=0), + ) + ), + ] + + send_responses(1) + + await task