Implementation of many modify tests
Use --test-params option instead of old --param option.
Implemented pkt_action_setup that does the core work of
generating in/out pkts and action lists. Use this for a bunch
of modify test cases.
diff --git a/tests/pktact.py b/tests/pktact.py
index be7d7ed..7c3fcbb 100644
--- a/tests/pktact.py
+++ b/tests/pktact.py
@@ -624,9 +624,7 @@
"""
def runTest(self):
- vid = TEST_VID_DEFAULT
- if pa_config["param"] is not None:
- vid = pa_config["param"]
+ vid = test_param_get(self.config, 'vid', default=TEST_VID_DEFAULT)
flow_match_test(self, pa_port_map, dl_vlan=vid)
class ExactMatchTaggedMany(BaseMatchCase):
@@ -665,9 +663,7 @@
SingleWildcardMatch with tagged packets
"""
def runTest(self):
- vid = TEST_VID_DEFAULT
- if pa_config["param"] is not None:
- vid = pa_config["param"]
+ vid = test_param_get(self.config, 'vid', default=TEST_VID_DEFAULT)
for wc in WILDCARD_VALUES:
flow_match_test(self, pa_port_map, wildcards=wc, dl_vlan=vid,
max_test=10)
@@ -693,9 +689,7 @@
Match one field with tagged packets
"""
def runTest(self):
- vid = TEST_VID_DEFAULT
- if pa_config["param"] is not None:
- vid = pa_config["param"]
+ vid = test_param_get(self.config, 'vid', default=TEST_VID_DEFAULT)
for wc in WILDCARD_VALUES:
all_exp_one_wildcard = ofp.OFPFW_ALL ^ wc
flow_match_test(self, pa_port_map, wildcards=all_exp_one_wildcard,
@@ -720,9 +714,7 @@
AllWildcardMatch with tagged packets
"""
def runTest(self):
- vid = TEST_VID_DEFAULT
- if pa_config["param"] is not None:
- vid = pa_config["param"]
+ vid = test_param_get(self.config, 'vid', default=TEST_VID_DEFAULT)
flow_match_test(self, pa_port_map, wildcards=ofp.OFPFW_ALL,
dl_vlan=vid)
@@ -766,10 +758,7 @@
Just send a packet thru the switch
"""
def runTest(self):
- vid = TEST_VID_DEFAULT
- if pa_config["param"] is not None:
- vid = pa_config["param"]
- print "Param is " + str(pa_config["param"])
+ vid = test_param_get(self.config, 'vid', default=TEST_VID_DEFAULT)
pkt = simple_tcp_packet(dl_vlan_enable=True, dl_vlan=vid)
of_ports = pa_port_map.keys()
of_ports.sort()
@@ -782,11 +771,14 @@
test_prio["PacketOnlyTagged"] = -1
class ModifyVID(BaseMatchCase):
+ """
+ Modify the VLAN ID in the VLAN tag of a tagged packet
+ """
def runTest(self):
old_vid = 2
new_vid = 3
sup_acts = supported_actions_get(self)
- if not(sup_acts & 1<<ofp.OFPAT_SET_VLAN_VID):
+ if not (sup_acts & 1 << ofp.OFPAT_SET_VLAN_VID):
pa_logger.info("Skipping modify VLAN tag test")
return
@@ -799,10 +791,13 @@
action_list=[vid_act])
class StripVLANTag(BaseMatchCase):
+ """
+ Strip the VLAN tag from a tagged packet
+ """
def runTest(self):
old_vid = 2
sup_acts = supported_actions_get(self)
- if not(sup_acts & 1<<ofp.OFPAT_STRIP_VLAN):
+ if not (sup_acts & 1 << ofp.OFPAT_STRIP_VLAN):
pa_logger.info("Skipping strip VLAN tag test")
return
@@ -816,42 +811,159 @@
flow_match_test(self, pa_port_map, pkt=pkt, exp_pkt=exp_pkt,
action_list=[vid_act])
+def init_pkt_args():
+ """
+ Pass back a dictionary with default packet arguments
+ """
+ args = {}
+ args["dl_src"] = '00:23:45:67:89:AB'
+
+ dl_vlan_enable=False
+ dl_vlan=-1
+ if pa_config["test-params"]["vid"]:
+ dl_vlan_enable=True
+ dl_vlan = pa_config["test-params"]["vid"]
+
+# Unpack operator is ** on a dictionary
+
+ return args
+
class ModifyL2Src(BaseMatchCase):
+ """
+ Modify the source MAC address (TP1)
+ """
def runTest(self):
- pa_logger("To be implemented")
+ sup_acts = supported_actions_get(self)
+ if not (sup_acts & 1 << ofp.OFPAT_SET_DL_SRC):
+ pa_logger.info("Skipping ModifyL2Src test")
+ return
+
+ (pkt, exp_pkt, acts) = pkt_action_setup(self, mod_fields=['dl_src'],
+ check_test_params=True)
+ flow_match_test(self, pa_port_map, pkt=pkt, exp_pkt=exp_pkt,
+ action_list=acts, max_test=2)
class ModifyL2Dst(BaseMatchCase):
+ """
+ Modify the dest MAC address (TP1)
+ """
def runTest(self):
- pa_logger("To be implemented")
+ sup_acts = supported_actions_get(self)
+ if not (sup_acts & 1 << ofp.OFPAT_SET_DL_DST):
+ pa_logger.info("Skipping ModifyL2Dst test")
+ return
+
+ (pkt, exp_pkt, acts) = pkt_action_setup(self, mod_fields=['dl_dst'],
+ check_test_params=True)
+ flow_match_test(self, pa_port_map, pkt=pkt, exp_pkt=exp_pkt,
+ action_list=acts, max_test=2)
class ModifyL3Src(BaseMatchCase):
+ """
+ Modify the source IP address of an IP packet (TP1)
+ """
def runTest(self):
- pa_logger("To be implemented")
+ sup_acts = supported_actions_get(self)
+ if not (sup_acts & 1 << ofp.OFPAT_SET_NW_SRC):
+ pa_logger.info("Skipping ModifyL3Src test")
+ return
+
+ (pkt, exp_pkt, acts) = pkt_action_setup(self, mod_fields=['ip_src'],
+ check_test_params=True)
+ flow_match_test(self, pa_port_map, pkt=pkt, exp_pkt=exp_pkt,
+ action_list=acts, max_test=2)
class ModifyL3Dst(BaseMatchCase):
+ """
+ Modify the dest IP address of an IP packet (TP1)
+ """
def runTest(self):
- pa_logger("To be implemented")
+ sup_acts = supported_actions_get(self)
+ if not (sup_acts & 1 << ofp.OFPAT_SET_NW_DST):
+ pa_logger.info("Skipping ModifyL3Dst test")
+ return
+
+ (pkt, exp_pkt, acts) = pkt_action_setup(self, mod_fields=['ip_dst'],
+ check_test_params=True)
+ flow_match_test(self, pa_port_map, pkt=pkt, exp_pkt=exp_pkt,
+ action_list=acts, max_test=2)
class ModifyL4Src(BaseMatchCase):
+ """
+ Modify the source TCP port of a TCP packet (TP1)
+ """
def runTest(self):
- pa_logger("To be implemented")
+ sup_acts = supported_actions_get(self)
+ if not (sup_acts & 1 << ofp.OFPAT_SET_TP_SRC):
+ pa_logger.info("Skipping ModifyL4Src test")
+ return
+
+ (pkt, exp_pkt, acts) = pkt_action_setup(self, mod_fields=['tcp_sport'],
+ check_test_params=True)
+ flow_match_test(self, pa_port_map, pkt=pkt, exp_pkt=exp_pkt,
+ action_list=acts, max_test=2)
class ModifyL4Dst(BaseMatchCase):
+ """
+ Modify the dest TCP port of a TCP packet (TP1)
+ """
def runTest(self):
- pa_logger("To be implemented")
+ sup_acts = supported_actions_get(self)
+ if not (sup_acts & 1 << ofp.OFPAT_SET_TP_DST):
+ pa_logger.info("Skipping ModifyL4Dst test")
+ return
+
+ (pkt, exp_pkt, acts) = pkt_action_setup(self, mod_fields=['tcp_dport'],
+ check_test_params=True)
+ flow_match_test(self, pa_port_map, pkt=pkt, exp_pkt=exp_pkt,
+ action_list=acts, max_test=2)
class ModifyTOS(BaseMatchCase):
+ """
+ Modify the IP type of service of an IP packet (TP1)
+ """
def runTest(self):
- pa_logger("To be implemented")
+ sup_acts = supported_actions_get(self)
+ if not (sup_acts & 1 << ofp.OFPAT_SET_NW_TOS):
+ pa_logger.info("Skipping ModifyTOS test")
+ return
-test_prio["ModifyL2Src"] = -1
-test_prio["ModifyL2Dst"] = -1
-test_prio["ModifyL3Src"] = -1
-test_prio["ModifyL3Dst"] = -1
-test_prio["ModifyL4Src"] = -1
-test_prio["ModifyL4Dst"] = -1
-test_prio["ModifyTOS"] = -1
+ (pkt, exp_pkt, acts) = pkt_action_setup(self, mod_fields=['ip_tos'],
+ check_test_params=True)
+ flow_match_test(self, pa_port_map, pkt=pkt, exp_pkt=exp_pkt,
+ action_list=acts, max_test=2)
+#@todo Need to implement tagged versions of the above tests
+#
+#@todo Implement a test case that strips tag 2, adds tag 3
+# and modifies tag 4 to tag 5. Then verify (in addition) that
+# tag 6 does not get modified.
+
+class MixedVLAN(BaseMatchCase):
+ """
+ Test mixture of VLAN tag actions
+
+ Strip tag 2 on port 1, send to port 2
+ Add tag 3 on port 1, send to port 2
+ Modify tag 4 to 5 on port 1, send to port 2
+ All other traffic from port 1, send to port 3
+ All traffic from port 2 sent to port 4
+ Use exact matches with different packets for all mods
+ Verify the following: (port, vid)
+ (port 1, vid 2) => VLAN tag stripped, out port 2
+ (port 1, no tag) => tagged packet w/ vid 2 out port 2
+ (port 1, vid 4) => tagged packet w/ vid 5 out port 2
+ (port 1, vid 5) => tagged packet w/ vid 5 out port 2
+ (port 1, vid 6) => tagged packet w/ vid 6 out port 2
+ (port 2, no tag) => untagged packet out port 4
+ (port 2, vid 2-6) => unmodified packet out port 4
+
+ Variation: Might try sending VID 5 to port 3 and check.
+ If only VID 5 distinguishes pkt, this will fail on some platforms
+ """
+
+test_prio["MixedVLAN"] = -1
+
def supported_actions_get(parent, use_cache=True):
"""
Get the bitmap of supported actions from the switch