test/resources/omf/omf6_vlc_traces.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: Julien Tribino <julien.tribino@inria.fr>
  19
  20
  21 from nepi.execution.resource import ResourceFactory, ResourceManager, ResourceAction, ResourceState
  22 from nepi.execution.ec import ExperimentController
  23
  24 from nepi.resources.omf.node import OMFNode
  25 from nepi.resources.omf.application import OMFApplication
  26 from nepi.resources.omf.interface import OMFWifiInterface
  27 from nepi.resources.omf.channel import OMFChannel
  28 from nepi.resources.omf.omf_api_factory import OMFAPIFactory
  29
  30 from nepi.util.timefuncs import *
  31
  32 import os
  33 import time
  34 import unittest
  35
  36 class OMFPingNormalCase(unittest.TestCase):
  37     def test_deploy(self):
  38         ec = ExperimentController(exp_id = "5421" )
  39
  40         self.node1 = ec.register_resource("omf::Node")
  41         ec.set(self.node1, 'hostname', 'wlab12')
  42         ec.set(self.node1, 'xmppUser', "nepi")
  43         ec.set(self.node1, 'xmppServer', "xmpp-plexus.onelab.eu")
  44         ec.set(self.node1, 'xmppPort', "5222")
  45         ec.set(self.node1, 'xmppPassword', "1234")
  46
  47         self.iface1 = ec.register_resource("omf::WifiInterface")
  48         ec.set(self.iface1, 'name', "wlan0")
  49         ec.set(self.iface1, 'mode', "adhoc")
  50         ec.set(self.iface1, 'hw_mode', "g")
  51         ec.set(self.iface1, 'essid', "vlcexp")
  52         ec.set(self.iface1, 'ip', "10.0.0.17/24")
  53
  54         self.channel = ec.register_resource("omf::Channel")
  55         ec.set(self.channel, 'channel', "6")
  56         ec.set(self.channel, 'xmppUser', "nepi")
  57         ec.set(self.channel, 'xmppServer', "xmpp-plexus.onelab.eu")
  58         ec.set(self.channel, 'xmppPort', "5222")
  59         ec.set(self.channel, 'xmppPassword', "1234")
  60
  61         self.app1 = ec.register_resource("omf::Application")
  62         ec.set(self.app1, 'appid', 'Vlc#1')
  63         ec.set(self.app1, 'command', "ping -c5 10.0.0.17")
  64
  65         ec.register_connection(self.app1, self.node1)
  66         ec.register_connection(self.node1, self.iface1)
  67         ec.register_connection(self.iface1, self.channel)
  68
  69         ec.register_condition(self.app1, ResourceAction.STOP, self.app1, ResourceState.STARTED , "10s")
  70
  71         ec.deploy()
  72
  73         ec.wait_finished(self.app1)
  74
  75         stdout_1 = ec.trace(self.app1, "stdout")
  76         stderr_1 = ec.trace(self.app1, "stderr")
  77
  78         if stdout_1:
  79             with open("app1_out.txt", "w") as f:
  80                 f.write(stdout_1)
  81
  82         if stderr_1:
  83             with open("app1_err.txt", "w") as f:
  84                 f.write(stderr_1)
  85
  86         self.assertEqual(ec.get_resource(self.node1).state, ResourceState.STARTED)
  87         self.assertEqual(ec.get_resource(self.iface1).state, ResourceState.STARTED)
  88         self.assertEqual(ec.get_resource(self.channel).state, ResourceState.STARTED)
  89         self.assertEqual(ec.get_resource(self.app1).state, ResourceState.STOPPED)
  90
  91         ec.shutdown()
  92
  93         self.assertEqual(ec.get_resource(self.node1).state, ResourceState.RELEASED)
  94         self.assertEqual(ec.get_resource(self.iface1).state, ResourceState.RELEASED)
  95         self.assertEqual(ec.get_resource(self.channel).state, ResourceState.RELEASED)
  96         self.assertEqual(ec.get_resource(self.app1).state, ResourceState.RELEASED)
  97
  98         with open("app1_out.txt", "r") as t:
  99             l = t.readlines()
 100             self.assertEqual(l[0], "PING 10.0.0.17 (10.0.0.17) 56(84) bytes of data.\n")
 101             self.assertIn("5 packets transmitted, 5 received, 0% packet loss, time", l[-2])
 102             self.assertIn("rtt min/avg/max/mdev = ", l[-1])
 103
 104         os.remove("app1_out.txt")
 105
 106
 107
 108 if __name__ == '__main__':
 109     unittest.main()
 110
 111
 112