test/execution/ec.py

   1 #!/usr/bin/env python
   2
   3 from neco.execution.ec import ExperimentController
   4 from neco.execution.tasks import TaskStatus
   5
   6 import datetime
   7 import unittest
   8
   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)
  14
  15         try:
  16             ec = ExperimentController()
  17
  18             tid = ec.schedule("0s", myfunc)
  19             status = None
  20             while status != TaskStatus.SUCCESS:
  21                 (status, result) = ec.task_info(tid)
  22
  23             self.assertEquals(id(ec), result)
  24         finally:
  25             ec.terminate()
  26
  27     def test_schedule_date(self):
  28         def get_time(ec_weakref):
  29             timestamp = datetime.datetime.now()
  30             return (TaskStatus.SUCCESS, timestamp)
  31
  32         try:
  33             ec = ExperimentController()
  34
  35             schedule_time = datetime.datetime.now()
  36
  37             tid = ec.schedule("4s", get_time)
  38             status = None
  39             while status != TaskStatus.SUCCESS:
  40                 (status, execution_time) = ec.task_info(tid)
  41
  42             delta = execution_time - schedule_time
  43             self.assertTrue(delta > datetime.timedelta(seconds=4))
  44             self.assertTrue(delta < datetime.timedelta(seconds=5))
  45
  46         finally:
  47             ec.terminate()
  48
  49
  50 if __name__ == '__main__':
  51     unittest.main()
  52