Vlan

dsatest/bench/interface.py

@@ -10,6 +10,7 @@ def __init__(self, name, machine, switch=None, port_id=None):
         self.machine = machine
         self.switch = switch
         self.port_id = port_id
+        self.vlan_interfaces = {}
 
 
     def __repr__(self):
@@ -39,3 +40,19 @@ def ping(self, destination, count=None, deadline=None):
 
     def arp_get(self, address):
         return self.machine.arp_get(address, self.name)
+
+    def add_vlan(self, vid):
+        ret = self.machine.add_vid(self.name, vid)
+        """ Treat existing interfaces as okay """
+        if ret != 0:
+            return ret
+        vlan = Interface("{}.{}".format(self.name, vid),
+                         self.machine, self.switch, self.port_id)
+        self.vlan_interfaces[vid] = vlan
+
+    def del_vlan(self, vid):
+        ret = self.machine.del_vid(self.name, vid)
+        if ret != 0:
+            return ret
+        del self.vlan_interfaces[vid]
+

dsatest/bench/machine.py

@@ -80,3 +80,13 @@ def arp_get(self, address, interface):
                 return mac
 
         return None
+
+    def add_vid(self, interface, vid):
+        command = "ip link add {0}.{1} link {0} type vlan id {1}".format(interface, vid)
+        ret, _, _ = self.control.execute(command)
+        return ret
+
+    def del_vid(self, interface, vid):
+        command = "ip link del {0}.{1}".format(interface, vid)
+        ret, _, _, = self.control.execute(command)
+        return ret

dsatest/tests/helpers.py

@@ -53,3 +53,19 @@ def up_and_wait(up_interfaces, monitored=None, expand=True):
         timeout = timeout - 1
 
     raise RuntimeError("some interfaces did not up within alloted period")
+
+def get_address(offset, side, prefix_length=None):
+    if side == "host":
+        side = "1"
+    elif side == "target":
+        side = "2"
+    else:
+        raise ValueError("unexpected side")
+
+    address = "192.168.{}.{}".format(str(10 + offset), side)
+    if prefix_length:
+        address = "{}/{}".format(address, prefix_length)
+
+    return address
+
+

dsatest/tests/port/ping.py

@@ -2,48 +2,33 @@
 import unittest
 
 from dsatest.bench import bench
-from dsatest.tests.helpers import up_and_wait
+from dsatest.tests.helpers import up_and_wait, get_address
 
 @unittest.skipIf(not bench.links, "Empty link list")
 class TestPing(unittest.TestCase):
 
-    def get_address(self, offset, side, prefix_length=None):
-        if side == "host":
-            side = "1"
-        elif side == "target":
-            side = "2"
-        else:
-            raise ValueError("unexpected side")
-
-        address = "192.168.{}.{}".format(str(10 + offset), side)
-        if prefix_length:
-            address = "{}/{}".format(address, prefix_length)
-
-        return address
-
-
     def setUp(self):
         links = bench.links
 
         for i, link in enumerate(links):
             up_and_wait(link)
             link.host_if.flush_addresses()
-            link.host_if.add_address(self.get_address(i, "host", 24))
+            link.host_if.add_address(get_address(i, "host", 24))
             link.target_if.flush_addresses()
-            link.target_if.add_address(self.get_address(i, "target", 24))
+            link.target_if.add_address(get_address(i, "target", 24))
 
 
     def tearDown(self):
         links = bench.links
 
         for i, link in enumerate(links):
-            link.host_if.del_address(self.get_address(i, "host", 24))
+            link.host_if.del_address(get_address(i, "host", 24))
             link.host_if.down()
-            link.target_if.del_address(self.get_address(i, "target", 24))
+            link.target_if.del_address(get_address(i, "target", 24))
             link.target_if.down()
 
 
     def test_port_ping_all(self):
         for i, link in enumerate(bench.links):
-            addr = self.get_address(i, "target")
+            addr = get_address(i, "target")
             link.host_if.ping(addr, count=1, deadline=10)

dsatest/tests/port/vlan.py

@@ -0,0 +1,50 @@
+
+import unittest
+
+from dsatest.bench import bench
+from dsatest.tests.helpers import up_and_wait, get_address
+
+@unittest.skipIf(not bench.links, "Empty link list")
+class TestPingVlan(unittest.TestCase):
+
+    VID_START = 0
+    VID_END = 10
+
+    def setUp(self):
+        links = bench.links
+
+        for i, link in enumerate(links):
+            up_and_wait(link)
+            link.host_if.flush_addresses()
+            link.target_if.flush_addresses()
+            for vid in range(TestPingVlan.VID_START, TestPingVlan.VID_END):
+                link.host_if.add_vlan(vid)
+                link.target_if.add_vlan(vid)
+
+
+    def tearDown(self):
+        links = bench.links
+
+        for i, link in enumerate(links):
+            for vid in range(TestPingVlan.VID_START, TestPingVlan.VID_END):
+                link.target_if.del_vlan(vid)
+                link.host_if.del_vlan(vid)
+            link.host_if.down()
+            link.target_if.down()
+
+
+    def test_port_ping_vlan_all(self):
+        for i, link in enumerate(bench.links):
+            for vid in range(TestPingVlan.VID_START, TestPingVlan.VID_END):
+                host_vlan = link.host_if.vlan_interfaces[vid]
+                target_vlan = link.target_if.vlan_interfaces[vid]
+                host_vlan.add_address(get_address(vid, "host", 24))
+                host_vlan.up()
+                target_vlan.add_address(get_address(vid, "target", 24))
+                target_vlan.up()
+                addr = get_address(vid, "target")
+                host_vlan.ping(addr, count=1, deadline=10)
+                host_vlan.flush_addresses()
+                target_vlan.flush_addresses()
+                host_vlan.down()
+                target_vlan.down()