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.constants import AF_INET, STATUS_FINISHED
   8 import os
   9 import shutil
  10 import test_util
  11 import time
  12 import unittest
  13 import uuid
  14
  15 class NetnsIntegrationTestCase(unittest.TestCase):
  16     def setUp(self):
  17         self._home_dir = os.path.join(os.getenv("HOME"), ".nepi",
  18                 str(uuid.uuid1()))
  19         os.makedirs(self._home_dir)
  20
  21     @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
  22     def test_local_if(self):
  23         exp_desc = ExperimentDescription()
  24         testbed_version = "01"
  25         testbed_id = "netns"
  26         user = getpass.getuser()
  27         netns_provider = FactoriesProvider(testbed_id, testbed_version)
  28         netns_desc = exp_desc.add_testbed_description(netns_provider)
  29         netns_desc.set_attribute_value("homeDirectory", self._home_dir)
  30         #netns_desc.set_attribute_value("enableDebug", True)
  31         node1 = netns_desc.create("Node")
  32         node2 = netns_desc.create("Node")
  33         iface1 = netns_desc.create("NodeInterface")
  34         iface1.set_attribute_value("up", True)
  35         node1.connector("devs").connect(iface1.connector("node"))
  36         ip1 = iface1.add_address()
  37         ip1.set_attribute_value("Address", "10.0.0.1")
  38         iface2 = netns_desc.create("NodeInterface")
  39         iface2.set_attribute_value("up", True)
  40         node2.connector("devs").connect(iface2.connector("node"))
  41         ip2 = iface2.add_address()
  42         ip2.set_attribute_value("Address", "10.0.0.2")
  43         switch = netns_desc.create("Switch")
  44         switch.set_attribute_value("up", True)
  45         iface1.connector("switch").connect(switch.connector("devs"))
  46         iface2.connector("switch").connect(switch.connector("devs"))
  47         app = netns_desc.create("Application")
  48         app.set_attribute_value("command", "ping -qc1 10.0.0.2")
  49         app.set_attribute_value("user", user)
  50         app.connector("node").connect(node1.connector("apps"))
  51         app.enable_trace("stdout")
  52         xml = exp_desc.to_xml()
  53
  54         controller = ExperimentController(xml)
  55         controller.start()
  56         while controller.status(app.guid) != STATUS_FINISHED:
  57             time.sleep(0.5)
  58         ping_result = controller.trace(netns_desc.guid, app.guid, "stdout")
  59         comp_result = """PING 10.0.0.2 (10.0.0.2) 56(84) bytes of data.
  60
  61 --- 10.0.0.2 ping statistics ---
  62 1 packets transmitted, 1 received, 0% packet loss, time 0ms
  63 """
  64         self.assertTrue(ping_result.startswith(comp_result))
  65         controller.stop()
  66         controller.shutdown()
  67
  68     def tearDown(self):
  69         shutil.rmtree(self._home_dir)
  70
  71 if __name__ == '__main__':
  72     unittest.main()
  73