Andreas/evennia
1
0
Fork 0
mirror of https://github.com/evennia/evennia.git synced 2026-03-23 00:06:30 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

TaskHandler.active method created

Browse source 
TaskHandler.active method created to check if a task is currently active.

test_delay unit test passes.
This commit is contained in:
davewiththenicehat 2021-04-18 09:59:45 -04:00
parent af44237838
commit bbc60b0340
2 changed files with 35 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

26
evennia/scripts/taskhandler.py
View file

@ -230,6 +230,32 @@ class TaskHandler(object):
else:
return False
def active(self, task_id):
"""
Check if a task is active (has not been called yet).
Args:
task_id (int): an existing task ID.
Returns:
True (bool): If a task is active (has not been called yet).
False (bool): if the task
is not active (has already been called),
does not exist
"""
if task_id in self.tasks:
# if the task has not been run, cancel it
d = self.get_deferred(task_id)
if d: # it is remotely possible for a task to not have a deferral
if d.called:
return False
else: # the callback has not been called yet.
return True
else: # this task has no deferral, and could not have been called
return True
else:
return False
def cancel(self, task_id):
"""

9
evennia/utils/tests/test_utils.py
View file

@ -331,23 +331,28 @@ class TestDelay(EvenniaTest):
self.char1.ndb.dummy_var = False
# test a persistent deferral, that completes after delay time
task_id = utils.delay(timedelay, dummy_func, self.char1.dbref, persistent=True)
self.assertTrue(_TASK_HANDLER.active(task_id))
_TASK_HANDLER.clock.advance(timedelay) # make time pass
self.assertEqual(self.char1.ndb.dummy_var, 'dummy_func ran')
self.char1.ndb.dummy_var = False
# test a non persisten deferral, that completes after delay time.
utils.delay(timedelay, dummy_func, self.char1.dbref)
self.assertTrue(_TASK_HANDLER.active(task_id))
_TASK_HANDLER.clock.advance(timedelay) # make time pass
self.assertEqual(self.char1.ndb.dummy_var, 'dummy_func ran')
self.char1.ndb.dummy_var = False
# test a non persisten deferral, with a short timedelay
utils.delay(.1, dummy_func, self.char1.dbref)
self.assertTrue(_TASK_HANDLER.active(task_id))
_TASK_HANDLER.clock.advance(.1) # make time pass
self.assertEqual(self.char1.ndb.dummy_var, 'dummy_func ran')
self.char1.ndb.dummy_var = False
# test canceling a deferral.
# after this the task_id 1 remains used by this canceled but unused task
task_id = utils.delay(timedelay, dummy_func, self.char1.dbref)
self.assertTrue(_TASK_HANDLER.active(task_id))
success = _TASK_HANDLER.cancel(task_id)
self.assertFalse(_TASK_HANDLER.active(task_id))
self.assertTrue(success)
self.assertTrue(_TASK_HANDLER.exists(task_id))
_TASK_HANDLER.clock.advance(timedelay) # make time pass
@ -355,15 +360,19 @@ class TestDelay(EvenniaTest):
self.char1.ndb.dummy_var = False
# test removing an active task
task_id = utils.delay(timedelay, dummy_func, self.char1.dbref)
self.assertTrue(_TASK_HANDLER.active(task_id))
success = _TASK_HANDLER.remove(task_id)
self.assertFalse(_TASK_HANDLER.active(task_id))
_TASK_HANDLER.clock.advance(timedelay) # make time pass
self.assertEqual(self.char1.ndb.dummy_var, False)
self.assertFalse(_TASK_HANDLER.exists(task_id))
self.char1.ndb.dummy_var = False
# test removing a canceled active task
task_id = utils.delay(timedelay, dummy_func, self.char1.dbref)
self.assertTrue(_TASK_HANDLER.active(task_id))
deferal_inst = _TASK_HANDLER.get_deferred(task_id)
deferal_inst.cancel()
self.assertFalse(_TASK_HANDLER.active(task_id))
success = _TASK_HANDLER.remove(task_id)
_TASK_HANDLER.clock.advance(timedelay) # make time pass
self.assertEqual(self.char1.ndb.dummy_var, False)