Skip to content

Commit

Permalink
Avoid timeout and speed-up test.
Browse files Browse the repository at this point in the history
  • Loading branch information
Olivier Pieters committed Nov 5, 2024
1 parent 0b0b3e6 commit c37898a
Showing 1 changed file with 22 additions and 25 deletions.
47 changes: 22 additions & 25 deletions test/management_tests/management_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

0 comments on commit c37898a

Please sign in to comment.