test/resources/omf/set_hook.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: Julien Tribino <julien.tribino@inria.fr>
  20
  21
  22 from nepi.execution.resource import ResourceFactory, clsinit_copy, ResourceManager, ResourceAction, ResourceState
  23 from nepi.execution.ec import ExperimentController
  24 from nepi.execution.attribute import Attribute, Flags
  25
  26 from nepi.resources.omf.node import OMFNode
  27 from nepi.resources.omf.application import OMFApplication
  28
  29 from nepi.util.timefuncs import *
  30
  31 import time
  32 import unittest
  33
  34 class DummyEC(ExperimentController):
  35     pass
  36
  37 @clsinit_copy
  38 class OMFDummyApplication(OMFApplication):
  39     _rtype = "omf::DummyApplication"
  40
  41     @classmethod
  42     def _register_attributes(cls):
  43         test = Attribute("test", "Input of the application", default = 0, set_hook = cls.test_hook)
  44         cls._register_attribute(test)
  45
  46     @classmethod
  47     def test_hook(cls, old_value, new_value):
  48         new_value *= 10
  49         print "Change the value of test from "+ str(old_value) +" to : " + str(new_value)
  50         return new_value
  51
  52
  53 class OMFTestSet(unittest.TestCase):
  54
  55     def test_set_hook(self):
  56         self.ec = DummyEC(exp_id = "30")
  57
  58         ResourceFactory.register_type(OMFDummyApplication)
  59
  60         self.node1 = self.ec.register_resource("omf::Node")
  61         self.ec.set(self.node1, 'hostname', 'omf.plexus.wlab17')
  62         self.ec.set(self.node1, 'xmppSlice', "nepi")
  63         self.ec.set(self.node1, 'xmppHost', "xmpp-plexus.onelab.eu")
  64         self.ec.set(self.node1, 'xmppPort', "5222")
  65         self.ec.set(self.node1, 'xmppPassword', "1234")
  66
  67         self.app1 = self.ec.register_resource("omf::DummyApplication")
  68         self.ec.set(self.app1, 'appid', 'Test#1')
  69         self.ec.set(self.app1, 'path', "/usr/bin/ping")
  70         self.ec.set(self.app1, 'args', "")
  71         self.ec.set(self.app1, 'env', "")
  72         self.ec.set(self.app1, 'xmppSlice', "nepi")
  73         self.ec.set(self.app1, 'xmppHost', "xmpp-plexus.onelab.eu")
  74         self.ec.set(self.app1, 'xmppPort', "5222")
  75         self.ec.set(self.app1, 'xmppPassword', "1234")
  76
  77         self.ec.register_connection(self.app1, self.node1)
  78
  79         self.ec.register_condition(self.app1, ResourceAction.STOP, self.app1, ResourceState.STARTED , "10s")
  80
  81         self.ec.deploy()
  82
  83         time.sleep(3)
  84         print "First try to change the STDIN"
  85         self.ec.set(self.app1, 'test', 3)
  86
  87         self.assertEquals(self.ec.get(self.app1, 'test'), 30)
  88
  89         time.sleep(3)
  90         print "Second try to change the STDIN"
  91         self.ec.set(self.app1, 'test', 101)
  92         self.assertEquals(self.ec.get(self.app1, 'test'), 1010)
  93
  94         self.ec.wait_finished([self.app1])
  95
  96         # Stop Experiment
  97         self.ec.shutdown()
  98
  99
 100 if __name__ == '__main__':
 101     unittest.main()
 102
 103
 104