aboutsummaryrefslogtreecommitdiffstats
path: root/test/test_acl_plugin_l2l3.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_acl_plugin_l2l3.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_acl_plugin_l2l3.py')
-rw-r--r--test/test_acl_plugin_l2l3.py743
1 files changed, 410 insertions, 333 deletions
diff --git a/test/test_acl_plugin_l2l3.py b/test/test_acl_plugin_l2l3.py
index 48faafb7398..343e611751b 100644
--- a/test/test_acl_plugin_l2l3.py
+++ b/test/test_acl_plugin_l2l3.py
@@ -76,12 +76,16 @@ class TestACLpluginL2L3(VppTestCase):
# Create BD with MAC learning enabled and put interfaces to this BD
cls.vapi.sw_interface_set_l2_bridge(
- rx_sw_if_index=cls.loop0.sw_if_index, bd_id=cls.bd_id,
- port_type=L2_PORT_TYPE.BVI)
- cls.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=cls.pg0.sw_if_index,
- bd_id=cls.bd_id)
- cls.vapi.sw_interface_set_l2_bridge(rx_sw_if_index=cls.pg1.sw_if_index,
- bd_id=cls.bd_id)
+ rx_sw_if_index=cls.loop0.sw_if_index,
+ bd_id=cls.bd_id,
+ port_type=L2_PORT_TYPE.BVI,
+ )
+ cls.vapi.sw_interface_set_l2_bridge(
+ rx_sw_if_index=cls.pg0.sw_if_index, bd_id=cls.bd_id
+ )
+ cls.vapi.sw_interface_set_l2_bridge(
+ rx_sw_if_index=cls.pg1.sw_if_index, bd_id=cls.bd_id
+ )
# Configure IPv4 addresses on loopback interface and routed interface
cls.loop0.config_ip4()
@@ -125,8 +129,7 @@ class TestACLpluginL2L3(VppTestCase):
self.logger.info(self.vapi.cli("show l2patch"))
self.logger.info(self.vapi.cli("show classify tables"))
self.logger.info(self.vapi.cli("show l2fib verbose"))
- self.logger.info(self.vapi.cli("show bridge-domain %s detail" %
- self.bd_id))
+ self.logger.info(self.vapi.cli("show bridge-domain %s detail" % self.bd_id))
self.logger.info(self.vapi.cli("show ip neighbors"))
cmd = "show acl-plugin sessions verbose 1"
self.logger.info(self.vapi.cli(cmd))
@@ -134,23 +137,33 @@ class TestACLpluginL2L3(VppTestCase):
self.logger.info(self.vapi.cli("show acl-plugin interface"))
self.logger.info(self.vapi.cli("show acl-plugin tables"))
- def create_stream(self, src_ip_if, dst_ip_if, reverse, packet_sizes,
- is_ip6, expect_blocked, expect_established,
- add_extension_header, icmp_stateful=False):
+ def create_stream(
+ self,
+ src_ip_if,
+ dst_ip_if,
+ reverse,
+ packet_sizes,
+ is_ip6,
+ expect_blocked,
+ expect_established,
+ add_extension_header,
+ icmp_stateful=False,
+ ):
pkts = []
rules = []
permit_rules = []
permit_and_reflect_rules = []
total_packet_count = 8
for i in range(0, total_packet_count):
- modulo = (i//2) % 2
+ modulo = (i // 2) % 2
icmp_type_delta = i % 2
icmp_code = i
- is_udp_packet = (modulo == 0)
+ is_udp_packet = modulo == 0
if is_udp_packet and icmp_stateful:
continue
- is_reflectable_icmp = (icmp_stateful and icmp_type_delta == 0 and
- not is_udp_packet)
+ is_reflectable_icmp = (
+ icmp_stateful and icmp_type_delta == 0 and not is_udp_packet
+ )
is_reflected_icmp = is_reflectable_icmp and expect_established
can_reflect_this_packet = is_udp_packet or is_reflectable_icmp
is_permit = i % 2
@@ -161,9 +174,9 @@ class TestACLpluginL2L3(VppTestCase):
payload = self.info_to_payload(info)
else:
to_be_blocked = False
- if (expect_blocked and not expect_established):
+ if expect_blocked and not expect_established:
to_be_blocked = True
- if (not can_reflect_this_packet):
+ if not can_reflect_this_packet:
to_be_blocked = True
if to_be_blocked:
payload = "to be blocked"
@@ -171,7 +184,7 @@ class TestACLpluginL2L3(VppTestCase):
info = self.create_packet_info(src_ip_if, dst_ip_if)
payload = self.info_to_payload(info)
if reverse:
- dst_mac = 'de:ad:00:00:00:00'
+ dst_mac = "de:ad:00:00:00:00"
src_mac = remote_dst_host._mac
dst_ip6 = src_ip_if.remote_ip6
src_ip6 = remote_dst_host.ip6
@@ -201,42 +214,53 @@ class TestACLpluginL2L3(VppTestCase):
ulp_l4 = UDP(sport=src_l4, dport=dst_l4)
if add_extension_header:
# prepend some extension headers
- ulp = (IPv6ExtHdrRouting() / IPv6ExtHdrRouting() /
- IPv6ExtHdrFragment(offset=0, m=1) / ulp_l4)
+ ulp = (
+ IPv6ExtHdrRouting()
+ / IPv6ExtHdrRouting()
+ / IPv6ExtHdrFragment(offset=0, m=1)
+ / ulp_l4
+ )
# uncomment below to test invalid ones
# ulp = IPv6ExtHdrRouting(len = 200) / ulp_l4
else:
ulp = ulp_l4
- p = (Ether(dst=dst_mac, src=src_mac) /
- IPv6(src=src_ip6, dst=dst_ip6) /
- ulp /
- Raw(payload))
+ p = (
+ Ether(dst=dst_mac, src=src_mac)
+ / IPv6(src=src_ip6, dst=dst_ip6)
+ / ulp
+ / Raw(payload)
+ )
else:
ulp_l4 = UDP(sport=src_l4, dport=dst_l4)
# IPv4 does not allow extension headers,
# but we rather make it a first fragment
flags = 1 if add_extension_header else 0
ulp = ulp_l4
- p = (Ether(dst=dst_mac, src=src_mac) /
- IP(src=src_ip4, dst=dst_ip4, frag=0, flags=flags) /
- ulp /
- Raw(payload))
+ p = (
+ Ether(dst=dst_mac, src=src_mac)
+ / IP(src=src_ip4, dst=dst_ip4, frag=0, flags=flags)
+ / ulp
+ / Raw(payload)
+ )
elif modulo == 1:
if is_ip6:
- ulp_l4 = ICMPv6Unknown(type=128 + icmp_type_delta,
- code=icmp_code)
+ ulp_l4 = ICMPv6Unknown(type=128 + icmp_type_delta, code=icmp_code)
ulp = ulp_l4
- p = (Ether(dst=dst_mac, src=src_mac) /
- IPv6(src=src_ip6, dst=dst_ip6) /
- ulp /
- Raw(payload))
+ p = (
+ Ether(dst=dst_mac, src=src_mac)
+ / IPv6(src=src_ip6, dst=dst_ip6)
+ / ulp
+ / Raw(payload)
+ )
else:
- ulp_l4 = ICMP(type=8 - 8*icmp_type_delta, code=icmp_code)
+ ulp_l4 = ICMP(type=8 - 8 * icmp_type_delta, code=icmp_code)
ulp = ulp_l4
- p = (Ether(dst=dst_mac, src=src_mac) /
- IP(src=src_ip4, dst=dst_ip4) /
- ulp /
- Raw(payload))
+ p = (
+ Ether(dst=dst_mac, src=src_mac)
+ / IP(src=src_ip4, dst=dst_ip4)
+ / ulp
+ / Raw(payload)
+ )
if i % 2 == 1:
info.data = p.copy()
@@ -259,19 +283,20 @@ class TestACLpluginL2L3(VppTestCase):
rule_l4_sport = p[ICMPv6Unknown].type
rule_l4_dport = p[ICMPv6Unknown].code
if p.haslayer(IPv6):
- rule_l4_proto = ulp_l4.overload_fields[IPv6]['nh']
+ rule_l4_proto = ulp_l4.overload_fields[IPv6]["nh"]
else:
rule_l4_proto = p[IP].proto
- new_rule = AclRule(is_permit=is_permit, proto=rule_l4_proto,
- src_prefix=ip_network(
- (p[rule_l3_layer].src, rule_prefix_len)),
- dst_prefix=ip_network(
- (p[rule_l3_layer].dst, rule_prefix_len)),
- sport_from=rule_l4_sport,
- sport_to=rule_l4_sport,
- dport_from=rule_l4_dport,
- dport_to=rule_l4_dport)
+ new_rule = AclRule(
+ is_permit=is_permit,
+ proto=rule_l4_proto,
+ src_prefix=ip_network((p[rule_l3_layer].src, rule_prefix_len)),
+ dst_prefix=ip_network((p[rule_l3_layer].dst, rule_prefix_len)),
+ sport_from=rule_l4_sport,
+ sport_to=rule_l4_sport,
+ dport_from=rule_l4_dport,
+ dport_to=rule_l4_dport,
+ )
rules.append(new_rule)
new_rule_permit = copy.copy(new_rule)
@@ -287,10 +312,12 @@ class TestACLpluginL2L3(VppTestCase):
permit_and_reflect_rules.append(new_rule_permit_and_reflect)
self.logger.info("create_stream pkt#%d: %s" % (i, payload))
- return {'stream': pkts,
- 'rules': rules,
- 'permit_rules': permit_rules,
- 'permit_and_reflect_rules': permit_and_reflect_rules}
+ return {
+ "stream": pkts,
+ "rules": rules,
+ "permit_rules": permit_rules,
+ "permit_and_reflect_rules": permit_and_reflect_rules,
+ }
def verify_capture(self, dst_ip_if, src_ip_if, capture, reverse):
last_info = dict()
@@ -316,11 +343,13 @@ class TestACLpluginL2L3(VppTestCase):
data = scapy.compat.raw(packet[UDP][Raw])
else:
if l3 == IP:
- data = scapy.compat.raw(ICMP(
- scapy.compat.raw(packet[l3].payload))[Raw])
+ data = scapy.compat.raw(
+ ICMP(scapy.compat.raw(packet[l3].payload))[Raw]
+ )
else:
- data = scapy.compat.raw(ICMPv6Unknown(
- scapy.compat.raw(packet[l3].payload)).msgbody)
+ data = scapy.compat.raw(
+ ICMPv6Unknown(scapy.compat.raw(packet[l3].payload)).msgbody
+ )
udp_or_icmp = packet[l3].payload
data_obj = Raw(data)
# FIXME: make framework believe we are on object
@@ -330,8 +359,8 @@ class TestACLpluginL2L3(VppTestCase):
self.assertEqual(payload_info.dst, dst_ip_sw_if_index)
next_info = self.get_next_packet_info_for_interface2(
- payload_info.src, dst_ip_sw_if_index,
- last_info[payload_info.src])
+ payload_info.src, dst_ip_sw_if_index, last_info[payload_info.src]
+ )
last_info[payload_info.src] = next_info
self.assertTrue(next_info is not None)
self.assertEqual(packet_index, next_info.index)
@@ -400,13 +429,13 @@ class TestACLpluginL2L3(VppTestCase):
# change the ACLs a few times
for i in range(1, 10):
shuffle(all_rules)
- acl1.modify_vpp_config(all_rules[::1+(i % 2)])
+ acl1.modify_vpp_config(all_rules[:: 1 + (i % 2)])
shuffle(all_rules)
- acl2.modify_vpp_config(all_rules[::1+(i % 3)])
+ acl2.modify_vpp_config(all_rules[:: 1 + (i % 3)])
shuffle(all_rules)
- acl3.modify_vpp_config(all_rules[::1+(i % 5)])
+ acl3.modify_vpp_config(all_rules[:: 1 + (i % 5)])
# restore to how it was before and clean up
acl_if.n_input = saved_n_input
@@ -417,143 +446,200 @@ class TestACLpluginL2L3(VppTestCase):
acl2.remove_vpp_config()
acl3.remove_vpp_config()
- def create_acls_for_a_stream(self, stream_dict,
- test_l2_action, is_reflect):
- r = stream_dict['rules']
- r_permit = stream_dict['permit_rules']
- r_permit_reflect = stream_dict['permit_and_reflect_rules']
+ def create_acls_for_a_stream(self, stream_dict, test_l2_action, is_reflect):
+ r = stream_dict["rules"]
+ r_permit = stream_dict["permit_rules"]
+ r_permit_reflect = stream_dict["permit_and_reflect_rules"]
r_action = r_permit_reflect if is_reflect else r
action_acl = VppAcl(self, rules=r_action, tag="act. acl")
action_acl.add_vpp_config()
permit_acl = VppAcl(self, rules=r_permit, tag="perm. acl")
permit_acl.add_vpp_config()
- return {'L2': action_acl if test_l2_action else permit_acl,
- 'L3': permit_acl if test_l2_action else action_acl,
- 'permit': permit_acl, 'action': action_acl}
-
- def apply_acl_ip46_x_to_y(self, bridged_to_routed, test_l2_deny,
- is_ip6, is_reflect, add_eh):
- """ Apply the ACLs
- """
+ return {
+ "L2": action_acl if test_l2_action else permit_acl,
+ "L3": permit_acl if test_l2_action else action_acl,
+ "permit": permit_acl,
+ "action": action_acl,
+ }
+
+ def apply_acl_ip46_x_to_y(
+ self, bridged_to_routed, test_l2_deny, is_ip6, is_reflect, add_eh
+ ):
+ """Apply the ACLs"""
self.reset_packet_infos()
stream_dict = self.create_stream(
- self.pg2, self.loop0,
+ self.pg2,
+ self.loop0,
bridged_to_routed,
- self.pg_if_packet_sizes, is_ip6,
- not is_reflect, False, add_eh)
- stream = stream_dict['stream']
- acl_idx = self.create_acls_for_a_stream(stream_dict, test_l2_deny,
- is_reflect)
+ self.pg_if_packet_sizes,
+ is_ip6,
+ not is_reflect,
+ False,
+ add_eh,
+ )
+ stream = stream_dict["stream"]
+ acl_idx = self.create_acls_for_a_stream(stream_dict, test_l2_deny, is_reflect)
n_input_l3 = 0 if bridged_to_routed else 1
n_input_l2 = 1 if bridged_to_routed else 0
- acl_if_pg2 = VppAclInterface(self, sw_if_index=self.pg2.sw_if_index,
- n_input=n_input_l3, acls=[acl_idx['L3']])
+ acl_if_pg2 = VppAclInterface(
+ self,
+ sw_if_index=self.pg2.sw_if_index,
+ n_input=n_input_l3,
+ acls=[acl_idx["L3"]],
+ )
acl_if_pg2.add_vpp_config()
- acl_if_pg0 = VppAclInterface(self, sw_if_index=self.pg0.sw_if_index,
- n_input=n_input_l2, acls=[acl_idx['L2']])
+ acl_if_pg0 = VppAclInterface(
+ self,
+ sw_if_index=self.pg0.sw_if_index,
+ n_input=n_input_l2,
+ acls=[acl_idx["L2"]],
+ )
acl_if_pg0.add_vpp_config()
- acl_if_pg1 = VppAclInterface(self, sw_if_index=self.pg1.sw_if_index,
- n_input=n_input_l2, acls=[acl_idx['L2']])
+ acl_if_pg1 = VppAclInterface(
+ self,
+ sw_if_index=self.pg1.sw_if_index,
+ n_input=n_input_l2,
+ acls=[acl_idx["L2"]],
+ )
acl_if_pg1.add_vpp_config()
self.applied_acl_shuffle(acl_if_pg0)
self.applied_acl_shuffle(acl_if_pg1)
- return {'L2': acl_idx['L2'], 'L3': acl_idx['L3']}
+ return {"L2": acl_idx["L2"], "L3": acl_idx["L3"]}
- def apply_acl_ip46_both_directions_reflect(self,
- primary_is_bridged_to_routed,
- reflect_on_l2, is_ip6, add_eh,
- stateful_icmp):
+ def apply_acl_ip46_both_directions_reflect(
+ self, primary_is_bridged_to_routed, reflect_on_l2, is_ip6, add_eh, stateful_icmp
+ ):
primary_is_routed_to_bridged = not primary_is_bridged_to_routed
self.reset_packet_infos()
- stream_dict_fwd = self.create_stream(self.pg2, self.loop0,
- primary_is_bridged_to_routed,
- self.pg_if_packet_sizes, is_ip6,
- False, False, add_eh,
- stateful_icmp)
- acl_idx_fwd = self.create_acls_for_a_stream(stream_dict_fwd,
- reflect_on_l2, True)
-
- stream_dict_rev = self.create_stream(self.pg2, self.loop0,
- not primary_is_bridged_to_routed,
- self.pg_if_packet_sizes, is_ip6,
- True, True, add_eh, stateful_icmp)
+ stream_dict_fwd = self.create_stream(
+ self.pg2,
+ self.loop0,
+ primary_is_bridged_to_routed,
+ self.pg_if_packet_sizes,
+ is_ip6,
+ False,
+ False,
+ add_eh,
+ stateful_icmp,
+ )
+ acl_idx_fwd = self.create_acls_for_a_stream(
+ stream_dict_fwd, reflect_on_l2, True
+ )
+
+ stream_dict_rev = self.create_stream(
+ self.pg2,
+ self.loop0,
+ not primary_is_bridged_to_routed,
+ self.pg_if_packet_sizes,
+ is_ip6,
+ True,
+ True,
+ add_eh,
+ stateful_icmp,
+ )
# We want the primary action to be "deny" rather than reflect
- acl_idx_rev = self.create_acls_for_a_stream(stream_dict_rev,
- reflect_on_l2, False)
+ acl_idx_rev = self.create_acls_for_a_stream(
+ stream_dict_rev, reflect_on_l2, False
+ )
if primary_is_bridged_to_routed:
- inbound_l2_acl = acl_idx_fwd['L2']
+ inbound_l2_acl = acl_idx_fwd["L2"]
else:
- inbound_l2_acl = acl_idx_rev['L2']
+ inbound_l2_acl = acl_idx_rev["L2"]
if primary_is_routed_to_bridged:
- outbound_l2_acl = acl_idx_fwd['L2']
+ outbound_l2_acl = acl_idx_fwd["L2"]
else:
- outbound_l2_acl = acl_idx_rev['L2']
+ outbound_l2_acl = acl_idx_rev["L2"]
if primary_is_routed_to_bridged:
- inbound_l3_acl = acl_idx_fwd['L3']
+ inbound_l3_acl = acl_idx_fwd["L3"]
else:
- inbound_l3_acl = acl_idx_rev['L3']
+ inbound_l3_acl = acl_idx_rev["L3"]
if primary_is_bridged_to_routed:
- outbound_l3_acl = acl_idx_fwd['L3']
+ outbound_l3_acl = acl_idx_fwd["L3"]
else:
- outbound_l3_acl = acl_idx_rev['L3']
-
- acl_if_pg2 = VppAclInterface(self, sw_if_index=self.pg2.sw_if_index,
- n_input=1,
- acls=[inbound_l3_acl, outbound_l3_acl])
+ outbound_l3_acl = acl_idx_rev["L3"]
+
+ acl_if_pg2 = VppAclInterface(
+ self,
+ sw_if_index=self.pg2.sw_if_index,
+ n_input=1,
+ acls=[inbound_l3_acl, outbound_l3_acl],
+ )
acl_if_pg2.add_vpp_config()
- acl_if_pg0 = VppAclInterface(self, sw_if_index=self.pg0.sw_if_index,
- n_input=1,
- acls=[inbound_l2_acl, outbound_l2_acl])
+ acl_if_pg0 = VppAclInterface(
+ self,
+ sw_if_index=self.pg0.sw_if_index,
+ n_input=1,
+ acls=[inbound_l2_acl, outbound_l2_acl],
+ )
acl_if_pg0.add_vpp_config()
- acl_if_pg1 = VppAclInterface(self, sw_if_index=self.pg1.sw_if_index,
- n_input=1,
- acls=[inbound_l2_acl, outbound_l2_acl])
+ acl_if_pg1 = VppAclInterface(
+ self,
+ sw_if_index=self.pg1.sw_if_index,
+ n_input=1,
+ acls=[inbound_l2_acl, outbound_l2_acl],
+ )
acl_if_pg1.add_vpp_config()
self.applied_acl_shuffle(acl_if_pg0)
self.applied_acl_shuffle(acl_if_pg2)
- def apply_acl_ip46_routed_to_bridged(self, test_l2_deny, is_ip6,
- is_reflect, add_eh):
- return self.apply_acl_ip46_x_to_y(False, test_l2_deny, is_ip6,
- is_reflect, add_eh)
+ def apply_acl_ip46_routed_to_bridged(
+ self, test_l2_deny, is_ip6, is_reflect, add_eh
+ ):
+ return self.apply_acl_ip46_x_to_y(
+ False, test_l2_deny, is_ip6, is_reflect, add_eh
+ )
- def apply_acl_ip46_bridged_to_routed(self, test_l2_deny, is_ip6,
- is_reflect, add_eh):
- return self.apply_acl_ip46_x_to_y(True, test_l2_deny, is_ip6,
- is_reflect, add_eh)
+ def apply_acl_ip46_bridged_to_routed(
+ self, test_l2_deny, is_ip6, is_reflect, add_eh
+ ):
+ return self.apply_acl_ip46_x_to_y(
+ True, test_l2_deny, is_ip6, is_reflect, add_eh
+ )
def verify_acl_packet_count(self, acl_idx, packet_count):
- matches = self.statistics.get_counter('/acl/%d/matches' % acl_idx)
+ matches = self.statistics.get_counter("/acl/%d/matches" % acl_idx)
self.logger.info("stat seg for ACL %d: %s" % (acl_idx, repr(matches)))
total_count = 0
for m in matches:
for p in m:
- total_count = total_count + p['packets']
+ total_count = total_count + p["packets"]
self.assertEqual(total_count, packet_count)
- def run_traffic_ip46_x_to_y(self, bridged_to_routed,
- test_l2_deny, is_ip6,
- is_reflect, is_established, add_eh,
- stateful_icmp=False):
+ def run_traffic_ip46_x_to_y(
+ self,
+ bridged_to_routed,
+ test_l2_deny,
+ is_ip6,
+ is_reflect,
+ is_established,
+ add_eh,
+ stateful_icmp=False,
+ ):
self.reset_packet_infos()
- stream_dict = self.create_stream(self.pg2, self.loop0,
- bridged_to_routed,
- self.pg_if_packet_sizes, is_ip6,
- not is_reflect, is_established,
- add_eh, stateful_icmp)
- stream = stream_dict['stream']
+ stream_dict = self.create_stream(
+ self.pg2,
+ self.loop0,
+ bridged_to_routed,
+ self.pg_if_packet_sizes,
+ is_ip6,
+ not is_reflect,
+ is_established,
+ add_eh,
+ stateful_icmp,
+ )
+ stream = stream_dict["stream"]
tx_if = self.pg0 if bridged_to_routed else self.pg2
rx_if = self.pg2 if bridged_to_routed else self.pg0
@@ -566,68 +652,90 @@ class TestACLpluginL2L3(VppTestCase):
self.verify_capture(self.loop0, self.pg2, rcvd1, bridged_to_routed)
return len(stream)
- def run_traffic_ip46_routed_to_bridged(self, test_l2_deny, is_ip6,
- is_reflect, is_established, add_eh,
- stateful_icmp=False):
- return self.run_traffic_ip46_x_to_y(False, test_l2_deny, is_ip6,
- is_reflect, is_established, add_eh,
- stateful_icmp)
-
- def run_traffic_ip46_bridged_to_routed(self, test_l2_deny, is_ip6,
- is_reflect, is_established, add_eh,
- stateful_icmp=False):
- return self.run_traffic_ip46_x_to_y(True, test_l2_deny, is_ip6,
- is_reflect, is_established, add_eh,
- stateful_icmp)
-
- def run_test_ip46_routed_to_bridged(self, test_l2_deny,
- is_ip6, is_reflect, add_eh):
- acls = self.apply_acl_ip46_routed_to_bridged(test_l2_deny,
- is_ip6, is_reflect,
- add_eh)
- pkts = self.run_traffic_ip46_routed_to_bridged(test_l2_deny, is_ip6,
- is_reflect, False,
- add_eh)
- self.verify_acl_packet_count(acls['L3'].acl_index, pkts)
-
- def run_test_ip46_bridged_to_routed(self, test_l2_deny,
- is_ip6, is_reflect, add_eh):
- acls = self.apply_acl_ip46_bridged_to_routed(test_l2_deny,
- is_ip6, is_reflect,
- add_eh)
- pkts = self.run_traffic_ip46_bridged_to_routed(test_l2_deny, is_ip6,
- is_reflect, False,
- add_eh)
- self.verify_acl_packet_count(acls['L2'].acl_index, pkts)
-
- def run_test_ip46_routed_to_bridged_and_back(self, test_l2_action,
- is_ip6, add_eh,
- stateful_icmp=False):
- self.apply_acl_ip46_both_directions_reflect(False, test_l2_action,
- is_ip6, add_eh,
- stateful_icmp)
- self.run_traffic_ip46_routed_to_bridged(test_l2_action, is_ip6,
- True, False, add_eh,
- stateful_icmp)
- self.run_traffic_ip46_bridged_to_routed(test_l2_action, is_ip6,
- False, True, add_eh,
- stateful_icmp)
-
- def run_test_ip46_bridged_to_routed_and_back(self, test_l2_action,
- is_ip6, add_eh,
- stateful_icmp=False):
- self.apply_acl_ip46_both_directions_reflect(True, test_l2_action,
- is_ip6, add_eh,
- stateful_icmp)
- self.run_traffic_ip46_bridged_to_routed(test_l2_action, is_ip6,
- True, False, add_eh,
- stateful_icmp)
- self.run_traffic_ip46_routed_to_bridged(test_l2_action, is_ip6,
- False, True, add_eh,
- stateful_icmp)
+ def run_traffic_ip46_routed_to_bridged(
+ self,
+ test_l2_deny,
+ is_ip6,
+ is_reflect,
+ is_established,
+ add_eh,
+ stateful_icmp=False,
+ ):
+ return self.run_traffic_ip46_x_to_y(
+ False,
+ test_l2_deny,
+ is_ip6,
+ is_reflect,
+ is_established,
+ add_eh,
+ stateful_icmp,
+ )
+
+ def run_traffic_ip46_bridged_to_routed(
+ self,
+ test_l2_deny,
+ is_ip6,
+ is_reflect,
+ is_established,
+ add_eh,
+ stateful_icmp=False,
+ ):
+ return self.run_traffic_ip46_x_to_y(
+ True,
+ test_l2_deny,
+ is_ip6,
+ is_reflect,
+ is_established,
+ add_eh,
+ stateful_icmp,
+ )
+
+ def run_test_ip46_routed_to_bridged(self, test_l2_deny, is_ip6, is_reflect, add_eh):
+ acls = self.apply_acl_ip46_routed_to_bridged(
+ test_l2_deny, is_ip6, is_reflect, add_eh
+ )
+ pkts = self.run_traffic_ip46_routed_to_bridged(
+ test_l2_deny, is_ip6, is_reflect, False, add_eh
+ )
+ self.verify_acl_packet_count(acls["L3"].acl_index, pkts)
+
+ def run_test_ip46_bridged_to_routed(self, test_l2_deny, is_ip6, is_reflect, add_eh):
+ acls = self.apply_acl_ip46_bridged_to_routed(
+ test_l2_deny, is_ip6, is_reflect, add_eh
+ )
+ pkts = self.run_traffic_ip46_bridged_to_routed(
+ test_l2_deny, is_ip6, is_reflect, False, add_eh
+ )
+ self.verify_acl_packet_count(acls["L2"].acl_index, pkts)
+
+ def run_test_ip46_routed_to_bridged_and_back(
+ self, test_l2_action, is_ip6, add_eh, stateful_icmp=False
+ ):
+ self.apply_acl_ip46_both_directions_reflect(
+ False, test_l2_action, is_ip6, add_eh, stateful_icmp
+ )
+ self.run_traffic_ip46_routed_to_bridged(
+ test_l2_action, is_ip6, True, False, add_eh, stateful_icmp
+ )
+ self.run_traffic_ip46_bridged_to_routed(
+ test_l2_action, is_ip6, False, True, add_eh, stateful_icmp
+ )
+
+ def run_test_ip46_bridged_to_routed_and_back(
+ self, test_l2_action, is_ip6, add_eh, stateful_icmp=False
+ ):
+ self.apply_acl_ip46_both_directions_reflect(
+ True, test_l2_action, is_ip6, add_eh, stateful_icmp
+ )
+ self.run_traffic_ip46_bridged_to_routed(
+ test_l2_action, is_ip6, True, False, add_eh, stateful_icmp
+ )
+ self.run_traffic_ip46_routed_to_bridged(
+ test_l2_action, is_ip6, False, True, add_eh, stateful_icmp
+ )
def test_0000_ip6_irb_1(self):
- """ ACL plugin prepare"""
+ """ACL plugin prepare"""
if not self.vpp_dead:
cmd = "set acl-plugin session timeout udp idle 2000"
self.logger.info(self.vapi.ppcli(cmd))
@@ -646,219 +754,188 @@ class TestACLpluginL2L3(VppTestCase):
# "set acl-plugin l2-datapath old"))
def test_0001_ip6_irb_1(self):
- """ ACL IPv6 routed -> bridged, L2 ACL deny"""
- self.run_test_ip46_routed_to_bridged(True, True, False,
- self.WITHOUT_EH)
+ """ACL IPv6 routed -> bridged, L2 ACL deny"""
+ self.run_test_ip46_routed_to_bridged(True, True, False, self.WITHOUT_EH)
def test_0002_ip6_irb_1(self):
- """ ACL IPv6 routed -> bridged, L3 ACL deny"""
- self.run_test_ip46_routed_to_bridged(False, True, False,
- self.WITHOUT_EH)
+ """ACL IPv6 routed -> bridged, L3 ACL deny"""
+ self.run_test_ip46_routed_to_bridged(False, True, False, self.WITHOUT_EH)
def test_0003_ip4_irb_1(self):
- """ ACL IPv4 routed -> bridged, L2 ACL deny"""
- self.run_test_ip46_routed_to_bridged(True, False, False,
- self.WITHOUT_EH)
+ """ACL IPv4 routed -> bridged, L2 ACL deny"""
+ self.run_test_ip46_routed_to_bridged(True, False, False, self.WITHOUT_EH)
def test_0004_ip4_irb_1(self):
- """ ACL IPv4 routed -> bridged, L3 ACL deny"""
- self.run_test_ip46_routed_to_bridged(False, False, False,
- self.WITHOUT_EH)
+ """ACL IPv4 routed -> bridged, L3 ACL deny"""
+ self.run_test_ip46_routed_to_bridged(False, False, False, self.WITHOUT_EH)
def test_0005_ip6_irb_1(self):
- """ ACL IPv6 bridged -> routed, L2 ACL deny """
- self.run_test_ip46_bridged_to_routed(True, True, False,
- self.WITHOUT_EH)
+ """ACL IPv6 bridged -> routed, L2 ACL deny"""
+ self.run_test_ip46_bridged_to_routed(True, True, False, self.WITHOUT_EH)
def test_0006_ip6_irb_1(self):
- """ ACL IPv6 bridged -> routed, L3 ACL deny """
- self.run_test_ip46_bridged_to_routed(False, True, False,
- self.WITHOUT_EH)
+ """ACL IPv6 bridged -> routed, L3 ACL deny"""
+ self.run_test_ip46_bridged_to_routed(False, True, False, self.WITHOUT_EH)
def test_0007_ip6_irb_1(self):
- """ ACL IPv4 bridged -> routed, L2 ACL deny """
- self.run_test_ip46_bridged_to_routed(True, False, False,
- self.WITHOUT_EH)
+ """ACL IPv4 bridged -> routed, L2 ACL deny"""
+ self.run_test_ip46_bridged_to_routed(True, False, False, self.WITHOUT_EH)
def test_0008_ip6_irb_1(self):
- """ ACL IPv4 bridged -> routed, L3 ACL deny """
- self.run_test_ip46_bridged_to_routed(False, False, False,
- self.WITHOUT_EH)
+ """ACL IPv4 bridged -> routed, L3 ACL deny"""
+ self.run_test_ip46_bridged_to_routed(False, False, False, self.WITHOUT_EH)
# Stateful ACL tests
def test_0101_ip6_irb_1(self):
- """ ACL IPv6 routed -> bridged, L2 ACL permit+reflect"""
- self.run_test_ip46_routed_to_bridged_and_back(True, True,
- self.WITHOUT_EH)
+ """ACL IPv6 routed -> bridged, L2 ACL permit+reflect"""
+ self.run_test_ip46_routed_to_bridged_and_back(True, True, self.WITHOUT_EH)
def test_0102_ip6_irb_1(self):
- """ ACL IPv6 bridged -> routed, L2 ACL permit+reflect"""
- self.run_test_ip46_bridged_to_routed_and_back(True, True,
- self.WITHOUT_EH)
+ """ACL IPv6 bridged -> routed, L2 ACL permit+reflect"""
+ self.run_test_ip46_bridged_to_routed_and_back(True, True, self.WITHOUT_EH)
def test_0103_ip6_irb_1(self):
- """ ACL IPv4 routed -> bridged, L2 ACL permit+reflect"""
- self.run_test_ip46_routed_to_bridged_and_back(True, False,
- self.WITHOUT_EH)
+ """ACL IPv4 routed -> bridged, L2 ACL permit+reflect"""
+ self.run_test_ip46_routed_to_bridged_and_back(True, False, self.WITHOUT_EH)
def test_0104_ip6_irb_1(self):
- """ ACL IPv4 bridged -> routed, L2 ACL permit+reflect"""
- self.run_test_ip46_bridged_to_routed_and_back(True, False,
- self.WITHOUT_EH)
+ """ACL IPv4 bridged -> routed, L2 ACL permit+reflect"""
+ self.run_test_ip46_bridged_to_routed_and_back(True, False, self.WITHOUT_EH)
def test_0111_ip6_irb_1(self):
- """ ACL IPv6 routed -> bridged, L3 ACL permit+reflect"""
- self.run_test_ip46_routed_to_bridged_and_back(False, True,
- self.WITHOUT_EH)
+ """ACL IPv6 routed -> bridged, L3 ACL permit+reflect"""
+ self.run_test_ip46_routed_to_bridged_and_back(False, True, self.WITHOUT_EH)
def test_0112_ip6_irb_1(self):
- """ ACL IPv6 bridged -> routed, L3 ACL permit+reflect"""
- self.run_test_ip46_bridged_to_routed_and_back(False, True,
- self.WITHOUT_EH)
+ """ACL IPv6 bridged -> routed, L3 ACL permit+reflect"""
+ self.run_test_ip46_bridged_to_routed_and_back(False, True, self.WITHOUT_EH)
def test_0113_ip6_irb_1(self):
- """ ACL IPv4 routed -> bridged, L3 ACL permit+reflect"""
- self.run_test_ip46_routed_to_bridged_and_back(False, False,
- self.WITHOUT_EH)
+ """ACL IPv4 routed -> bridged, L3 ACL permit+reflect"""
+ self.run_test_ip46_routed_to_bridged_and_back(False, False, self.WITHOUT_EH)
def test_0114_ip6_irb_1(self):
- """ ACL IPv4 bridged -> routed, L3 ACL permit+reflect"""
- self.run_test_ip46_bridged_to_routed_and_back(False, False,
- self.WITHOUT_EH)
+ """ACL IPv4 bridged -> routed, L3 ACL permit+reflect"""
+ self.run_test_ip46_bridged_to_routed_and_back(False, False, self.WITHOUT_EH)
# A block of tests with extension headers
def test_1001_ip6_irb_1(self):
- """ ACL IPv6+EH routed -> bridged, L2 ACL deny"""
- self.run_test_ip46_routed_to_bridged(True, True, False,
- self.WITH_EH)
+ """ACL IPv6+EH routed -> bridged, L2 ACL deny"""
+ self.run_test_ip46_routed_to_bridged(True, True, False, self.WITH_EH)
def test_1002_ip6_irb_1(self):
- """ ACL IPv6+EH routed -> bridged, L3 ACL deny"""
- self.run_test_ip46_routed_to_bridged(False, True, False,
- self.WITH_EH)
+ """ACL IPv6+EH routed -> bridged, L3 ACL deny"""
+ self.run_test_ip46_routed_to_bridged(False, True, False, self.WITH_EH)
def test_1005_ip6_irb_1(self):
- """ ACL IPv6+EH bridged -> routed, L2 ACL deny """
- self.run_test_ip46_bridged_to_routed(True, True, False,
- self.WITH_EH)
+ """ACL IPv6+EH bridged -> routed, L2 ACL deny"""
+ self.run_test_ip46_bridged_to_routed(True, True, False, self.WITH_EH)
def test_1006_ip6_irb_1(self):
- """ ACL IPv6+EH bridged -> routed, L3 ACL deny """
- self.run_test_ip46_bridged_to_routed(False, True, False,
- self.WITH_EH)
+ """ACL IPv6+EH bridged -> routed, L3 ACL deny"""
+ self.run_test_ip46_bridged_to_routed(False, True, False, self.WITH_EH)
def test_1101_ip6_irb_1(self):
- """ ACL IPv6+EH routed -> bridged, L2 ACL permit+reflect"""
- self.run_test_ip46_routed_to_bridged_and_back(True, True,
- self.WITH_EH)
+ """ACL IPv6+EH routed -> bridged, L2 ACL permit+reflect"""
+ self.run_test_ip46_routed_to_bridged_and_back(True, True, self.WITH_EH)
def test_1102_ip6_irb_1(self):
- """ ACL IPv6+EH bridged -> routed, L2 ACL permit+reflect"""
- self.run_test_ip46_bridged_to_routed_and_back(True, True,
- self.WITH_EH)
+ """ACL IPv6+EH bridged -> routed, L2 ACL permit+reflect"""
+ self.run_test_ip46_bridged_to_routed_and_back(True, True, self.WITH_EH)
def test_1111_ip6_irb_1(self):
- """ ACL IPv6+EH routed -> bridged, L3 ACL permit+reflect"""
- self.run_test_ip46_routed_to_bridged_and_back(False, True,
- self.WITH_EH)
+ """ACL IPv6+EH routed -> bridged, L3 ACL permit+reflect"""
+ self.run_test_ip46_routed_to_bridged_and_back(False, True, self.WITH_EH)
def test_1112_ip6_irb_1(self):
- """ ACL IPv6+EH bridged -> routed, L3 ACL permit+reflect"""
- self.run_test_ip46_bridged_to_routed_and_back(False, True,
- self.WITH_EH)
+ """ACL IPv6+EH bridged -> routed, L3 ACL permit+reflect"""
+ self.run_test_ip46_bridged_to_routed_and_back(False, True, self.WITH_EH)
# IPv4 with "MF" bit set
def test_1201_ip6_irb_1(self):
- """ ACL IPv4+MF routed -> bridged, L2 ACL deny"""
- self.run_test_ip46_routed_to_bridged(True, False, False,
- self.WITH_EH)
+ """ACL IPv4+MF routed -> bridged, L2 ACL deny"""
+ self.run_test_ip46_routed_to_bridged(True, False, False, self.WITH_EH)
def test_1202_ip6_irb_1(self):
- """ ACL IPv4+MF routed -> bridged, L3 ACL deny"""
- self.run_test_ip46_routed_to_bridged(False, False, False,
- self.WITH_EH)
+ """ACL IPv4+MF routed -> bridged, L3 ACL deny"""
+ self.run_test_ip46_routed_to_bridged(False, False, False, self.WITH_EH)
def test_1205_ip6_irb_1(self):
- """ ACL IPv4+MF bridged -> routed, L2 ACL deny """
- self.run_test_ip46_bridged_to_routed(True, False, False,
- self.WITH_EH)
+ """ACL IPv4+MF bridged -> routed, L2 ACL deny"""
+ self.run_test_ip46_bridged_to_routed(True, False, False, self.WITH_EH)
def test_1206_ip6_irb_1(self):
- """ ACL IPv4+MF bridged -> routed, L3 ACL deny """
- self.run_test_ip46_bridged_to_routed(False, False, False,
- self.WITH_EH)
+ """ACL IPv4+MF bridged -> routed, L3 ACL deny"""
+ self.run_test_ip46_bridged_to_routed(False, False, False, self.WITH_EH)
def test_1301_ip6_irb_1(self):
- """ ACL IPv4+MF routed -> bridged, L2 ACL permit+reflect"""
- self.run_test_ip46_routed_to_bridged_and_back(True, False,
- self.WITH_EH)
+ """ACL IPv4+MF routed -> bridged, L2 ACL permit+reflect"""
+ self.run_test_ip46_routed_to_bridged_and_back(True, False, self.WITH_EH)
def test_1302_ip6_irb_1(self):
- """ ACL IPv4+MF bridged -> routed, L2 ACL permit+reflect"""
- self.run_test_ip46_bridged_to_routed_and_back(True, False,
- self.WITH_EH)
+ """ACL IPv4+MF bridged -> routed, L2 ACL permit+reflect"""
+ self.run_test_ip46_bridged_to_routed_and_back(True, False, self.WITH_EH)
def test_1311_ip6_irb_1(self):
- """ ACL IPv4+MF routed -> bridged, L3 ACL permit+reflect"""
- self.run_test_ip46_routed_to_bridged_and_back(False, False,
- self.WITH_EH)
+ """ACL IPv4+MF routed -> bridged, L3 ACL permit+reflect"""
+ self.run_test_ip46_routed_to_bridged_and_back(False, False, self.WITH_EH)
def test_1312_ip6_irb_1(self):
- """ ACL IPv4+MF bridged -> routed, L3 ACL permit+reflect"""
- self.run_test_ip46_bridged_to_routed_and_back(False, False,
- self.WITH_EH)
+ """ACL IPv4+MF bridged -> routed, L3 ACL permit+reflect"""
+ self.run_test_ip46_bridged_to_routed_and_back(False, False, self.WITH_EH)
+
# Stateful ACL tests with stateful ICMP
def test_1401_ip6_irb_1(self):
- """ IPv6 routed -> bridged, L2 ACL permit+reflect, ICMP reflect"""
- self.run_test_ip46_routed_to_bridged_and_back(True, True,
- self.WITHOUT_EH,
- self.STATEFUL_ICMP)
+ """IPv6 routed -> bridged, L2 ACL permit+reflect, ICMP reflect"""
+ self.run_test_ip46_routed_to_bridged_and_back(
+ True, True, self.WITHOUT_EH, self.STATEFUL_ICMP
+ )
def test_1402_ip6_irb_1(self):
- """ IPv6 bridged -> routed, L2 ACL permit+reflect, ICMP reflect"""
- self.run_test_ip46_bridged_to_routed_and_back(True, True,
- self.WITHOUT_EH,
- self.STATEFUL_ICMP)
+ """IPv6 bridged -> routed, L2 ACL permit+reflect, ICMP reflect"""
+ self.run_test_ip46_bridged_to_routed_and_back(
+ True, True, self.WITHOUT_EH, self.STATEFUL_ICMP
+ )
def test_1403_ip4_irb_1(self):
- """ IPv4 routed -> bridged, L2 ACL permit+reflect, ICMP reflect"""
- self.run_test_ip46_routed_to_bridged_and_back(True, False,
- self.WITHOUT_EH,
- self.STATEFUL_ICMP)
+ """IPv4 routed -> bridged, L2 ACL permit+reflect, ICMP reflect"""
+ self.run_test_ip46_routed_to_bridged_and_back(
+ True, False, self.WITHOUT_EH, self.STATEFUL_ICMP
+ )
def test_1404_ip4_irb_1(self):
- """ IPv4 bridged -> routed, L2 ACL permit+reflect, ICMP reflect"""
- self.run_test_ip46_bridged_to_routed_and_back(True, False,
- self.WITHOUT_EH,
- self.STATEFUL_ICMP)
+ """IPv4 bridged -> routed, L2 ACL permit+reflect, ICMP reflect"""
+ self.run_test_ip46_bridged_to_routed_and_back(
+ True, False, self.WITHOUT_EH, self.STATEFUL_ICMP
+ )
def test_1411_ip6_irb_1(self):
- """ IPv6 routed -> bridged, L3 ACL permit+reflect, ICMP reflect"""
- self.run_test_ip46_routed_to_bridged_and_back(False, True,
- self.WITHOUT_EH,
- self.STATEFUL_ICMP)
+ """IPv6 routed -> bridged, L3 ACL permit+reflect, ICMP reflect"""
+ self.run_test_ip46_routed_to_bridged_and_back(
+ False, True, self.WITHOUT_EH, self.STATEFUL_ICMP
+ )
def test_1412_ip6_irb_1(self):
- """ IPv6 bridged -> routed, L3 ACL permit+reflect, ICMP reflect"""
- self.run_test_ip46_bridged_to_routed_and_back(False, True,
- self.WITHOUT_EH,
- self.STATEFUL_ICMP)
+ """IPv6 bridged -> routed, L3 ACL permit+reflect, ICMP reflect"""
+ self.run_test_ip46_bridged_to_routed_and_back(
+ False, True, self.WITHOUT_EH, self.STATEFUL_ICMP
+ )
def test_1413_ip4_irb_1(self):
- """ IPv4 routed -> bridged, L3 ACL permit+reflect, ICMP reflect"""
- self.run_test_ip46_routed_to_bridged_and_back(False, False,
- self.WITHOUT_EH,
- self.STATEFUL_ICMP)
+ """IPv4 routed -> bridged, L3 ACL permit+reflect, ICMP reflect"""
+ self.run_test_ip46_routed_to_bridged_and_back(
+ False, False, self.WITHOUT_EH, self.STATEFUL_ICMP
+ )
def test_1414_ip4_irb_1(self):
- """ IPv4 bridged -> routed, L3 ACL permit+reflect, ICMP reflect"""
- self.run_test_ip46_bridged_to_routed_and_back(False, False,
- self.WITHOUT_EH,
- self.STATEFUL_ICMP)
+ """IPv4 bridged -> routed, L3 ACL permit+reflect, ICMP reflect"""
+ self.run_test_ip46_bridged_to_routed_and_back(
+ False, False, self.WITHOUT_EH, self.STATEFUL_ICMP
+ )
-if __name__ == '__main__':
+if __name__ == "__main__":
unittest.main(testRunner=VppTestRunner)