StealthOS/subiquity
StealthOS
/
subiquity
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

async_helpers: simplify task done check 

Another version of the task-created-yet problem is the non-blocking
check to see if it is done or not.  Add a wrapper here to simplify
calling code.
This commit is contained in:
Dan Bungert 2023-03-10 16:41:23 -07:00
parent 075b06ce70
commit 98a2ff8647
2 changed files with 8 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
subiquitycore/async_helpers.py
View File

@ -113,3 +113,8 @@ class SingleInstanceTask:
return await self.task return await self.task
except asyncio.CancelledError: except asyncio.CancelledError:
pass pass
def done(self):
if self.task is None:
return False
return self.task.done()

3
subiquitycore/tests/test_async_helpers.py
View File

@ -56,10 +56,13 @@ class TestSITWait(unittest.IsolatedAsyncioTestCase):
sit = SingleInstanceTask(fn) sit = SingleInstanceTask(fn)
await sit.start() await sit.start()
await asyncio.wait_for(sit.wait(), timeout=1.0) await asyncio.wait_for(sit.wait(), timeout=1.0)
self.assertTrue(sit.done())
async def test_wait_not_started(self): async def test_wait_not_started(self):
async def fn(): async def fn():
self.fail('not supposed to be called') self.fail('not supposed to be called')
sit = SingleInstanceTask(fn) sit = SingleInstanceTask(fn)
self.assertFalse(sit.done())
with self.assertRaises(asyncio.TimeoutError): with self.assertRaises(asyncio.TimeoutError):
await asyncio.wait_for(sit.wait(), timeout=0.1) await asyncio.wait_for(sit.wait(), timeout=0.1)
self.assertFalse(sit.done())