test/execution/ec.py

   1 #!/usr/bin/env python
   2
   3 from neco.execution.ec import ExperimentController
   4 from neco.execution.scheduler import TaskStatus
   5
   6 import datetime
   7 import time
   8 import unittest
   9
  10 class ExecuteControllersTestCase(unittest.TestCase):
  11     def test_schedule_print(self):
  12         def myfunc():
  13             return 'hola!'
  14
  15         ec = ExperimentController()
  16
  17         try:
  18             tid = ec.schedule("0s", myfunc, track=True)
  19
  20             while True:
  21                 task = ec.get_task(tid)
  22                 if task.status != TaskStatus.NEW:
  23                     break
  24
  25                 time.sleep(1)
  26
  27             self.assertEquals('hola!', task.result)
  28
  29         finally:
  30             ec.shutdown()
  31
  32     def test_schedule_date(self):
  33         def get_time():
  34             return datetime.datetime.now()
  35
  36         ec = ExperimentController()
  37
  38         try:
  39             schedule_time = datetime.datetime.now()
  40
  41             tid = ec.schedule("4s", get_time, track=True)
  42
  43             while True:
  44                 task = ec.get_task(tid)
  45                 if task.status != TaskStatus.NEW:
  46                     break
  47
  48                 time.sleep(1)
  49
  50             execution_time = task.result
  51             delta = execution_time - schedule_time
  52             self.assertTrue(delta > datetime.timedelta(seconds=4))
  53             self.assertTrue(delta < datetime.timedelta(seconds=5))
  54
  55         finally:
  56             ec.shutdown()
  57
  58
  59 if __name__ == '__main__':
  60     unittest.main()
  61