test/testbeds/netns/integration.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3
   4 import getpass
   5 from nepi.core.design import ExperimentDescription, FactoriesProvider
   6 from nepi.core.execute import ExperimentController
   7 from nepi.util import proxy
   8 from nepi.util.constants import DeploymentConfiguration as DC
   9 import os
  10 import shutil
  11 import tempfile
  12 import test_util
  13 import time
  14 import unittest
  15
  16 class NetnsIntegrationTestCase(unittest.TestCase):
  17     def setUp(self):
  18         self.root_dir = tempfile.mkdtemp()
  19
  20     def _test_switched(self, controller_access_config = None,
  21             testbed_access_config = None):
  22         exp_desc = ExperimentDescription()
  23         testbed_id = "netns"
  24         user = getpass.getuser()
  25         netns_provider = FactoriesProvider(testbed_id)
  26         netns_desc = exp_desc.add_testbed_description(netns_provider)
  27         netns_desc.set_attribute_value("homeDirectory", self.root_dir)
  28         #netns_desc.set_attribute_value("enableDebug", True)
  29         node1 = netns_desc.create("Node")
  30         node2 = netns_desc.create("Node")
  31         iface1 = netns_desc.create("NodeInterface")
  32         iface1.set_attribute_value("up", True)
  33         node1.connector("devs").connect(iface1.connector("node"))
  34         ip1 = iface1.add_address()
  35         ip1.set_attribute_value("Address", "10.0.0.1")
  36         iface2 = netns_desc.create("NodeInterface")
  37         iface2.set_attribute_value("up", True)
  38         node2.connector("devs").connect(iface2.connector("node"))
  39         ip2 = iface2.add_address()
  40         ip2.set_attribute_value("Address", "10.0.0.2")
  41         switch = netns_desc.create("Switch")
  42         switch.set_attribute_value("up", True)
  43         iface1.connector("switch").connect(switch.connector("devs"))
  44         iface2.connector("switch").connect(switch.connector("devs"))
  45         app = netns_desc.create("Application")
  46         app.set_attribute_value("command", "ping -qc1 10.0.0.2")
  47         app.set_attribute_value("user", user)
  48         app.connector("node").connect(node1.connector("apps"))
  49         app.enable_trace("stdout")
  50
  51         if testbed_access_config:
  52             for attr in testbed_access_config.attributes:
  53                 if attr.value:
  54                     netns_desc.set_attribute_value(attr.name, attr.value)
  55
  56         xml = exp_desc.to_xml()
  57
  58         controller = proxy.create_experiment_controller(xml,
  59                 controller_access_config)
  60
  61         try:
  62             controller.start()
  63             while not controller.is_finished(app.guid):
  64                 time.sleep(0.5)
  65             ping_result = controller.trace(app.guid, "stdout")
  66             comp_result = """PING 10.0.0.2 (10.0.0.2) 56(84) bytes of data.
  67
  68 --- 10.0.0.2 ping statistics ---
  69 1 packets transmitted, 1 received, 0% packet loss, time 0ms
  70 """
  71             self.assertTrue(ping_result.startswith(comp_result))
  72         finally:
  73             controller.stop()
  74             controller.shutdown()
  75
  76     @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
  77     def test_switched(self):
  78         self._test_switched()
  79
  80     @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
  81     def test_daemonized_controller(self):
  82         access_config = proxy.AccessConfiguration()
  83         access_config.set_attribute_value(DC.DEPLOYMENT_MODE, DC.MODE_DAEMON)
  84         access_config.set_attribute_value(DC.LOG_LEVEL, DC.DEBUG_LEVEL)
  85         access_config.set_attribute_value(DC.ROOT_DIRECTORY, self.root_dir)
  86
  87         self._test_switched(controller_access_config = access_config)
  88
  89     @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
  90     def test_daemonized_tbd(self):
  91         access_config = proxy.AccessConfiguration()
  92         access_config.set_attribute_value(DC.DEPLOYMENT_MODE, DC.MODE_DAEMON)
  93         access_config.set_attribute_value(DC.LOG_LEVEL, DC.DEBUG_LEVEL)
  94         inst_root_dir = os.path.join(self.root_dir, "instance")
  95         os.mkdir(inst_root_dir)
  96         access_config.set_attribute_value(DC.ROOT_DIRECTORY, inst_root_dir)
  97
  98         self._test_switched(testbed_access_config = access_config)
  99
 100     @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
 101     def test_daemonized_all(self):
 102         controller_access_config = proxy.AccessConfiguration()
 103         controller_access_config.set_attribute_value(DC.DEPLOYMENT_MODE, DC.MODE_DAEMON)
 104         controller_access_config.set_attribute_value(DC.ROOT_DIRECTORY, self.root_dir)
 105         controller_access_config.set_attribute_value(DC.LOG_LEVEL, DC.DEBUG_LEVEL)
 106
 107         testbed_access_config = proxy.AccessConfiguration()
 108         testbed_access_config.set_attribute_value(DC.DEPLOYMENT_MODE, DC.MODE_DAEMON)
 109         testbed_access_config.set_attribute_value(DC.LOG_LEVEL, DC.DEBUG_LEVEL)
 110         inst_root_dir = os.path.join(self.root_dir, "instance")
 111         os.mkdir(inst_root_dir)
 112         testbed_access_config.set_attribute_value(DC.ROOT_DIRECTORY, inst_root_dir)
 113
 114         self._test_switched(
 115                 controller_access_config = controller_access_config,
 116                 testbed_access_config = testbed_access_config)
 117
 118     @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
 119     def test_ssh_daemonized_tbd(self):
 120         env = test_util.test_environment()
 121
 122         testbed_access_config = proxy.AccessConfiguration()
 123         testbed_access_config.set_attribute_value(DC.DEPLOYMENT_MODE, DC.MODE_DAEMON)
 124         testbed_access_config.set_attribute_value(DC.LOG_LEVEL, DC.DEBUG_LEVEL)
 125         testbed_access_config.set_attribute_value(DC.DEPLOYMENT_COMMUNICATION, DC.ACCESS_SSH)
 126         testbed_access_config.set_attribute_value(DC.DEPLOYMENT_PORT, env.port)
 127         testbed_access_config.set_attribute_value(DC.USE_AGENT, True)
 128         inst_root_dir = os.path.join(self.root_dir, "instance")
 129         os.mkdir(inst_root_dir)
 130         testbed_access_config.set_attribute_value(DC.ROOT_DIRECTORY, inst_root_dir)
 131
 132         self._test_switched(
 133                 testbed_access_config = testbed_access_config)
 134
 135     def test_sudo_daemonized_tbd(self):
 136         env = test_util.test_environment()
 137
 138         testbed_access_config = proxy.AccessConfiguration()
 139         testbed_access_config.set_attribute_value(DC.DEPLOYMENT_MODE, DC.MODE_DAEMON)
 140         testbed_access_config.set_attribute_value(DC.LOG_LEVEL, DC.DEBUG_LEVEL)
 141         testbed_access_config.set_attribute_value(DC.DEPLOYMENT_COMMUNICATION, DC.ACCESS_SSH)
 142         testbed_access_config.set_attribute_value(DC.DEPLOYMENT_PORT, env.port)
 143         testbed_access_config.set_attribute_value(DC.USE_AGENT, True)
 144         testbed_access_config.set_attribute_value(DC.USE_SUDO, True)
 145         inst_root_dir = os.path.join(self.root_dir, "instance")
 146         os.mkdir(inst_root_dir)
 147         testbed_access_config.set_attribute_value(DC.ROOT_DIRECTORY, inst_root_dir)
 148
 149         self._test_switched(
 150                 testbed_access_config = testbed_access_config)
 151
 152     @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
 153     def test_ssh_daemonized_all(self):
 154         env = test_util.test_environment()
 155
 156         controller_access_config = proxy.AccessConfiguration()
 157         controller_access_config.set_attribute_value(DC.DEPLOYMENT_MODE, DC.MODE_DAEMON)
 158         controller_access_config.set_attribute_value(DC.ROOT_DIRECTORY, self.root_dir)
 159         controller_access_config.set_attribute_value(DC.LOG_LEVEL, DC.DEBUG_LEVEL)
 160         controller_access_config.set_attribute_value(DC.DEPLOYMENT_COMMUNICATION, DC.ACCESS_SSH)
 161         controller_access_config.set_attribute_value(DC.DEPLOYMENT_PORT, env.port)
 162         controller_access_config.set_attribute_value(DC.USE_AGENT, True)
 163
 164         testbed_access_config = proxy.AccessConfiguration()
 165         testbed_access_config.set_attribute_value(DC.DEPLOYMENT_MODE, DC.MODE_DAEMON)
 166         testbed_access_config.set_attribute_value(DC.LOG_LEVEL, DC.DEBUG_LEVEL)
 167         # BUG! IT DOESN'T WORK WITH 2 LEVELS OF SSH!
 168         #testbed_access_config.set_attribute_value(DC.DEPLOYMENT_COMMUNICATION, DC.ACCESS_SSH)
 169         #testbed_access_config.set_attribute_value(DC.DEPLOYMENT_PORT, env.port)
 170         #testbed_access_config.set_attribute_value(DC.USE_AGENT, True)
 171         inst_root_dir = os.path.join(self.root_dir, "instance")
 172         os.mkdir(inst_root_dir)
 173         testbed_access_config.set_attribute_value(DC.ROOT_DIRECTORY, inst_root_dir)
 174
 175         self._test_switched(
 176                 controller_access_config = controller_access_config,
 177                 testbed_access_config = testbed_access_config)
 178
 179     def tearDown(self):
 180         try:
 181             #shutil.rmtree(self.root_dir)
 182             pass
 183         except:
 184             # retry
 185             time.sleep(0.1)
 186             shutil.rmtree(self.root_dir)
 187
 188 if __name__ == '__main__':
 189     unittest.main()
 190