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