test/execution/ec.py

   1 #!/usr/bin/env python
   2 #
   3 #    NEPI, a framework to manage network experiments
   4 #    Copyright (C) 2013 INRIA
   5 #
   6 #    This program is free software: you can redistribute it and/or modify
   7 #    it under the terms of the GNU General Public License as published by
   8 #    the Free Software Foundation, either version 3 of the License, or
   9 #    (at your option) any later version.
  10 #
  11 #    This program is distributed in the hope that it will be useful,
  12 #    but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 #    GNU General Public License for more details.
  15 #
  16 #    You should have received a copy of the GNU General Public License
  17 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18 #
  19 # Author: Alina Quereilhac <alina.quereilhac@inria.fr>
  20
  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             # When this task is executed and the error raise,
  76             # the FailureManager should set its failure level to
  77             # TASK_FAILURE
  78             raise RuntimeError, "NOT A REAL ERROR. JUST TESTING!"
  79
  80         ec = ExperimentController()
  81
  82         tid = ec.schedule("2s", raise_error, track = True)
  83
  84         while True:
  85             task = ec.get_task(tid)
  86             if task.status != TaskStatus.NEW:
  87                 break
  88
  89             time.sleep(1)
  90
  91         self.assertEquals(task.status, TaskStatus.ERROR)
  92
  93 if __name__ == '__main__':
  94     unittest.main()
  95