Skip to content

Commit

Permalink
ptf: update stringify for compatibility with Py3 (p4lang#600)
Browse files Browse the repository at this point in the history
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 p4lang#599 

Signed-off-by: Radostin Stoyanov <rstoyanov@fedoraproject.org>
  • Loading branch information
rst0git authored Aug 5, 2023
1 parent 5a4c56e commit e86801b
Show file tree
Hide file tree
Showing 2 changed files with 119 additions and 25 deletions.
6 changes: 6 additions & 0 deletions .github/workflows/test.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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 \
Expand Down
138 changes: 113 additions & 25 deletions proto/ptf/base_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import threading
import time
import queue
import unittest

import ptf
from ptf.base_tests import BaseTest
Expand All @@ -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):
Expand All @@ -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
Expand All @@ -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.
Expand Down Expand Up @@ -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()

0 comments on commit e86801b

Please sign in to comment.