aboutsummaryrefslogtreecommitdiffstats
path: root/test/test_abf.py
diff options
context:
space:
mode:
authorKlement Sekera <klement.sekera@gmail.com>2022-04-26 19:02:15 +0200
committerOle Tr�an <otroan@employees.org>2022-05-10 18:52:08 +0000
commitd9b0c6fbf7aa5bd9af84264105b39c82028a4a29 (patch)
tree4f786cfd8ebc2443cb11e11b74c8657204068898 /test/test_abf.py
parentf90348bcb4afd0af2611cefc43b17ef3042b511c (diff)
tests: replace pycodestyle with black
Drop pycodestyle for code style checking in favor of black. Black is much faster, stable PEP8 compliant code style checker offering also automatic formatting. It aims to be very stable and produce smallest diffs. It's used by many small and big projects. Running checkstyle with black takes a few seconds with a terse output. Thus, test-checkstyle-diff is no longer necessary. Expand scope of checkstyle to all python files in the repo, replacing test-checkstyle with checkstyle-python. Also, fixstyle-python is now available for automatic style formatting. Note: python virtualenv has been consolidated in test/Makefile, test/requirements*.txt which will eventually be moved to a central location. This is required to simply the automated generation of docker executor images in the CI. Type: improvement Change-Id: I022a326603485f58585e879ac0f697fceefbc9c8 Signed-off-by: Klement Sekera <klement.sekera@gmail.com> Signed-off-by: Dave Wallace <dwallacelf@gmail.com>
Diffstat (limited to 'test/test_abf.py')
-rw-r--r--test/test_abf.py200
1 files changed, 109 insertions, 91 deletions
diff --git a/test/test_abf.py b/test/test_abf.py
index 097476b879a..856d02a8185 100644
--- a/test/test_abf.py
+++ b/test/test_abf.py
@@ -5,8 +5,13 @@ import unittest
from framework import VppTestCase, VppTestRunner
from vpp_ip import DpoProto
-from vpp_ip_route import VppIpRoute, VppRoutePath, VppMplsLabel, \
- VppIpTable, FibPathProto
+from vpp_ip_route import (
+ VppIpRoute,
+ VppRoutePath,
+ VppMplsLabel,
+ VppIpTable,
+ FibPathProto,
+)
from vpp_acl import AclRule, VppAcl
from scapy.packet import Raw
@@ -31,19 +36,13 @@ def find_abf_policy(test, id):
def find_abf_itf_attach(test, id, sw_if_index):
attachs = test.vapi.abf_itf_attach_dump()
for a in attachs:
- if id == a.attach.policy_id and \
- sw_if_index == a.attach.sw_if_index:
+ if id == a.attach.policy_id and sw_if_index == a.attach.sw_if_index:
return True
return False
class VppAbfPolicy(VppObject):
-
- def __init__(self,
- test,
- policy_id,
- acl,
- paths):
+ def __init__(self, test, policy_id, acl, paths):
self._test = test
self.policy_id = policy_id
self.acl = acl
@@ -55,35 +54,35 @@ class VppAbfPolicy(VppObject):
def add_vpp_config(self):
self._test.vapi.abf_policy_add_del(
1,
- {'policy_id': self.policy_id,
- 'acl_index': self.acl.acl_index,
- 'n_paths': len(self.paths),
- 'paths': self.encoded_paths})
+ {
+ "policy_id": self.policy_id,
+ "acl_index": self.acl.acl_index,
+ "n_paths": len(self.paths),
+ "paths": self.encoded_paths,
+ },
+ )
self._test.registry.register(self, self._test.logger)
def remove_vpp_config(self):
self._test.vapi.abf_policy_add_del(
0,
- {'policy_id': self.policy_id,
- 'acl_index': self.acl.acl_index,
- 'n_paths': len(self.paths),
- 'paths': self.encoded_paths})
+ {
+ "policy_id": self.policy_id,
+ "acl_index": self.acl.acl_index,
+ "n_paths": len(self.paths),
+ "paths": self.encoded_paths,
+ },
+ )
def query_vpp_config(self):
return find_abf_policy(self._test, self.policy_id)
def object_id(self):
- return ("abf-policy-%d" % self.policy_id)
+ return "abf-policy-%d" % self.policy_id
class VppAbfAttach(VppObject):
-
- def __init__(self,
- test,
- policy_id,
- sw_if_index,
- priority,
- is_ipv6=0):
+ def __init__(self, test, policy_id, sw_if_index, priority, is_ipv6=0):
self._test = test
self.policy_id = policy_id
self.sw_if_index = sw_if_index
@@ -93,31 +92,35 @@ class VppAbfAttach(VppObject):
def add_vpp_config(self):
self._test.vapi.abf_itf_attach_add_del(
1,
- {'policy_id': self.policy_id,
- 'sw_if_index': self.sw_if_index,
- 'priority': self.priority,
- 'is_ipv6': self.is_ipv6})
+ {
+ "policy_id": self.policy_id,
+ "sw_if_index": self.sw_if_index,
+ "priority": self.priority,
+ "is_ipv6": self.is_ipv6,
+ },
+ )
self._test.registry.register(self, self._test.logger)
def remove_vpp_config(self):
self._test.vapi.abf_itf_attach_add_del(
0,
- {'policy_id': self.policy_id,
- 'sw_if_index': self.sw_if_index,
- 'priority': self.priority,
- 'is_ipv6': self.is_ipv6})
+ {
+ "policy_id": self.policy_id,
+ "sw_if_index": self.sw_if_index,
+ "priority": self.priority,
+ "is_ipv6": self.is_ipv6,
+ },
+ )
def query_vpp_config(self):
- return find_abf_itf_attach(self._test,
- self.policy_id,
- self.sw_if_index)
+ return find_abf_itf_attach(self._test, self.policy_id, self.sw_if_index)
def object_id(self):
- return ("abf-attach-%d-%d" % (self.policy_id, self.sw_if_index))
+ return "abf-attach-%d-%d" % (self.policy_id, self.sw_if_index)
class TestAbf(VppTestCase):
- """ ABF Test Case """
+ """ABF Test Case"""
@classmethod
def setUpClass(cls):
@@ -147,8 +150,7 @@ class TestAbf(VppTestCase):
super(TestAbf, self).tearDown()
def test_abf4(self):
- """ IPv4 ACL Based Forwarding
- """
+ """IPv4 ACL Based Forwarding"""
#
# We are not testing the various matching capabilities
@@ -163,18 +165,22 @@ class TestAbf(VppTestCase):
#
# Rule 1
#
- rule_1 = AclRule(is_permit=1, proto=17, ports=1234,
- src_prefix=IPv4Network("1.1.1.1/32"),
- dst_prefix=IPv4Network("1.1.1.2/32"))
+ rule_1 = AclRule(
+ is_permit=1,
+ proto=17,
+ ports=1234,
+ src_prefix=IPv4Network("1.1.1.1/32"),
+ dst_prefix=IPv4Network("1.1.1.2/32"),
+ )
acl_1 = VppAcl(self, rules=[rule_1])
acl_1.add_vpp_config()
#
# ABF policy for ACL 1 - path via interface 1
#
- abf_1 = VppAbfPolicy(self, 10, acl_1,
- [VppRoutePath(self.pg1.remote_ip4,
- self.pg1.sw_if_index)])
+ abf_1 = VppAbfPolicy(
+ self, 10, acl_1, [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)]
+ )
abf_1.add_vpp_config()
#
@@ -187,42 +193,43 @@ class TestAbf(VppTestCase):
# fire in packet matching the ACL src,dst. If it's forwarded
# then the ABF was successful, since default routing will drop it
#
- p_1 = (Ether(src=self.pg0.remote_mac,
- dst=self.pg0.local_mac) /
- IP(src="1.1.1.1", dst="1.1.1.2") /
- UDP(sport=1234, dport=1234) /
- Raw(b'\xa5' * 100))
- self.send_and_expect(self.pg0, p_1*NUM_PKTS, self.pg1)
+ p_1 = (
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+ / IP(src="1.1.1.1", dst="1.1.1.2")
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
+ self.send_and_expect(self.pg0, p_1 * NUM_PKTS, self.pg1)
#
# Attach a 'better' priority policy to the same interface
#
- abf_2 = VppAbfPolicy(self, 11, acl_1,
- [VppRoutePath(self.pg2.remote_ip4,
- self.pg2.sw_if_index)])
+ abf_2 = VppAbfPolicy(
+ self, 11, acl_1, [VppRoutePath(self.pg2.remote_ip4, self.pg2.sw_if_index)]
+ )
abf_2.add_vpp_config()
attach_2 = VppAbfAttach(self, 11, self.pg0.sw_if_index, 40)
attach_2.add_vpp_config()
- self.send_and_expect(self.pg0, p_1*NUM_PKTS, self.pg2)
+ self.send_and_expect(self.pg0, p_1 * NUM_PKTS, self.pg2)
#
# Attach a policy with priority in the middle
#
- abf_3 = VppAbfPolicy(self, 12, acl_1,
- [VppRoutePath(self.pg3.remote_ip4,
- self.pg3.sw_if_index)])
+ abf_3 = VppAbfPolicy(
+ self, 12, acl_1, [VppRoutePath(self.pg3.remote_ip4, self.pg3.sw_if_index)]
+ )
abf_3.add_vpp_config()
attach_3 = VppAbfAttach(self, 12, self.pg0.sw_if_index, 45)
attach_3.add_vpp_config()
- self.send_and_expect(self.pg0, p_1*NUM_PKTS, self.pg2)
+ self.send_and_expect(self.pg0, p_1 * NUM_PKTS, self.pg2)
#
# remove the best priority
#
attach_2.remove_vpp_config()
- self.send_and_expect(self.pg0, p_1*NUM_PKTS, self.pg3)
+ self.send_and_expect(self.pg0, p_1 * NUM_PKTS, self.pg3)
#
# Attach one of the same policies to Pg1
@@ -230,11 +237,12 @@ class TestAbf(VppTestCase):
attach_4 = VppAbfAttach(self, 12, self.pg1.sw_if_index, 45)
attach_4.add_vpp_config()
- p_2 = (Ether(src=self.pg1.remote_mac,
- dst=self.pg1.local_mac) /
- IP(src="1.1.1.1", dst="1.1.1.2") /
- UDP(sport=1234, dport=1234) /
- Raw(b'\xa5' * 100))
+ p_2 = (
+ Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
+ / IP(src="1.1.1.1", dst="1.1.1.2")
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
self.send_and_expect(self.pg1, p_2 * NUM_PKTS, self.pg3)
#
@@ -255,22 +263,27 @@ class TestAbf(VppTestCase):
self.pg4.config_ip4()
self.pg4.resolve_arp()
- abf_13 = VppAbfPolicy(self, 13, acl_1,
- [VppRoutePath(self.pg4.remote_ip4,
- 0xffffffff,
- nh_table_id=table_20.table_id)])
+ abf_13 = VppAbfPolicy(
+ self,
+ 13,
+ acl_1,
+ [
+ VppRoutePath(
+ self.pg4.remote_ip4, 0xFFFFFFFF, nh_table_id=table_20.table_id
+ )
+ ],
+ )
abf_13.add_vpp_config()
attach_5 = VppAbfAttach(self, 13, self.pg0.sw_if_index, 30)
attach_5.add_vpp_config()
- self.send_and_expect(self.pg0, p_1*NUM_PKTS, self.pg4)
+ self.send_and_expect(self.pg0, p_1 * NUM_PKTS, self.pg4)
self.pg4.unconfig_ip4()
self.pg4.set_table_ip4(0)
def test_abf6(self):
- """ IPv6 ACL Based Forwarding
- """
+ """IPv6 ACL Based Forwarding"""
#
# Simple test for matching IPv6 packets
@@ -279,32 +292,34 @@ class TestAbf(VppTestCase):
#
# Rule 1
#
- rule_1 = AclRule(is_permit=1, proto=17, ports=1234,
- src_prefix=IPv6Network("2001::2/128"),
- dst_prefix=IPv6Network("2001::1/128"))
+ rule_1 = AclRule(
+ is_permit=1,
+ proto=17,
+ ports=1234,
+ src_prefix=IPv6Network("2001::2/128"),
+ dst_prefix=IPv6Network("2001::1/128"),
+ )
acl_1 = VppAcl(self, rules=[rule_1])
acl_1.add_vpp_config()
#
# ABF policy for ACL 1 - path via interface 1
#
- abf_1 = VppAbfPolicy(self, 10, acl_1,
- [VppRoutePath("3001::1",
- 0xffffffff)])
+ abf_1 = VppAbfPolicy(self, 10, acl_1, [VppRoutePath("3001::1", 0xFFFFFFFF)])
abf_1.add_vpp_config()
- attach_1 = VppAbfAttach(self, 10, self.pg0.sw_if_index,
- 45, is_ipv6=True)
+ attach_1 = VppAbfAttach(self, 10, self.pg0.sw_if_index, 45, is_ipv6=True)
attach_1.add_vpp_config()
#
# a packet matching the rule
#
- p = (Ether(src=self.pg0.remote_mac,
- dst=self.pg0.local_mac) /
- IPv6(src="2001::2", dst="2001::1") /
- UDP(sport=1234, dport=1234) /
- Raw(b'\xa5' * 100))
+ p = (
+ Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
+ / IPv6(src="2001::2", dst="2001::1")
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
#
# packets are dropped because there is no route to the policy's
@@ -315,9 +330,12 @@ class TestAbf(VppTestCase):
#
# add a route resolving the next-hop
#
- route = VppIpRoute(self, "3001::1", 32,
- [VppRoutePath(self.pg1.remote_ip6,
- self.pg1.sw_if_index)])
+ route = VppIpRoute(
+ self,
+ "3001::1",
+ 32,
+ [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)],
+ )
route.add_vpp_config()
#
@@ -326,5 +344,5 @@ class TestAbf(VppTestCase):
self.send_and_expect(self.pg0, p * NUM_PKTS, self.pg1)
-if __name__ == '__main__':
+if __name__ == "__main__":
unittest.main(testRunner=VppTestRunner)