test/testbeds/netns/execute.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3
   4 import getpass
   5 from nepi.core.design import AF_INET
   6 from nepi.testbeds import netns
   7 import os
   8 import shutil
   9 import test_util
  10 import time
  11 import unittest
  12 import uuid
  13
  14 class NetnsExecuteTestCase(unittest.TestCase):
  15     def setUp(self):
  16         self._home_dir = os.path.join(os.getenv("HOME"), ".nepi",
  17                 str(uuid.uuid1()))
  18         os.makedirs(self._home_dir)
  19
  20     @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
  21     def test_run_ping_if(self):
  22         user = getpass.getuser()
  23         testbed_version = "01"
  24         config = netns.TestbedConfiguration()
  25         config.set_attribute_value("HomeDirectory", self._home_dir)
  26         instance = netns.TestbedInstance(testbed_version, config)
  27         instance.create(2, "Node")
  28         instance.create(3, "Node")
  29         instance.create(4, "NodeInterface")
  30         instance.create_set(4, "up", True)
  31         instance.connect(2, "devs", 4, "node")
  32         instance.add_adddress(4, AF_INET, "10.0.0.1", 24, None)
  33         instance.create(5, "NodeInterface")
  34         instance.create_set(5, "up", True)
  35         instance.connect(3, "devs", 5, "node")
  36         instance.add_adddress(5, AF_INET, "10.0.0.2", 24, None)
  37         instance.create(6, "Switch")
  38         instance.create_set(6, "up", True)
  39         instance.connect(4, "switch", 6, "devs")
  40         instance.connect(5, "switch", 6, "devs")
  41         instance.create(7, "Application")
  42         instance.create_set(7, "command", "ping -qc1 10.0.0.2")
  43         instance.create_set(7, "user", user)
  44         instance.add_trace(7, "stdout")
  45         instance.connect(7, "node", 2, "apps")
  46
  47         instance.do_create()
  48         instance.do_connect()
  49         instance.do_configure()
  50         instance.start()
  51         time.sleep(2)
  52         ping_result = instance.trace(7, "stdout")
  53         comp_result = """PING 10.0.0.2 (10.0.0.2) 56(84) bytes of data.
  54
  55 --- 10.0.0.2 ping statistics ---
  56 1 packets transmitted, 1 received, 0% packet loss, time 0ms
  57 """
  58         self.assertTrue(ping_result.startswith(comp_result))
  59         instance.stop()
  60         instance.shutdown()
  61
  62     @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
  63     def test_run_ping_p2pif(self):
  64         user = getpass.getuser()
  65         testbed_version = "01"
  66         config = netns.TestbedConfiguration()
  67         config.set_attribute_value("HomeDirectory", self._home_dir)
  68         instance = netns.TestbedInstance(testbed_version, config)
  69         instance.create(2, "Node")
  70         instance.create(3, "Node")
  71         instance.create(4, "P2PNodeInterface")
  72         instance.create_set(4, "up", True)
  73         instance.connect(2, "devs", 4, "node")
  74         instance.add_adddress(4, AF_INET, "10.0.0.1", 24, None)
  75         instance.create(5, "P2PNodeInterface")
  76         instance.create_set(5, "up", True)
  77         instance.connect(3, "devs", 5, "node")
  78         instance.add_adddress(5, AF_INET, "10.0.0.2", 24, None)
  79         instance.connect(4, "p2p", 5, "p2p")
  80         instance.create(6, "Application")
  81         instance.create_set(6, "command", "ping -qc1 10.0.0.2")
  82         instance.create_set(6, "user", user)
  83         instance.add_trace(6, "stdout")
  84         instance.connect(6, "node", 2, "apps")
  85
  86         instance.do_create()
  87         instance.do_connect()
  88         instance.do_configure()
  89         instance.start()
  90         time.sleep(2)
  91         ping_result = instance.trace(6, "stdout")
  92         comp_result = """PING 10.0.0.2 (10.0.0.2) 56(84) bytes of data.
  93
  94 --- 10.0.0.2 ping statistics ---
  95 1 packets transmitted, 1 received, 0% packet loss, time 0ms
  96 """
  97         self.assertTrue(ping_result.startswith(comp_result))
  98         instance.stop()
  99         instance.shutdown()
 100
 101     @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
 102     def test_run_ping_routing(self):
 103         user = getpass.getuser()
 104         testbed_version = "01"
 105         config = netns.TestbedConfiguration()
 106         config.set_attribute_value("HomeDirectory", self._home_dir)
 107         instance = netns.TestbedInstance(testbed_version, config)
 108         instance.create(2, "Node")
 109         instance.create(3, "Node")
 110         instance.create(4, "Node")
 111         instance.create(5, "NodeInterface")
 112         instance.create_set(5, "up", True)
 113         instance.connect(2, "devs", 5, "node")
 114         instance.add_adddress(5, AF_INET, "10.0.0.1", 24, None)
 115         instance.create(6, "NodeInterface")
 116         instance.create_set(6, "up", True)
 117         instance.connect(3, "devs", 6, "node")
 118         instance.add_adddress(6, AF_INET, "10.0.0.2", 24, None)
 119         instance.create(7, "NodeInterface")
 120         instance.create_set(7, "up", True)
 121         instance.connect(3, "devs", 7, "node")
 122         instance.add_adddress(7, AF_INET, "10.0.1.1", 24, None)
 123         instance.create(8, "NodeInterface")
 124         instance.create_set(8, "up", True)
 125         instance.connect(4, "devs", 8, "node")
 126         instance.add_adddress(8, AF_INET, "10.0.1.2", 24, None)
 127         instance.create(9, "Switch")
 128         instance.create_set(9, "up", True)
 129         instance.connect(5, "switch", 9, "devs")
 130         instance.connect(6, "switch", 9, "devs")
 131         instance.create(10, "Switch")
 132         instance.create_set(10, "up", True)
 133         instance.connect(7, "switch", 10, "devs")
 134         instance.connect(8, "switch", 10, "devs")
 135         instance.create(11, "Application")
 136         instance.create_set(11, "command", "ping -qc1 10.0.1.2")
 137         instance.create_set(11, "user", user)
 138         instance.add_trace(11, "stdout")
 139         instance.connect(11, "node", 2, "apps")
 140
 141         instance.add_route(2, "10.0.1.0", 24, "10.0.0.2")
 142         instance.add_route(4, "10.0.0.0", 24, "10.0.1.1")
 143
 144         instance.do_create()
 145         instance.do_connect()
 146         instance.do_configure()
 147         instance.start()
 148         time.sleep(2)
 149         ping_result = instance.trace(11, "stdout")
 150         comp_result = """PING 10.0.0.2 (10.0.0.2) 56(84) bytes of data.
 151
 152 --- 10.0.0.2 ping statistics ---
 153 1 packets transmitted, 1 received, 0% packet loss, time 0ms
 154 """
 155         self.assertTrue(ping_result.startswith(comp_result))
 156         instance.stop()
 157         instance.shutdown()
 158
 159     def tearDown(self):
 160         shutil.rmtree(self._home_dir)
 161
 162 if __name__ == '__main__':
 163     unittest.main()
 164