Some changes to enable proto/ptf test to pass on system using Python3

proto/ptf/base_test.py

@@ -410,21 +410,24 @@ def add_to(self, mf_id, mk):
             mf = mk.add()
             mf.field_id = mf_id
             mf.lpm.prefix_len = self.pLen
-            mf.lpm.value = ''
+            assert isinstance(self.v, bytes)
+            orig_v_list = list(self.v)
+            mod_v_list = []
 
             # P4Runtime now has strict rules regarding ternary matches: in the
             # case of LPM, trailing bits in the value (after prefix) must be set
             # to 0.
-            first_byte_masked = self.pLen / 8
+            first_byte_masked = self.pLen // 8
             for i in range(first_byte_masked):
-                mf.lpm.value += self.v[i]
-            if first_byte_masked == len(self.v):
+                mod_v_list.append(orig_v_list[i])
+            if first_byte_masked == len(orig_v_list):
+                mf.lpm.value = bytes(mod_v_list)
                 return
             r = self.pLen % 8
-            mf.lpm.value += chr(
-                ord(self.v[first_byte_masked]) & (0xff << (8 - r)))
-            for i in range(first_byte_masked + 1, len(self.v)):
-                mf.lpm.value += '\x00'
+            mod_v_list.append(orig_v_list[first_byte_masked] & (0xff << (8 - r)))
+            for i in range(first_byte_masked + 1, len(orig_v_list)):
+                mod_v_list.append(0)
+            mf.lpm.value = bytes(mod_v_list)
 
     class Ternary(MF):
         def __init__(self, name, v, mask):

proto/ptf/l3_host_fwd/test/l3_host_fwd.py

@@ -28,7 +28,7 @@
 
 from google.rpc import code_pb2
 
-from base_test import P4RuntimeTest, autocleanup, stringify, ipv4_to_binary
+from base_test import P4RuntimeTest, autocleanup, stringify, ipv4_to_binary, mac_to_binary
 
 class L3HostFwdTest(P4RuntimeTest):
     pass
@@ -37,13 +37,15 @@ class FwdTest(L3HostFwdTest):
     @autocleanup
     def runTest(self):
         ip_dst_addr = "10.0.0.1"
-        ip_dst_addr_str = ipv4_to_binary(ip_dst_addr)
+        ip_dst_addr_bin = ipv4_to_binary(ip_dst_addr)
         ig_port = self.swports(1)
         eg_port = self.swports(2)
         # port is 9-bit in v1model, i.e. 2 bytes
         eg_port_str = stringify(eg_port, 2)
-        smac = "\xee\xcd\x00\x7e\x70\x00"
-        dmac = "\xee\x30\xca\x9d\x1e\x00"
+        smac = "ee:cd:00:7e:70:00"
+        dmac = "ee:30:ca:9d:1e:00"
+        smac_bin = mac_to_binary(smac)
+        dmac_bin = mac_to_binary(dmac)
 
         # we do not care about the src mac address or the src IP address
         pkt = testutils.simple_tcp_packet(
@@ -55,9 +57,9 @@ def runTest(self):
 
         # add a forwarding entry
         self.send_request_add_entry_to_action(
-            "l3_host_fwd", [self.Exact("hdr.ipv4.dst_addr", ip_dst_addr_str)],
+            "l3_host_fwd", [self.Exact("hdr.ipv4.dst_addr", ip_dst_addr_bin)],
             "set_nexthop",
-            [("port", eg_port_str), ("smac", smac), ("dmac", dmac)])
+            [("port", eg_port_str), ("smac", smac_bin), ("dmac", dmac_bin)])
 
         # check that the entry is hit and that no other packets are received
         exp_pkt = testutils.simple_tcp_packet(
@@ -68,18 +70,21 @@ def runTest(self):
 class DupEntryTest(L3HostFwdTest):
     @autocleanup
     def runTest(self):
-        ip_dst_addr_str = "\x0a\x00\x00\x01"
+        ip_dst_addr = "10.0.0.1"
+        ip_dst_addr_bin = ipv4_to_binary(ip_dst_addr)
         eg_port = self.swports(2)
         eg_port_str = stringify(eg_port, 2)
-        smac = "\xee\xcd\x00\x7e\x70\x00"
-        dmac = "\xee\x30\xca\x9d\x1e\x00"
+        smac = "ee:cd:00:7e:70:00"
+        dmac = "ee:30:ca:9d:1e:00"
+        smac_bin = mac_to_binary(smac)
+        dmac_bin = mac_to_binary(dmac)
 
         def add_entry_once():
             self.send_request_add_entry_to_action(
                 "l3_host_fwd",
-                [self.Exact("hdr.ipv4.dst_addr", ip_dst_addr_str)],
+                [self.Exact("hdr.ipv4.dst_addr", ip_dst_addr_bin)],
                 "set_nexthop",
-                [("port", eg_port_str), ("smac", smac), ("dmac", dmac)])
+                [("port", eg_port_str), ("smac", smac_bin), ("dmac", dmac_bin)])
 
         add_entry_once()
         with self.assertP4RuntimeError():
@@ -88,28 +93,31 @@ def add_entry_once():
 class BadMatchKeyTest(L3HostFwdTest):
     @autocleanup
     def runTest(self):
-        ip_dst_addr_str = "\x0a\x00\x00\x01"
-        bad_ip_dst_addr_str = "\x0a\x00\x00"  # missing one byte
+        ip_dst_addr = "10.0.0.1"
+        ip_dst_addr_bin = ipv4_to_binary(ip_dst_addr)
+        bad_ip_dst_addr_bin = ip_dst_addr_bin[0:3]   # missing one byte
         eg_port = self.swports(2)
         eg_port_str = stringify(eg_port, 2)
-        smac = "\xee\xcd\x00\x7e\x70\x00"
-        dmac = "\xee\x30\xca\x9d\x1e\x00"
+        smac = "ee:cd:00:7e:70:00"
+        dmac = "ee:30:ca:9d:1e:00"
+        smac_bin = mac_to_binary(smac)
+        dmac_bin = mac_to_binary(dmac)
 
         # missing one byte
         with self.assertP4RuntimeError(code_pb2.INVALID_ARGUMENT):
             self.send_request_add_entry_to_action(
                 "l3_host_fwd",
-                [self.Exact("hdr.ipv4.dst_addr", bad_ip_dst_addr_str)],
+                [self.Exact("hdr.ipv4.dst_addr", bad_ip_dst_addr_bin)],
                 "set_nexthop",
-                [("port", eg_port_str), ("smac", smac), ("dmac", dmac)])
+                [("port", eg_port_str), ("smac", smac_bin), ("dmac", dmac_bin)])
 
         # unexpected match type
         with self.assertP4RuntimeError(code_pb2.INVALID_ARGUMENT):
             self.send_request_add_entry_to_action(
                 "l3_host_fwd",
-                [self.Lpm("hdr.ipv4.dst_addr", ip_dst_addr_str, 24)],
+                [self.Lpm("hdr.ipv4.dst_addr", ip_dst_addr_bin, 24)],
                 "set_nexthop",
-                [("port", eg_port_str), ("smac", smac), ("dmac", dmac)])
+                [("port", eg_port_str), ("smac", smac_bin), ("dmac", dmac_bin)])
 
 class BadChecksumTest(L3HostFwdTest):
     @autocleanup

proto/ptf/ptf_runner.py

@@ -54,7 +54,7 @@ def check_ifaces(ifaces):
     '''
     Checks that required interfaces exist.
     '''
-    ip_out = subprocess.check_output(['ip', 'link'])
+    ip_out = subprocess.check_output(['ip', 'link'], text=True)
     # 'ip link' returns a list of interfaces as
     # <idx>: <iface>: ...
     # <idx>: <veth>@<peer veth>: ...
@@ -106,7 +106,7 @@ def run_test(config_path, p4info_path, grpc_addr, device_id,
             iface_name = entry["iface_name"]
             port_map[p4_port] = iface_name
 
-    if not check_ifaces(port_map.values()):
+    if not check_ifaces(list(port_map.values())):
         error("Some interfaces are missing")
         return False