test/execution/ec.py

   1 """
   2     NEPI, a framework to manage network experiments
   3     Copyright (C) 2013 INRIA
   4
   5     This program is free software: you can redistribute it and/or modify
   6     it under the terms of the GNU General Public License as published by
   7     the Free Software Foundation, either version 3 of the License, or
   8     (at your option) any later version.
   9
  10     This program is distributed in the hope that it will be useful,
  11     but WITHOUT ANY WARRANTY; without even the implied warranty of
  12     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13     GNU General Public License for more details.
  14
  15     You should have received a copy of the GNU General Public License
  16     along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17
  18 """
  19
  20 #!/usr/bin/env python
  21
  22 from nepi.execution.ec import ExperimentController, ECState
  23 from nepi.execution.scheduler import TaskStatus
  24
  25 import datetime
  26 import time
  27 import unittest
  28
  29 class ExecuteControllersTestCase(unittest.TestCase):
  30     def test_schedule_print(self):
  31         def myfunc():
  32             return 'hola!'
  33
  34         ec = ExperimentController()
  35
  36         tid = ec.schedule("0s", myfunc, track=True)
  37
  38         while True:
  39             task = ec.get_task(tid)
  40             if task.status != TaskStatus.NEW:
  41                 break
  42
  43             time.sleep(1)
  44
  45         self.assertEquals('hola!', task.result)
  46
  47         ec.shutdown()
  48
  49     def test_schedule_date(self):
  50         def get_time():
  51             return datetime.datetime.now()
  52
  53         ec = ExperimentController()
  54
  55         schedule_time = datetime.datetime.now()
  56
  57         tid = ec.schedule("4s", get_time, track=True)
  58
  59         while True:
  60             task = ec.get_task(tid)
  61             if task.status != TaskStatus.NEW:
  62                 break
  63
  64             time.sleep(1)
  65
  66         execution_time = task.result
  67         delta = execution_time - schedule_time
  68         self.assertTrue(delta > datetime.timedelta(seconds=4))
  69         self.assertTrue(delta < datetime.timedelta(seconds=5))
  70
  71         ec.shutdown()
  72
  73     def test_schedule_exception(self):
  74         def raise_error():
  75             raise RuntimeError, "the error"
  76
  77         ec = ExperimentController()
  78         ec.schedule("2s", raise_error)
  79
  80         while ec.ecstate not in [ECState.FAILED, ECState.TERMINATED]:
  81            time.sleep(1)
  82
  83         self.assertEquals(ec.ecstate, ECState.FAILED)
  84         ec.shutdown()
  85
  86
  87 if __name__ == '__main__':
  88     unittest.main()
  89