more refactoring
diff --git a/olt.py b/olt.py
index 2c36d58..2a68933 100644
--- a/olt.py
+++ b/olt.py
@@ -4,138 +4,20 @@
import logging
from __builtin__ import xrange
-from oftest import config
from oltbase import OltBaseTest
-import oftest.base_tests as base_tests
import oftest.packet as scapy
-
-
import ofp
import time
-import copy
from oftest.testutils import *
from IGMP import IGMPv3, IGMPv3gr, IGMP_TYPE_V3_MEMBERSHIP_REPORT, IGMP_V3_GR_TYPE_INCLUDE
-# These parameters can be altered from the command line using the -t or --test-params= options.
-# Example: -t 'onu_port=129;olt_port=288;device_type=pmc'
-#
-onu_port = test_param_get("onu_port", 130)
-onu_port2 = test_param_get("onu_port2", 131)
-olt_port = test_param_get("olt_port", 258)
-device_type = test_param_get("device_type", "normal") # options: "normal", "pmc", "cpqd"
-logging.info("device_type: %s" % device_type)
+from oltconstants import *
-def double_vlan_udp_packet(pktlen=100,
- eth_dst='00:01:02:03:04:05',
- eth_src='00:06:07:08:09:0a',
- dl_vlan_enable=False,
- c_vlan_vid=0,
- c_vlan_pcp=0,
- s_vlan_vid=0,
- s_vlan_pcp=0,
- ip_src='192.168.0.1',
- ip_dst='192.168.0.2',
- ip_tos=0,
- ip_ttl=64,
- udp_sport=1234,
- udp_dport=80,
- ip_ihl=None,
- ip_options=False,
- eth_type=0x8100
- ):
- """
- Return a double vlan tagged dataplane UDP packet
- Supports a few parameters:
- @param len Length of packet in bytes w/o CRC
- @param eth_dst Destination MAC
- @param eth_src Source MAC
- @param dl_vlan_enable True if the packet is with vlan, False otherwise
- @param c_vlan_vid CVLAN ID
- @param c_vlan_pcp CVLAN priority
- @param s_vlan_vid SVLAN ID
- @param s_vlan_pcp SVLAN priority
- @param ip_src IP source
- @param ip_dst IP destination
- @param ip_tos IP ToS
- @param ip_ttl IP TTL
- @param udp_dport UDP destination port
- @param udp_sport UDP source port
-
- Generates a simple UDP packet. Users shouldn't assume anything about
- this packet other than that it is a valid ethernet/IP/UDP frame.
- """
-
- if MINSIZE > pktlen:
- pktlen = MINSIZE
-
- # Note Dot1Q.id is really CFI
- if (dl_vlan_enable):
- pkt = scapy.Ether(dst=eth_dst, src=eth_src, type=eth_type) / \
- scapy.Dot1Q(prio=s_vlan_pcp, vlan=s_vlan_vid) / \
- scapy.Dot1Q(prio=c_vlan_pcp, vlan=c_vlan_vid) / \
- scapy.IP(src=ip_src, dst=ip_dst, tos=ip_tos, ttl=ip_ttl, ihl=ip_ihl) / \
- scapy.UDP(sport=udp_sport, dport=udp_dport)
- else:
- if not ip_options:
- pkt = scapy.Ether(dst=eth_dst, src=eth_src)/ \
- scapy.IP(src=ip_src, dst=ip_dst, tos=ip_tos, ttl=ip_ttl, ihl=ip_ihl)/ \
- scapy.UDP(sport=udp_sport, dport=udp_dport)
-
- else:
- pkt = scapy.Ether(dst=eth_dst, src=eth_src) / \
- scapy.IP(src=ip_src, dst=ip_dst, tos=ip_tos, ttl=ip_ttl, ihl=ip_ihl, options=ip_options) / \
- scapy.UDP(sport=udp_sport, dport=udp_dport)
-
- pkt = pkt / ("D" * (pktlen - len(pkt)))
-
- return pkt
-
-def testPacketIn(self, match, parsed_pkt):
- delete_all_flows(self.controller)
-
- pkt = str(parsed_pkt)
-
- for of_port in config["port_map"]:
- m = copy.deepcopy(match)
- m.oxm_list.append(ofp.oxm.in_port(of_port))
- request = ofp.message.flow_add(
- table_id=test_param_get("table", 0),
- cookie=42,
- match=m,
- instructions=[
- ofp.instruction.apply_actions(
- actions=[
- ofp.action.output(
- port=ofp.OFPP_CONTROLLER,
- max_len=ofp.OFPCML_NO_BUFFER)])],
- buffer_id=ofp.OFP_NO_BUFFER,
- priority=1000)
- logging.info("Inserting flow sending matching packets to controller")
- self.controller.message_send(request)
- do_barrier(self.controller)
-
- for of_port in config["port_map"]:
- logging.info("PacketInExact test, port %d", of_port)
- self.dataplane.send(of_port, pkt)
- verify_packet_in(self, pkt, of_port, ofp.OFPR_ACTION)
- verify_packets(self, pkt, [])
-
-
-def buildIgmp(payload):
- pkt = pkt = IGMPv3.fixup(scapy.Ether() / scapy.IP() / payload)
- if len(pkt) < 60:
- pad_len = 60 - len(pkt)
- pad = scapy.scapy.layers.l2.Padding()
- pad.load = '\x00' * pad_len
- pkt = pkt / pad
- return pkt
-
-
-class EapolPacketIn(base_tests.SimpleDataPlane):
+class EapolPacketIn(OltBaseTest):
"""Verify packet-ins are sent for EAPOL packets """
def runTest(self):
@@ -146,10 +28,10 @@
# Use ethertype 0x888e and multicast destination MAC address
pkt = simple_eth_packet(pktlen=60, eth_dst='01:00:5E:7F:FF:FF', eth_type=0x888e)
- testPacketIn(self, match, pkt)
+ self.testPacketIn(match, pkt)
-class ARPPacketIn(base_tests.SimpleDataPlane):
+class ARPPacketIn(OltBaseTest):
"""Verify packet-ins are sent for ARP packets """
def runTest(self):
@@ -161,10 +43,10 @@
# Use ethertype 0x0806
pkt = simple_eth_packet(eth_type=0x0806)
- testPacketIn(self, match, pkt)
+ self.testPacketIn(match, pkt)
-class IGMPPacketIn(base_tests.SimpleDataPlane):
+class IGMPPacketIn(OltBaseTest):
"""Verify packet-ins are sent for IGMP packets """
def runTest(self):
@@ -179,10 +61,10 @@
pkt = IGMPv3.fixup( scapy.Ether(src='00:00:00:00:be:ef') / scapy.IP() / igmp )
pkt = pkt / ("0" * (100 - len(pkt)))
- testPacketIn(self, match, pkt)
+ self.testPacketIn(match, pkt)
-class IGMPQueryPacketOut(base_tests.SimpleDataPlane):
+class IGMPQueryPacketOut(OltBaseTest):
"""Verify sending multicast membership queries down to onu_ports"""
def runTest(self):
@@ -206,7 +88,7 @@
verify_packet(self, pkt, onu_port)
-class TestMeter(base_tests.SimpleDataPlane):
+class TestMeter(OltBaseTest):
def runTest(self):
logging.info("Running Meter tests")
@@ -278,7 +160,7 @@
verify_no_errors(self.controller)
-class TestDuplicateMeter(base_tests.SimpleDataPlane):
+class TestDuplicateMeter(OltBaseTest):
def runTest(self):
logging.info("Running Duplicate Meter Test")
dropMeterBand = ofp.meter_band.drop(rate=500)
@@ -295,7 +177,7 @@
raise AssertionError("Incorrect error type: %s" % e.message)
-class VlanTest(base_tests.SimpleDataPlane):
+class VlanTest(OltBaseTest):
"""Verify the switch can push/pop a VLAN tag and forward out a port """
def runTest(self):
@@ -390,32 +272,10 @@
verify_packets(self, outPkt, [])
-def createAllGroupAdd(group_id, ports=[]):
- buckets = []
-
- for portNum in ports:
- buckets.append(ofp.common.bucket(watch_port=ofp.OFPP_ANY, watch_group=ofp.OFPG_ANY,
- actions=[ofp.action.pop_vlan(), ofp.action.output(port=portNum)]))
-
- group_add = ofp.message.group_add(group_type=ofp.OFPGT_ALL, group_id=group_id, buckets=buckets)
-
- return group_add
-def createAllGroupMod(group_id, ports=[]):
- buckets = []
- for portNum in ports:
- buckets.append(ofp.common.bucket(watch_port=ofp.OFPP_ANY, watch_group=ofp.OFPG_ANY,
- actions=[ofp.action.output(port=portNum)]))
-
- group_mod = ofp.message.group_mod(command=ofp.OFPGC_MODIFY, group_type=ofp.OFPGT_ALL, group_id=group_id,
- buckets=buckets)
-
- return group_mod
-
-
-class TestGroupAdd(base_tests.SimpleDataPlane):
+class TestGroupAdd(OltBaseTest):
def runTest(self):
logging.info("Running Group tests")
@@ -451,7 +311,7 @@
verify_no_errors(self.controller)
-class TestGroupMod(base_tests.SimpleDataPlane):
+class TestGroupMod(OltBaseTest):
def runTest(self):
logging.info("Running Group tests")
@@ -503,7 +363,7 @@
verify_no_errors(self.controller)
-class TestDuplicateGroup(base_tests.SimpleDataPlane):
+class TestDuplicateGroup(OltBaseTest):
def runTest(self):
logging.info("Running Group tests")
@@ -539,7 +399,7 @@
verify_no_errors(self.controller)
-class TestGroupAndFlow(base_tests.SimpleDataPlane):
+class TestGroupAndFlow(OltBaseTest):
def runTest(self):
logging.info("Running Group tests")
@@ -605,7 +465,7 @@
verify_no_errors(self.controller)
-class TestGroupForwarding(base_tests.SimpleDataPlane):
+class TestGroupForwarding(OltBaseTest):
def runTest(self):
logging.info("Running Group datapath forwarding tests")
@@ -676,7 +536,7 @@
verify_no_errors(self.controller)
-class TestGroupModForwarding(base_tests.SimpleDataPlane):
+class TestGroupModForwarding(OltBaseTest):
def runTest(self):
logging.info("Running datapath forwarding tests for group mod")
@@ -743,7 +603,7 @@
verify_packets(self, outPkt, [])
-class TransparentVlanTest(base_tests.SimpleDataPlane):
+class TransparentVlanTest(OltBaseTest):
def runTest(self):
logging.info("Running transparent vlan tests")
@@ -806,7 +666,7 @@
do_barrier(self.controller)
verify_no_errors(self.controller)
-class RemoveRuleTest(base_tests.SimpleDataPlane):
+class RemoveRuleTest(OltBaseTest):
def runTest(self):
logging.info("Testing Rule removal")
@@ -843,7 +703,7 @@
self.assertFalse(found, "Removed incorrect flow rule")
-class MultipleDoubleTaggingForwarding(base_tests.SimpleDataPlane):
+class MultipleDoubleTaggingForwarding(OltBaseTest):
def runTests(self):
logging.info("Testing multiple Q-in-Q rules")
@@ -873,7 +733,7 @@
verify_no_errors(self.controller)
-class TestCyclingDoubleVlan(base_tests.SimpleDataPlane):
+class TestCyclingDoubleVlan(OltBaseTest):
"""Cycle through vlans and test traffic flow"""
def runTest(self):
@@ -887,74 +747,12 @@
time.sleep(5)
testPacketFlow(self, ctag, stag)
-def processEapolRule(test, in_port, install = True):
- match = ofp.match()
- match.oxm_list.append(ofp.oxm.eth_type(0x888e))
- match.oxm_list.append(ofp.oxm.in_port(in_port))
- if install:
- request = ofp.message.flow_add(
- table_id=test_param_get("table", 0),
- cookie=42,
- match=match,
- instructions=[
- ofp.instruction.apply_actions(
- actions=[
- ofp.action.output(
- port=ofp.OFPP_CONTROLLER,
- max_len=ofp.OFPCML_NO_BUFFER)])],
- buffer_id=ofp.OFP_NO_BUFFER,
- priority=1000)
- else:
- request = ofp.message.flow_delete(
- table_id=test_param_get("table", 0),
- cookie=42,
- match=match,
- instructions=[
- ofp.instruction.apply_actions(
- actions=[
- ofp.action.output(
- port=ofp.OFPP_CONTROLLER,
- max_len=ofp.OFPCML_NO_BUFFER)])],
- buffer_id=ofp.OFP_NO_BUFFER,
- priority=1000)
- logging.info("%s flow sending matching packets to controller" % "Install" if install else "Remove")
- test.controller.message_send(request)
- do_barrier(test.controller)
-def testPacketFlow(test, c_vlan_id, s_vlan_id):
- incorrectTagPkt = simple_udp_packet(pktlen=100, dl_vlan_enable=True, vlan_vid=100, vlan_pcp=1)
- zeroTaggedPkt = simple_udp_packet(pktlen=100, dl_vlan_enable=True, vlan_vid=0, vlan_pcp=0)
- untaggedPkt = simple_udp_packet(pktlen=96)
- upstreamDoubleTaggedPkt = double_vlan_udp_packet(pktlen=104, dl_vlan_enable=True,
- c_vlan_vid=c_vlan_id,
- s_vlan_vid=s_vlan_id,
- c_vlan_pcp=0, s_vlan_pcp=0)
- logging.info("Testing s-tag %d, c-tag %d" % (s_vlan_id, c_vlan_id))
- # test upstream untagged packet got double tag at OLT
- test.dataplane.send(onu_port, str(zeroTaggedPkt))
- verify_packet(test, upstreamDoubleTaggedPkt, olt_port)
- # test downstream doubletagged packet got untagged at ONU
- test.dataplane.send(olt_port, str(upstreamDoubleTaggedPkt))
- if device_type == "pmc":
- verify_packet(test, zeroTaggedPkt, onu_port)
- else:
- verify_packet(test, untaggedPkt, onu_port)
- # test upstream doubletagged packet got dropped
- test.dataplane.send(onu_port, str(upstreamDoubleTaggedPkt))
- verify_no_packet(test, upstreamDoubleTaggedPkt, olt_port)
-
- # test downstream untagged packet got dropped at ONU
- test.dataplane.send(olt_port, str(untaggedPkt))
- verify_no_packet(test, untaggedPkt, onu_port)
-
- # test upstream icorrectly tagged packet; should get dropped
- test.dataplane.send(onu_port, str(incorrectTagPkt))
- verify_no_packet(test, upstreamDoubleTaggedPkt, olt_port)
diff --git a/oltbase.py b/oltbase.py
index b058efc..2860404 100644
--- a/oltbase.py
+++ b/oltbase.py
@@ -1,21 +1,113 @@
import oftest.base_tests as base_tests
-import oftest.packet as scapy
-from oftest import config
-
+from oftest import config
import ofp
-import time
-
from oftest.testutils import *
+from oltconstants import *
+import copy
-# These parameters can be altered from the command line using the -t or --test-params= options.
-# Example: -t 'onu_port=129;olt_port=288;device_type=pmc'
-#
-onu_port = test_param_get("onu_port", 130)
-onu_port2 = test_param_get("onu_port2", 131)
-olt_port = test_param_get("olt_port", 258)
-device_type = test_param_get("device_type", "normal") # options: "normal", "pmc", "cpqd"
+
class OltBaseTest(base_tests.SimpleDataPlane):
+ def testPacketIn(self, match, parsed_pkt):
+ delete_all_flows(self.controller)
+
+ pkt = str(parsed_pkt)
+
+ for of_port in config["port_map"]:
+ m = copy.deepcopy(match)
+ m.oxm_list.append(ofp.oxm.in_port(of_port))
+ request = ofp.message.flow_add(
+ table_id=test_param_get("table", 0),
+ cookie=42,
+ match=m,
+ instructions=[
+ ofp.instruction.apply_actions(
+ actions=[
+ ofp.action.output(
+ port=ofp.OFPP_CONTROLLER,
+ max_len=ofp.OFPCML_NO_BUFFER)])],
+ buffer_id=ofp.OFP_NO_BUFFER,
+ priority=1000)
+ logging.info("Inserting flow sending matching packets to controller")
+ self.controller.message_send(request)
+ do_barrier(self.controller)
+
+ for of_port in config["port_map"]:
+ logging.info("PacketInExact test, port %d", of_port)
+ self.dataplane.send(of_port, pkt)
+ verify_packet_in(self, pkt, of_port, ofp.OFPR_ACTION)
+ verify_packets(self, pkt, [])
+
+ def processEapolRule(self, in_port, install = True):
+ match = ofp.match()
+ match.oxm_list.append(ofp.oxm.eth_type(0x888e))
+ match.oxm_list.append(ofp.oxm.in_port(in_port))
+ if install:
+ request = ofp.message.flow_add(
+ table_id=test_param_get("table", 0),
+ cookie=42,
+ match=match,
+ instructions=[
+ ofp.instruction.apply_actions(
+ actions=[
+ ofp.action.output(
+ port=ofp.OFPP_CONTROLLER,
+ max_len=ofp.OFPCML_NO_BUFFER)])],
+ buffer_id=ofp.OFP_NO_BUFFER,
+ priority=1000)
+ else:
+ request = ofp.message.flow_delete(
+ table_id=test_param_get("table", 0),
+ cookie=42,
+ match=match,
+ instructions=[
+ ofp.instruction.apply_actions(
+ actions=[
+ ofp.action.output(
+ port=ofp.OFPP_CONTROLLER,
+ max_len=ofp.OFPCML_NO_BUFFER)])],
+ buffer_id=ofp.OFP_NO_BUFFER,
+ priority=1000)
+ logging.info("%s flow sending matching packets to controller" % "Install" if install else "Remove")
+ self.controller.message_send(request)
+ do_barrier(self.controller)
+
+ def testPacketFlow(self, c_vlan_id, s_vlan_id):
+
+ incorrectTagPkt = simple_udp_packet(pktlen=100, dl_vlan_enable=True, vlan_vid=100, vlan_pcp=1)
+ zeroTaggedPkt = simple_udp_packet(pktlen=100, dl_vlan_enable=True, vlan_vid=0, vlan_pcp=0)
+ untaggedPkt = simple_udp_packet(pktlen=96)
+
+ upstreamDoubleTaggedPkt = double_vlan_udp_packet(pktlen=104, dl_vlan_enable=True,
+ c_vlan_vid=c_vlan_id,
+ s_vlan_vid=s_vlan_id,
+ c_vlan_pcp=0, s_vlan_pcp=0)
+
+ logging.info("Testing s-tag %d, c-tag %d" % (s_vlan_id, c_vlan_id))
+
+ # test upstream untagged packet got double tag at OLT
+ self.dataplane.send(onu_port, str(zeroTaggedPkt))
+ verify_packet(test, upstreamDoubleTaggedPkt, olt_port)
+
+ # test downstream doubletagged packet got untagged at ONU
+ self.dataplane.send(olt_port, str(upstreamDoubleTaggedPkt))
+ if device_type == "pmc":
+ verify_packet(self, zeroTaggedPkt, onu_port)
+ else:
+ verify_packet(self, untaggedPkt, onu_port)
+
+ # test upstream doubletagged packet got dropped
+ self.dataplane.send(onu_port, str(upstreamDoubleTaggedPkt))
+ verify_no_packet(self, upstreamDoubleTaggedPkt, olt_port)
+
+ # test downstream untagged packet got dropped at ONU
+ self.dataplane.send(olt_port, str(untaggedPkt))
+ verify_no_packet(self, untaggedPkt, onu_port)
+
+ # test upstream icorrectly tagged packet; should get dropped
+ self.dataplane.send(onu_port, str(incorrectTagPkt))
+ verify_no_packet(self, upstreamDoubleTaggedPkt, olt_port)
+
def installDoubleTaggingRules(self, s_vlan_id, c_vlan_id, cookie=42):
# upstream flow rule
diff --git a/oltconstants.py b/oltconstants.py
new file mode 100644
index 0000000..3e6d03c
--- /dev/null
+++ b/oltconstants.py
@@ -0,0 +1,112 @@
+from oftest.testutils import *
+import oftest.packet as scapy
+from oftest import config
+
+import ofp
+
+# These parameters can be altered from the command line using the -t or --test-params= options.
+# Example: -t 'onu_port=129;olt_port=288;device_type=pmc'
+#
+onu_port = test_param_get("onu_port", 130)
+onu_port2 = test_param_get("onu_port2", 131)
+olt_port = test_param_get("olt_port", 258)
+device_type = test_param_get("device_type", "normal") # options: "normal", "pmc", "cpqd"
+
+
+def createAllGroupAdd(group_id, ports=[]):
+ buckets = []
+
+ for portNum in ports:
+ buckets.append(ofp.common.bucket(watch_port=ofp.OFPP_ANY, watch_group=ofp.OFPG_ANY,
+ actions=[ofp.action.pop_vlan(), ofp.action.output(port=portNum)]))
+
+ group_add = ofp.message.group_add(group_type=ofp.OFPGT_ALL, group_id=group_id, buckets=buckets)
+
+ return group_add
+
+
+def createAllGroupMod(group_id, ports=[]):
+ buckets = []
+
+ for portNum in ports:
+ buckets.append(ofp.common.bucket(watch_port=ofp.OFPP_ANY, watch_group=ofp.OFPG_ANY,
+ actions=[ofp.action.output(port=portNum)]))
+
+ group_mod = ofp.message.group_mod(command=ofp.OFPGC_MODIFY, group_type=ofp.OFPGT_ALL, group_id=group_id,
+ buckets=buckets)
+
+ return group_mod
+
+def buildIgmp(payload):
+ pkt = pkt = IGMPv3.fixup(scapy.Ether() / scapy.IP() / payload)
+ if len(pkt) < 60:
+ pad_len = 60 - len(pkt)
+ pad = scapy.scapy.layers.l2.Padding()
+ pad.load = '\x00' * pad_len
+ pkt = pkt / pad
+ return pkt
+
+def double_vlan_udp_packet(pktlen=100,
+ eth_dst='00:01:02:03:04:05',
+ eth_src='00:06:07:08:09:0a',
+ dl_vlan_enable=False,
+ c_vlan_vid=0,
+ c_vlan_pcp=0,
+ s_vlan_vid=0,
+ s_vlan_pcp=0,
+ ip_src='192.168.0.1',
+ ip_dst='192.168.0.2',
+ ip_tos=0,
+ ip_ttl=64,
+ udp_sport=1234,
+ udp_dport=80,
+ ip_ihl=None,
+ ip_options=False,
+ eth_type=0x8100
+ ):
+ """
+ Return a double vlan tagged dataplane UDP packet
+ Supports a few parameters:
+ @param len Length of packet in bytes w/o CRC
+ @param eth_dst Destination MAC
+ @param eth_src Source MAC
+ @param dl_vlan_enable True if the packet is with vlan, False otherwise
+ @param c_vlan_vid CVLAN ID
+ @param c_vlan_pcp CVLAN priority
+ @param s_vlan_vid SVLAN ID
+ @param s_vlan_pcp SVLAN priority
+ @param ip_src IP source
+ @param ip_dst IP destination
+ @param ip_tos IP ToS
+ @param ip_ttl IP TTL
+ @param udp_dport UDP destination port
+ @param udp_sport UDP source port
+
+ Generates a simple UDP packet. Users shouldn't assume anything about
+ this packet other than that it is a valid ethernet/IP/UDP frame.
+ """
+
+ if MINSIZE > pktlen:
+ pktlen = MINSIZE
+
+ # Note Dot1Q.id is really CFI
+ if (dl_vlan_enable):
+ pkt = scapy.Ether(dst=eth_dst, src=eth_src, type=eth_type) / \
+ scapy.Dot1Q(prio=s_vlan_pcp, vlan=s_vlan_vid) / \
+ scapy.Dot1Q(prio=c_vlan_pcp, vlan=c_vlan_vid) / \
+ scapy.IP(src=ip_src, dst=ip_dst, tos=ip_tos, ttl=ip_ttl, ihl=ip_ihl) / \
+ scapy.UDP(sport=udp_sport, dport=udp_dport)
+ else:
+ if not ip_options:
+ pkt = scapy.Ether(dst=eth_dst, src=eth_src)/ \
+ scapy.IP(src=ip_src, dst=ip_dst, tos=ip_tos, ttl=ip_ttl, ihl=ip_ihl)/ \
+ scapy.UDP(sport=udp_sport, dport=udp_dport)
+
+ else:
+ pkt = scapy.Ether(dst=eth_dst, src=eth_src) / \
+ scapy.IP(src=ip_src, dst=ip_dst, tos=ip_tos, ttl=ip_ttl, ihl=ip_ihl, options=ip_options) / \
+ scapy.UDP(sport=udp_sport, dport=udp_dport)
+
+ pkt = pkt / ("D" * (pktlen - len(pkt)))
+
+ return pkt
\ No newline at end of file