[python3]migrate dhcp relay ptftests scripts from python2 to python3

ansible/roles/test/files/ptftests/dhcp_relay_test.py → ansible/roles/test/files/ptftests/py3/dhcp_relay_test.py

@@ -1,6 +1,7 @@
 import ast
 import struct
 import ipaddress
+import binascii
 
 # Packet Test Framework imports
 import ptf
@@ -17,7 +18,7 @@
 # ip_addr should be passed as a dot-decimal string
 # Return value is also a dot-decimal string
 def incrementIpAddress(ip_addr, by=1):
-    new_addr = ipaddress.ip_address(unicode(ip_addr))
+    new_addr = ipaddress.ip_address(str(ip_addr))
     new_addr = new_addr + by
     return str(new_addr)
 
@@ -102,7 +103,7 @@ def setUp(self):
         self.hostname = self.test_params['hostname']
         self.verified_option82 = False
 
-        if self.test_params.has_key('other_client_port'):
+        if 'other_client_port' in self.test_params:
             self.other_client_port = ast.literal_eval(self.test_params['other_client_port'])
 
         # These are the interfaces we are injected into that link to out leaf switches
@@ -142,7 +143,7 @@ def setUp(self):
         # Our circuit_id string is of the form "hostname:portname"
         circuit_id_string = self.hostname + ":" + self.client_iface_alias
         self.option82 = struct.pack('BB', 1, len(circuit_id_string))
-        self.option82 += circuit_id_string
+        self.option82 += circuit_id_string.encode('utf-8')
 
         # remote_id is stored as suboption 2 of option 82.
         # It consists of the following:
@@ -152,15 +153,15 @@ def setUp(self):
         # Our remote_id string simply consists of the MAC address of the port that received the request
         remote_id_string = self.relay_iface_mac
         self.option82 += struct.pack('BB', 2, len(remote_id_string))
-        self.option82 += remote_id_string
+        self.option82 += remote_id_string.encode('utf-8')
 
         # In 'dual' testing mode, vlan ip is stored as suboption 5 of option 82.
         # It consists of the following:
         #  Byte 0: Suboption number, always set to 5
         #  Byte 1: Length of suboption data in bytes, always set to 4 (ipv4 addr has 4 bytes)
         #  Bytes 2+: vlan ip addr
         if self.dual_tor:
-            link_selection = ''.join([chr(int(byte)) for byte in self.relay_iface_ip.split('.')])
+            link_selection = bytes(list(map(int, self.relay_iface_ip.split('.'))))
             self.option82 += struct.pack('BB', 5, 4)
             self.option82 += link_selection
 
@@ -194,7 +195,8 @@ def create_dhcp_discover_packet(self, dst_mac=BROADCAST_MAC, src_port=DHCP_CLIEN
         return discover_packet
 
     def create_dhcp_discover_relayed_packet(self):
-        my_chaddr = ''.join([chr(int(octet, 16)) for octet in self.client_mac.split(':')])
+        my_chaddr = binascii.unhexlify(self.client_mac.replace(':', ''))
+        my_chaddr += b'\x00\x00\x00\x00\x00\x00'
 
         # Relay modifies the DHCPDISCOVER message in the following ways:
         #  1.) Increments the hops count in the DHCP header
@@ -227,7 +229,7 @@ def create_dhcp_discover_relayed_packet(self):
                     giaddr=self.relay_iface_ip if not self.dual_tor else self.switch_loopback_ip,
                     chaddr=my_chaddr)
         bootp /= scapy.DHCP(options=[('message-type', 'discover'),
-                    ('relay_agent_Information', self.option82),
+                    (82, self.option82),
                     ('end')])
 
         # If our bootp layer is too small, pad it
@@ -253,7 +255,8 @@ def create_dhcp_offer_packet(self):
                     set_broadcast_bit=True)
 
     def create_dhcp_offer_relayed_packet(self):
-        my_chaddr = ''.join([chr(int(octet, 16)) for octet in self.client_mac.split(':')])
+        my_chaddr = binascii.unhexlify(self.client_mac.replace(':', ''))
+        my_chaddr += b'\x00\x00\x00\x00\x00\x00'
 
         # Relay modifies the DHCPOFFER message in the following ways:
         #  1.) Replaces the source MAC with the MAC of the interface it received it on
@@ -310,7 +313,8 @@ def create_dhcp_request_packet(self, dst_mac=BROADCAST_MAC, src_port=DHCP_CLIENT
         return request_packet
 
     def create_dhcp_request_relayed_packet(self):
-        my_chaddr = ''.join([chr(int(octet, 16)) for octet in self.client_mac.split(':')])
+        my_chaddr = binascii.unhexlify(self.client_mac.replace(':', ''))
+        my_chaddr += b'\x00\x00\x00\x00\x00\x00'
 
         # Here, the actual destination MAC should be the MAC of the leaf the relay
         # forwards through and the destination IP should be the IP of the DHCP server
@@ -338,7 +342,7 @@ def create_dhcp_request_relayed_packet(self):
         bootp /= scapy.DHCP(options=[('message-type', 'request'),
                     ('requested_addr', self.client_ip),
                     ('server_id', self.server_ip),
-                    ('relay_agent_Information', self.option82),
+                    (82, self.option82),
                     ('end')])
 
         # If our bootp layer is too small, pad it
@@ -364,7 +368,8 @@ def create_dhcp_ack_packet(self):
                     set_broadcast_bit=True)
 
     def create_dhcp_ack_relayed_packet(self):
-        my_chaddr = ''.join([chr(int(octet, 16)) for octet in self.client_mac.split(':')])
+        my_chaddr = binascii.unhexlify(self.client_mac.replace(':', ''))
+        my_chaddr += b'\x00\x00\x00\x00\x00\x00'
 
         # Relay modifies the DHCPACK message in the following ways:
         #  1.) Replaces the source MAC with the MAC of the interface it received it on
@@ -426,7 +431,7 @@ def pkt_callback(self, pkt):
                 self.verified_option82 = False
                 pkt_options = ''
                 for option in pkt.getlayer(scapy2.DHCP).options:
-                    if option[0] == 'relay_agent_Information':
+                    if option[0] == 'relay_agent_information':
                         pkt_options = option[1]
                         break
                 if self.option82 in pkt_options:
@@ -504,10 +509,6 @@ def verify_offer_received(self):
         masked_offer.set_do_not_care_scapy(scapy.BOOTP, "sname")
         masked_offer.set_do_not_care_scapy(scapy.BOOTP, "file")
 
-        masked_offer.set_do_not_care_scapy(scapy.DHCP, "lease_time")
-
-        #masked_offer.set_do_not_care_scapy(scapy.PADDING, "load")
-
         # NOTE: verify_packet() will fail for us via an assert, so no need to check a return value here
         testutils.verify_packet(self, masked_offer, self.client_port_index)
 
@@ -582,8 +583,6 @@ def verify_ack_received(self):
         masked_ack.set_do_not_care_scapy(scapy.BOOTP, "sname")
         masked_ack.set_do_not_care_scapy(scapy.BOOTP, "file")
 
-        masked_ack.set_do_not_care_scapy(scapy.DHCP, "lease_time")
-
         # NOTE: verify_packet() will fail for us via an assert, so no need to check a return value here
         testutils.verify_packet(self, masked_ack, self.client_port_index)
 
@@ -683,7 +682,6 @@ def runTest(self):
         self.assertTrue(self.verified_option82,"Failed: Verifying option 82")
 
         ## Below verification will be done only when client port is set in ptf_runner
-        if self.test_params.has_key('other_client_port'):
+        if 'other_client_port' in self.test_params:
             self.verify_dhcp_relay_pkt_on_other_client_port_with_no_padding(self.dest_mac_address, self.client_udp_src_port)
             self.verify_dhcp_relay_pkt_on_server_port_with_no_padding(self.dest_mac_address, self.client_udp_src_port)
-

ansible/roles/test/files/ptftests/dhcpv6_counter_test.py → ansible/roles/test/files/ptftests/py3/dhcpv6_counter_test.py

@@ -1,13 +1,24 @@
 import ast
 import subprocess
-
+import scapy
 # Packet Test Framework imports
 import ptf
 import ptf.testutils as testutils
 from ptf import config
 from ptf.base_tests import BaseTest
 
 IPv6 = scapy.layers.inet6.IPv6
+DHCP6_Solicit = scapy.layers.dhcp6.DHCP6_Solicit
+DHCP6_Request = scapy.layers.dhcp6.DHCP6_Request
+DHCP6_Confirm = scapy.layers.dhcp6.DHCP6_Confirm
+DHCP6_Renew = scapy.layers.dhcp6.DHCP6_Renew
+DHCP6_Rebind = scapy.layers.dhcp6.DHCP6_Rebind
+DHCP6_Release = scapy.layers.dhcp6.DHCP6_Release
+DHCP6_Decline = scapy.layers.dhcp6.DHCP6_Decline
+DHCP6_Advertise = scapy.layers.dhcp6.DHCP6_Advertise
+DHCP6_Reply = scapy.layers.dhcp6.DHCP6_Reply
+DHCP6_RelayReply = scapy.layers.dhcp6.DHCP6_RelayReply
+DHCP6OptRelayMsg = scapy.layers.dhcp6.DHCP6OptRelayMsg
 
 class DataplaneBaseTest(BaseTest):
     def __init__(self):
@@ -97,8 +108,7 @@ def create_server_packet(self, message):
         packet /= IPv6(src=self.server_ip, dst=self.relay_iface_ip)
         packet /= UDP(sport=self.DHCP_SERVER_PORT, dport=self.DHCP_SERVER_PORT)
         packet /= DHCP6_RelayReply(msgtype=13, linkaddr=self.vlan_ip, peeraddr=self.client_link_local)
-        packet /= DHCP6OptRelayMsg()
-        packet /= message(trid=12345)
+        packet /= DHCP6OptRelayMsg(message=[message(trid=12345)])
 
         return packet
 

ansible/roles/test/files/ptftests/dhcpv6_relay_test.py → ansible/roles/test/files/ptftests/py3/dhcpv6_relay_test.py

@@ -1,16 +1,27 @@
 import ast
 import subprocess
-
+import scapy
 # Packet Test Framework imports
 import ptf
 import ptf.packet as packet
 import ptf.testutils as testutils
 from ptf import config
 from ptf.base_tests import BaseTest
 from ptf.mask import Mask
+from scapy.fields import MACField, ShortEnumField, FieldLenField, ShortField
+from scapy.data import ETHER_ANY
+from scapy.layers.dhcp6 import _DHCP6OptGuessPayload
 
 IPv6 = scapy.layers.inet6.IPv6
-
+DHCP6_Solicit = scapy.layers.dhcp6.DHCP6_Solicit
+DHCP6_RelayForward = scapy.layers.dhcp6.DHCP6_RelayForward
+DHCP6_Request = scapy.layers.dhcp6.DHCP6_Request
+DHCP6_Advertise = scapy.layers.dhcp6.DHCP6_Advertise
+DHCP6_Reply = scapy.layers.dhcp6.DHCP6_Reply
+DHCP6_RelayReply = scapy.layers.dhcp6.DHCP6_RelayReply
+DHCP6OptRelayMsg = scapy.layers.dhcp6.DHCP6OptRelayMsg
+DHCP6OptUnknown = scapy.layers.dhcp6.DHCP6OptUnknown
+
 class DataplaneBaseTest(BaseTest):
     def __init__(self):
         BaseTest.__init__(self)
@@ -51,25 +62,6 @@ def tearDown(self):
 
 dhcp6opts = {79: "OPTION_CLIENT_LINKLAYER_ADDR",  # RFC6939
              }
-class _DHCP6OptGuessPayload(Packet):
-    @staticmethod
-    def _just_guess_payload_class(cls, payload):
-        # try to guess what option is in the payload
-        if len(payload) <= 2:
-            return conf.raw_layer
-        opt = struct.unpack("!H", payload[:2])[0]
-        clsname = dhcp6opts_by_code.get(opt, None)
-        if clsname is None:
-            return cls
-        return get_cls(clsname, cls)
-
-    def guess_payload_class(self, payload):
-        # this method is used in case of all derived classes
-        # from _DHCP6OptGuessPayload in this file
-        return _DHCP6OptGuessPayload._just_guess_payload_class(
-            DHCP6OptUnknown,
-            payload
-        )
 
 class _LLAddrField(MACField):
     pass
@@ -159,8 +151,7 @@ def create_dhcp_solicit_relay_forward_packet(self):
         solicit_relay_forward_packet /= UDP(sport=self.DHCP_SERVER_PORT, dport=self.DHCP_SERVER_PORT)
         solicit_relay_forward_packet /= DHCP6_RelayForward(msgtype=12, linkaddr=self.vlan_ip, peeraddr=self.client_link_local)
         solicit_relay_forward_packet /= DHCP6OptClientLinkLayerAddr()
-        solicit_relay_forward_packet /= DHCP6OptRelayMsg()
-        solicit_relay_forward_packet /= DHCP6_Solicit(trid=12345)
+        solicit_relay_forward_packet /= DHCP6OptRelayMsg(message=[DHCP6_Solicit(trid=12345)])
 
         return solicit_relay_forward_packet
 
@@ -177,9 +168,7 @@ def create_dhcp_advertise_relay_reply_packet(self):
         advertise_relay_reply_packet /= IPv6(src=self.server_ip, dst=self.relay_iface_ip)
         advertise_relay_reply_packet /= UDP(sport=self.DHCP_SERVER_PORT, dport=self.DHCP_SERVER_PORT)
         advertise_relay_reply_packet /= DHCP6_RelayReply(msgtype=13, linkaddr=self.vlan_ip, peeraddr=self.client_link_local)
-        advertise_relay_reply_packet /= DHCP6OptRelayMsg()
-        advertise_relay_reply_packet /= DHCP6_Advertise(trid=12345)
-
+        advertise_relay_reply_packet /= DHCP6OptRelayMsg(message=[DHCP6_Advertise(trid=12345)])
         return advertise_relay_reply_packet
 
     def create_dhcp_request_packet(self):
@@ -196,8 +185,7 @@ def create_dhcp_request_relay_forward_packet(self):
         request_relay_forward_packet /= UDP(sport=self.DHCP_SERVER_PORT, dport=self.DHCP_SERVER_PORT)
         request_relay_forward_packet /= DHCP6_RelayForward(msgtype=12, linkaddr=self.vlan_ip, peeraddr=self.client_link_local)
         request_relay_forward_packet /= DHCP6OptClientLinkLayerAddr()
-        request_relay_forward_packet /= DHCP6OptRelayMsg()
-        request_relay_forward_packet /= DHCP6_Request(trid=12345)
+        request_relay_forward_packet /= DHCP6OptRelayMsg(message=[DHCP6_Request(trid=12345)])
 
         return request_relay_forward_packet
 
@@ -214,8 +202,7 @@ def create_dhcp_reply_relay_reply_packet(self):
         reply_relay_reply_packet /= IPv6(src=self.server_ip, dst=self.relay_iface_ip)
         reply_relay_reply_packet /= UDP(sport=self.DHCP_SERVER_PORT, dport=self.DHCP_SERVER_PORT)
         reply_relay_reply_packet /= DHCP6_RelayReply(msgtype=13, linkaddr=self.vlan_ip, peeraddr=self.client_link_local)
-        reply_relay_reply_packet /= DHCP6OptRelayMsg()
-        reply_relay_reply_packet /= DHCP6_Reply(trid=12345)
+        reply_relay_reply_packet /= DHCP6OptRelayMsg(message=[DHCP6_Reply(trid=12345)])
 
         return reply_relay_reply_packet
 
@@ -224,20 +211,18 @@ def create_dhcp_relay_forward_packet(self):
         relay_forward_packet /= IPv6(src=self.client_link_local, dst=self.BROADCAST_IP)
         relay_forward_packet /= UDP(sport=self.DHCP_CLIENT_PORT, dport=self.DHCP_SERVER_PORT)
         relay_forward_packet /= DHCP6_RelayForward(msgtype=12, linkaddr=self.vlan_ip, peeraddr=self.client_link_local)
-        relay_forward_packet /= DHCP6OptRelayMsg()
-        relay_forward_packet /= DHCP6_Solicit(trid=12345)
+        relay_forward_packet /= DHCP6OptRelayMsg(message=[DHCP6_Solicit(trid=12345)])
 
         return relay_forward_packet
 
     def create_dhcp_relayed_relay_packet(self):
         relayed_relay_packet = Ether(src=self.relay_iface_mac)
         relayed_relay_packet /= IPv6()
         relayed_relay_packet /= UDP(sport=self.DHCP_SERVER_PORT, dport=self.DHCP_SERVER_PORT)
+        packet_inside = DHCP6_RelayForward(msgtype=12, linkaddr=self.vlan_ip, peeraddr=self.client_link_local)
+        packet_inside /= DHCP6OptRelayMsg(message=[DHCP6_Solicit(trid=12345)])
         relayed_relay_packet /= DHCP6_RelayForward(msgtype=12, hopcount = 1, linkaddr=self.relay_linkaddr, peeraddr=self.client_link_local)
-        relayed_relay_packet /= DHCP6OptRelayMsg()
-        relayed_relay_packet /= DHCP6_RelayForward(msgtype=12, linkaddr=self.vlan_ip, peeraddr=self.client_link_local)
-        relayed_relay_packet /= DHCP6OptRelayMsg()
-        relayed_relay_packet /= DHCP6_Solicit(trid=12345)
+        relayed_relay_packet /= DHCP6OptRelayMsg(message=[packet_inside])
 
         return relayed_relay_packet
 
@@ -246,10 +231,9 @@ def create_dhcp_relay_relay_reply_packet(self):
         relay_relay_reply_packet /= IPv6(src=self.server_ip, dst=self.relay_iface_ip)
         relay_relay_reply_packet /= UDP(sport=self.DHCP_SERVER_PORT, dport=self.DHCP_SERVER_PORT)
         relay_relay_reply_packet /= DHCP6_RelayReply(msgtype=13, hopcount = 1, linkaddr=self.vlan_ip, peeraddr=self.client_link_local)
-        relay_relay_reply_packet /= DHCP6OptRelayMsg()
-        relay_relay_reply_packet /= DHCP6_RelayReply(msgtype=13, linkaddr=self.vlan_ip, peeraddr=self.client_link_local)
-        relay_relay_reply_packet /= DHCP6OptRelayMsg()
-        relay_relay_reply_packet /= DHCP6_Reply(trid=12345)
+        packet_inside = DHCP6_RelayReply(msgtype=13, linkaddr=self.vlan_ip, peeraddr=self.client_link_local)
+        packet_inside /= DHCP6OptRelayMsg(message=[DHCP6_Reply(trid=12345)])
+        relay_relay_reply_packet /= DHCP6OptRelayMsg(message=[packet_inside])
 
         return relay_relay_reply_packet
 
@@ -258,8 +242,7 @@ def create_dhcp_relay_reply_packet(self):
         relay_reply_packet /= IPv6(src=self.relay_link_local, dst=self.client_link_local)
         relay_reply_packet /= UDP(sport=self.DHCP_SERVER_PORT, dport=self.DHCP_CLIENT_PORT)
         relay_reply_packet /= DHCP6_RelayReply(msgtype=13, linkaddr=self.vlan_ip, peeraddr=self.client_link_local)
-        relay_reply_packet /= DHCP6OptRelayMsg()
-        relay_reply_packet /= DHCP6_Reply(trid=12345)
+        relay_reply_packet /= DHCP6OptRelayMsg(message=[DHCP6_Reply(trid=12345)])
 
         return relay_reply_packet
 

tests/dhcp_relay/test_dhcp_relay.py

@@ -285,7 +285,7 @@ def test_dhcp_relay_default(ptfhost, dut_dhcp_relay_data, validate_dut_routes_ex
                            "uplink_mac": str(dhcp_relay['uplink_mac']),
                            "testbed_mode": testbed_mode,
                            "testing_mode": testing_mode},
-                   log_file="/tmp/dhcp_relay_test.DHCPTest.log")
+                   log_file="/tmp/dhcp_relay_test.DHCPTest.log", is_python3=True)
 
 
 def test_dhcp_relay_after_link_flap(ptfhost, dut_dhcp_relay_data, validate_dut_routes_exist, testing_config):
@@ -337,7 +337,7 @@ def test_dhcp_relay_after_link_flap(ptfhost, dut_dhcp_relay_data, validate_dut_r
                            "uplink_mac": str(dhcp_relay['uplink_mac']),
                            "testbed_mode": testbed_mode,
                            "testing_mode": testing_mode},
-                   log_file="/tmp/dhcp_relay_test.DHCPTest.log")
+                   log_file="/tmp/dhcp_relay_test.DHCPTest.log", is_python3=True)
 
 
 def test_dhcp_relay_start_with_uplinks_down(ptfhost, dut_dhcp_relay_data, validate_dut_routes_exist, testing_config):
@@ -400,7 +400,7 @@ def test_dhcp_relay_start_with_uplinks_down(ptfhost, dut_dhcp_relay_data, valida
                            "uplink_mac": str(dhcp_relay['uplink_mac']),
                            "testbed_mode": testbed_mode,
                            "testing_mode": testing_mode},
-                   log_file="/tmp/dhcp_relay_test.DHCPTest.log")
+                   log_file="/tmp/dhcp_relay_test.DHCPTest.log", is_python3=True)
 
 
 def test_dhcp_relay_unicast_mac(ptfhost, dut_dhcp_relay_data, validate_dut_routes_exist, testing_config, toggle_all_simulator_ports_to_rand_selected_tor_m):
@@ -436,7 +436,7 @@ def test_dhcp_relay_unicast_mac(ptfhost, dut_dhcp_relay_data, validate_dut_route
                            "uplink_mac": str(dhcp_relay['uplink_mac']),
                            "testbed_mode": testbed_mode,
                            "testing_mode": testing_mode},
-                   log_file="/tmp/dhcp_relay_test.DHCPTest.log")
+                   log_file="/tmp/dhcp_relay_test.DHCPTest.log", is_python3=True)
 
 
 def test_dhcp_relay_random_sport(ptfhost, dut_dhcp_relay_data, validate_dut_routes_exist, testing_config, toggle_all_simulator_ports_to_rand_selected_tor_m):
@@ -471,4 +471,4 @@ def test_dhcp_relay_random_sport(ptfhost, dut_dhcp_relay_data, validate_dut_rout
                            "uplink_mac": str(dhcp_relay['uplink_mac']),
                            "testbed_mode": testbed_mode,
                            "testing_mode": testing_mode},
-                   log_file="/tmp/dhcp_relay_test.DHCPTest.log")
+                   log_file="/tmp/dhcp_relay_test.DHCPTest.log", is_python3=True)