dekun
52 lines
1.3 KiB
Python
52 lines
1.3 KiB
Python
import asyncio
|
|
import time
|
|
import unittest
|
|
|
|
from lib.hub.hub_poll_wait_lib import wait_poll_interval
|
|
|
|
|
|
class TestWaitPollInterval(unittest.IsolatedAsyncioTestCase):
|
|
async def test_ignores_refresh_storm_until_interval(self):
|
|
refresh = asyncio.Event()
|
|
stop = asyncio.Event()
|
|
started = time.monotonic()
|
|
|
|
async def storm():
|
|
for _ in range(30):
|
|
refresh.set()
|
|
await asyncio.sleep(0.01)
|
|
|
|
t = asyncio.create_task(storm())
|
|
await wait_poll_interval(
|
|
refresh=refresh,
|
|
stop=stop,
|
|
interval_sec=0.25,
|
|
started_at=started,
|
|
min_early_wake_sec=0.2,
|
|
)
|
|
t.cancel()
|
|
elapsed = time.monotonic() - started
|
|
self.assertGreaterEqual(elapsed, 0.18)
|
|
|
|
async def test_stop_ends_early(self):
|
|
refresh = asyncio.Event()
|
|
stop = asyncio.Event()
|
|
started = time.monotonic()
|
|
|
|
async def stopper():
|
|
await asyncio.sleep(0.05)
|
|
stop.set()
|
|
|
|
asyncio.create_task(stopper())
|
|
await wait_poll_interval(
|
|
refresh=refresh,
|
|
stop=stop,
|
|
interval_sec=2.0,
|
|
started_at=started,
|
|
)
|
|
self.assertLess(time.monotonic() - started, 0.5)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|