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 version 2 as
   8 #    published by the Free Software Foundation;
   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 # Author: Alina Quereilhac <alina.quereilhac@inria.fr>
  19
  20
  21 from nepi.execution.ec import ExperimentController, ECState
  22 from nepi.execution.scheduler import TaskStatus
  23
  24 import datetime
  25 import time
  26 import unittest
  27
  28 class ExecuteControllersTestCase(unittest.TestCase):
  29     def test_schedule_print(self):
  30         def myfunc():
  31             return 'hola!'
  32
  33         ec = ExperimentController()
  34
  35         tid = ec.schedule("0s", myfunc, track=True)
  36
  37         while True:
  38             task = ec.get_task(tid)
  39             if task.status != TaskStatus.NEW:
  40                 break
  41
  42             time.sleep(1)
  43
  44         self.assertEquals('hola!', task.result)
  45
  46         ec.shutdown()
  47
  48     def test_schedule_date(self):
  49         def get_time():
  50             return datetime.datetime.now()
  51
  52         ec = ExperimentController()
  53
  54         schedule_time = datetime.datetime.now()
  55
  56         tid = ec.schedule("4s", get_time, track=True)
  57
  58         while True:
  59             task = ec.get_task(tid)
  60             if task.status != TaskStatus.NEW:
  61                 break
  62
  63             time.sleep(1)
  64
  65         execution_time = task.result
  66         delta = execution_time - schedule_time
  67         self.assertTrue(delta > datetime.timedelta(seconds=4))
  68         self.assertTrue(delta < datetime.timedelta(seconds=5))
  69
  70         ec.shutdown()
  71
  72     def test_schedule_exception(self):
  73         def raise_error():
  74             # When this task is executed and the error raise,
  75             # the FailureManager should set its failure level to
  76             # TASK_FAILURE
  77             raise RuntimeError("NOT A REAL ERROR. JUST TESTING!")
  78
  79         ec = ExperimentController()
  80
  81         tid = ec.schedule("2s", raise_error, track = True)
  82
  83         while True:
  84             task = ec.get_task(tid)
  85             if task.status != TaskStatus.NEW:
  86                 break
  87
  88             time.sleep(1)
  89
  90         self.assertEquals(task.status, TaskStatus.ERROR)
  91
  92 if __name__ == '__main__':
  93     unittest.main()
  94