test/execution/ec.py

   1 #!/usr/bin/env python
   2
   3 from neco.execution.ec import ExperimentController, ECState
   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         tid = ec.schedule("0s", myfunc, track=True)
  18
  19         while True:
  20             task = ec.get_task(tid)
  21             if task.status != TaskStatus.NEW:
  22                 break
  23
  24             time.sleep(1)
  25
  26         self.assertEquals('hola!', task.result)
  27
  28         ec.shutdown()
  29
  30     def test_schedule_date(self):
  31         def get_time():
  32             return datetime.datetime.now()
  33
  34         ec = ExperimentController()
  35
  36         schedule_time = datetime.datetime.now()
  37
  38         tid = ec.schedule("4s", get_time, track=True)
  39
  40         while True:
  41             task = ec.get_task(tid)
  42             if task.status != TaskStatus.NEW:
  43                 break
  44
  45             time.sleep(1)
  46
  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))
  51
  52         ec.shutdown()
  53
  54     def test_schedule_exception(self):
  55         def raise_error():
  56             raise RuntimeError, "the error"
  57
  58         ec = ExperimentController()
  59         ec.schedule("2s", raise_error)
  60
  61         while ec.ecstate not in [ECState.FAILED, ECState.TERMINATED]:
  62            time.sleep(1)
  63
  64         self.assertEquals(ec.ecstate, ECState.FAILED)
  65         ec.shutdown()
  66
  67
  68 if __name__ == '__main__':
  69     unittest.main()
  70