ptf: update stringify for compatibility with Py3 (#600)

The decode('hex') method [1] has been deprecated and removed in Python 3.
Equivalent functionality in Python 3 is provided by int.to_bytes [2].
This patch adds unit tests to verify the functionality of stringify.

[1] https://docs.python.org/2.7/library/codecs.html#python-specific-encodings
[2] https://docs.python.org/3/library/stdtypes.html#int.to_bytes

* proto/ptf: add unit tests for ipv4_to_binary

* proto/ptf: add unit tests for mac_to_binary

Fixes #599 

Signed-off-by: Radostin Stoyanov <rstoyanov@fedoraproject.org>

.github/workflows/test.yml

@@ -46,6 +46,12 @@ jobs:
         CXX: ${{ matrix.cxx }}
       run: |
         docker build -t pi --build-arg IMAGE_TYPE=test --build-arg CC=$CC --build-arg CXX=$CXX -f Dockerfile.bmv2 .
+    - name: Run unit tests
+      run: |
+        docker run -w /PI pi bash -c \
+          'apt-get update && apt-get install -y python3-pip && \
+          python3 -m pip install ptf scapy grpcio googleapis-common-protos protobuf==3.20 && \
+          python3 ./proto/ptf/base_test.py'
     - name: Run tests
       run: |
         docker run -w /PI pi bash -c \

proto/ptf/base_test.py

@@ -25,6 +25,7 @@
 import threading
 import time
 import queue
+import unittest
 
 import ptf
 from ptf.base_tests import BaseTest
@@ -48,23 +49,27 @@ def __get__(self, instance, owner):
         return partial(self.func, instance,
                        *(self.args or ()), **(self.keywords or {}))
 
-# Convert integer (with length) to binary byte string
-# Equivalent to Python 3.2 int.to_bytes
-# See
-# https://stackoverflow.com/questions/16022556/has-python-3-to-bytes-been-back-ported-to-python-2-7
-# TODO: When P4Runtime implementation is ready for it, use
-# minimum-length byte sequences to represent integers.  For unsigned
-# integers, this should only require removing the zfill() call below.
-def stringify(n, length):
+def stringify(n, length=0):
     """Take a non-negative integer 'n' as the first parameter, and a
     non-negative integer 'length' in units of _bytes_ as the second
-    parameter.  Return a string with binary contents expected by the
-    Python P4Runtime client operations.  If 'n' does not fit in
-    'length' bytes, it is represented in the fewest number of bytes it
-    does fit into without loss of precision.  It always returns a
-    string at least one byte long, even if value=width=0."""
-    h = '%x' % n
-    s = ('0'*(len(h) % 2) + h).zfill(length*2).decode('hex')
+    parameter (it defaults to 0 if not provided).  Return a string
+    with binary contents expected by the Python P4Runtime client
+    operations.  If 'n' does not fit in 'length' bytes, it is
+    represented in the fewest number of bytes it does fit into without
+    loss of precision.  It always returns a string at least one byte
+    long, even if n=length=0."""
+    assert isinstance(length, int)
+    assert length >= 0
+    assert isinstance(n, int)
+    assert n >= 0
+    if length == 0 and n == 0:
+        length = 1
+    else:
+        n_size_bits = n.bit_length()
+        n_size_bytes = (n_size_bits + 7) // 8
+        if n_size_bytes > length:
+            length = n_size_bytes
+    s = n.to_bytes(length, byteorder="big")
     return s
 
 def ipv4_to_binary(addr):
@@ -73,11 +78,12 @@ def ipv4_to_binary(addr):
     to a string with binary contents expected by the Python P4Runtime
     client operations."""
     bytes_ = [int(b, 10) for b in addr.split('.')]
-    assert len(bytes_) == 4
-    # Note: The chr(b) call below will throw exception if any b is
-    # outside of the range [0, 255]], so no need to add a separate
-    # check for that here.
-    return "".join(chr(b) for b in bytes_)
+    if len(bytes_) != 4:
+        raise ValueError("Invalid IPv4 address format")
+    for b in bytes_:
+        if b < 0 or b > 255:
+            raise ValueError("IPv4 address contains out-of-range value")
+    return bytes(bytes_)
 
 def mac_to_binary(addr):
     """Take an argument 'addr' containing an Ethernet MAC address written
@@ -86,11 +92,12 @@ def mac_to_binary(addr):
     binary contents expected by the Python P4Runtime client
     operations."""
     bytes_ = [int(b, 16) for b in addr.split(':')]
-    assert len(bytes_) == 6
-    # Note: The chr(b) call below will throw exception if any b is
-    # outside of the range [0, 255]], so no need to add a separate
-    # check for that here.
-    return "".join(chr(b) for b in bytes_)
+    if len(bytes_) != 6:
+        raise ValueError("Invalid MAC address format")
+    for b in bytes_:
+        if b < 0 or b > 255:
+            raise ValueError("MAC address contains out-of-range value")
+    return bytes(bytes_)
 
 # Used to indicate that the gRPC error Status object returned by the server has
 # an incorrect format.
@@ -663,3 +670,84 @@ def handle(*args, **kwargs):
         finally:
             test.undo_write_requests(test._reqs)
     return handle
+
+
+class TestStringify(unittest.TestCase):
+    def test_zero(self):
+        """Test for 0 value"""
+        self.assertEqual(stringify(0, 1), b'\x00')
+        self.assertEqual(stringify(0, 2), b'\x00\x00')
+
+    def test_single_byte(self):
+        """Test for single-byte values"""
+        self.assertEqual(stringify(42, 1), b'*')
+        self.assertEqual(stringify(255, 1), b'\xff')
+
+    def test_multi_byte(self):
+        """Test for multi-byte values"""
+        self.assertEqual(stringify(512, 2), b'\x02\x00')
+        self.assertEqual(stringify(1000, 2), b'\x03\xe8')
+
+    def test_large_value_overflow(self):
+        """Test for a large value that doesn't fit in 'length' bytes"""
+        self.assertEqual(stringify(1000, 1), b'\x03\xe8')
+
+    def test_zero_length(self):
+        """Test for length=0"""
+        self.assertEqual(stringify(0, 0), b'\x00')
+        self.assertEqual(stringify(512, 0), b'\x02\x00')
+
+    def test_default_length(self):
+        """Test for length=None (default length calculation)"""
+        self.assertEqual(stringify(0), b'\x00')
+        self.assertEqual(stringify(255), b'\xff')
+        self.assertEqual(stringify(512), b'\x02\x00')
+        self.assertEqual(stringify(1000), b'\x03\xe8')
+
+
+class TestIPv4ToBinary(unittest.TestCase):
+    def test_valid_ipv4(self):
+        """Test valid IPv4 addresses"""
+        self.assertEqual(ipv4_to_binary('0.0.0.0'), b'\x00\x00\x00\x00')
+        self.assertEqual(ipv4_to_binary('192.168.0.1'), b'\xc0\xa8\x00\x01')
+        self.assertEqual(ipv4_to_binary('255.255.255.255'), b'\xff\xff\xff\xff')
+
+    def test_invalid_ipv4(self):
+        """Test invalid IPv4 addresses (less or more than 4 bytes)"""
+        with self.assertRaises(ValueError):
+            ipv4_to_binary('10.0.1')
+        with self.assertRaises(ValueError):
+            ipv4_to_binary('10.0.1.2.3')
+
+    def test_out_of_range(self):
+        """Test IPv4 address with out-of-range values (above 255)"""
+        with self.assertRaises(ValueError):
+            ipv4_to_binary('256.0.0.0')
+        with self.assertRaises(ValueError):
+            ipv4_to_binary('10.1.300.4')
+
+
+class TestMacToBinary(unittest.TestCase):
+    def test_valid_mac(self):
+        """Test valid MAC addresses"""
+        self.assertEqual(mac_to_binary('00:00:00:00:00:00'), b'\x00\x00\x00\x00\x00\x00')
+        self.assertEqual(mac_to_binary('00:de:ad:be:ef:ff'), b'\x00\xde\xad\xbe\xef\xff')
+        self.assertEqual(mac_to_binary('a0:b1:c2:d3:e4:f5'), b'\xa0\xb1\xc2\xd3\xe4\xf5')
+
+    def test_invalid_mac(self):
+        """Test invalid MAC addresses (less or more than 6 bytes)"""
+        with self.assertRaises(ValueError):
+            mac_to_binary('00:de:ad:be:ef')
+        with self.assertRaises(ValueError):
+            mac_to_binary('00:de:ad:be:ef:ff:aa')
+
+    def test_out_of_range(self):
+        """Test MAC address with out-of-range values (above 255)"""
+        with self.assertRaises(ValueError):
+            mac_to_binary('00:de:ad:be:ef:gg')
+        with self.assertRaises(ValueError):
+            mac_to_binary('00:de:ad:be:ef:256')
+
+
+if __name__ == '__main__':
+    unittest.main()