3 from nepi.execution.ec import ExperimentController, ECState
4 from nepi.execution.scheduler import TaskStatus
10 class ExecuteControllersTestCase(unittest.TestCase):
11 def test_schedule_print(self):
15 ec = ExperimentController()
17 tid = ec.schedule("0s", myfunc, track=True)
20 task = ec.get_task(tid)
21 if task.status != TaskStatus.NEW:
26 self.assertEquals('hola!', task.result)
30 def test_schedule_date(self):
32 return datetime.datetime.now()
34 ec = ExperimentController()
36 schedule_time = datetime.datetime.now()
38 tid = ec.schedule("4s", get_time, track=True)
41 task = ec.get_task(tid)
42 if task.status != TaskStatus.NEW:
47 execution_time = task.result
48 delta = execution_time - schedule_time
49 self.assertTrue(delta > datetime.timedelta(seconds=4))
50 self.assertTrue(delta < datetime.timedelta(seconds=5))
54 def test_schedule_exception(self):
56 raise RuntimeError, "the error"
58 ec = ExperimentController()
59 ec.schedule("2s", raise_error)
61 while ec.ecstate not in [ECState.FAILED, ECState.TERMINATED]:
64 self.assertEquals(ec.ecstate, ECState.FAILED)
68 if __name__ == '__main__':