3 from neco.execution.ec import ExperimentController
4 from neco.execution.tasks import TaskStatus
9 class ExecuteControllersTestCase(unittest.TestCase):
10 def test_schedule_print(self):
11 def myfunc(ec_weakref):
12 result = id(ec_weakref())
13 return (TaskStatus.SUCCESS, result)
16 ec = ExperimentController()
18 tid = ec.schedule("0s", myfunc)
20 while status != TaskStatus.SUCCESS:
21 (status, result) = ec.task_info(tid)
23 self.assertEquals(id(ec), result)
27 def test_schedule_date(self):
28 def get_time(ec_weakref):
29 timestamp = datetime.datetime.now()
30 return (TaskStatus.SUCCESS, timestamp)
33 ec = ExperimentController()
35 schedule_time = datetime.datetime.now()
37 tid = ec.schedule("4s", get_time)
39 while status != TaskStatus.SUCCESS:
40 (status, execution_time) = ec.task_info(tid)
42 delta = execution_time - schedule_time
43 self.assertTrue(delta > datetime.timedelta(seconds=4))
44 self.assertTrue(delta < datetime.timedelta(seconds=5))
50 if __name__ == '__main__':