diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/test_srv6.py | 116 | ||||
| -rw-r--r-- | test/vpp_srv6.py | 79 |
2 files changed, 193 insertions, 2 deletions
diff --git a/test/test_srv6.py b/test/test_srv6.py index 40b53375ddb..a15c69713a5 100644 --- a/test/test_srv6.py +++ b/test/test_srv6.py @@ -10,6 +10,7 @@ from vpp_srv6 import ( SRv6LocalSIDBehaviors, VppSRv6LocalSID, VppSRv6Policy, + VppSRv6PolicyV2, SRv6PolicyType, VppSRv6Steering, SRv6PolicySteeringTypes, @@ -248,6 +249,121 @@ class TestSRv6(VppTestCase): # cleanup interfaces self.teardown_interfaces() + def test_SRv6_T_Encaps_with_v6src(self): + """Test SRv6 Transit.Encaps behavior for IPv6 and select multiple src v6addr case.""" + # send traffic to one destination interface + # source and destination are IPv6 only + self.setup_interfaces(ipv6=[True, True]) + + # configure FIB entries + route = VppIpRoute( + self, "a4::", 64, [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)] + ) + route.add_vpp_config() + + # configure encaps IPv6 source address + # needs to be done before SR Policy config + # TODO: API? + self.vapi.cli("set sr encaps source addr a3::") + + bsid = "a3::9999:1" + other_src_ip = "b1::" + # configure SRv6 Policy + # Note: segment list order: first -> last + sr_policy = VppSRv6PolicyV2( + self, + bsid=bsid, + is_encap=1, + sr_type=SRv6PolicyType.SR_POLICY_TYPE_DEFAULT, + weight=1, + fib_table=0, + segments=["a4::", "a5::", "a6::c7"], + encap_src=other_src_ip, + source=other_src_ip, + ) + sr_policy.add_vpp_config() + self.sr_policy = sr_policy + + # log the sr policies + self.logger.info(self.vapi.cli("show sr policies")) + + # steer IPv6 traffic to a7::/64 into SRv6 Policy + # use the bsid of the above self.sr_policy + pol_steering = VppSRv6Steering( + self, + bsid=self.sr_policy.bsid, + prefix="a7::", + mask_width=64, + traffic_type=SRv6PolicySteeringTypes.SR_STEER_IPV6, + sr_policy_index=0, + table_id=0, + sw_if_index=0, + ) + pol_steering.add_vpp_config() + + # log the sr steering policies + self.logger.info(self.vapi.cli("show sr steering-policies")) + + # create packets + count = len(self.pg_packet_sizes) + dst_inner = "a7::1234" + pkts = [] + + # create IPv6 packets without SRH + packet_header = self.create_packet_header_IPv6(dst_inner) + # create traffic stream pg0->pg1 + pkts.extend( + self.create_stream( + self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count + ) + ) + + # create IPv6 packets with SRH + # packets with segments-left 1, active segment a7:: + packet_header = self.create_packet_header_IPv6_SRH( + sidlist=["a8::", "a7::", "a6::"], segleft=1 + ) + # create traffic stream pg0->pg1 + pkts.extend( + self.create_stream( + self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count + ) + ) + + # create IPv6 packets with SRH and IPv6 + # packets with segments-left 1, active segment a7:: + packet_header = self.create_packet_header_IPv6_SRH_IPv6( + dst_inner, sidlist=["a8::", "a7::", "a6::"], segleft=1 + ) + # create traffic stream pg0->pg1 + pkts.extend( + self.create_stream( + self.pg0, self.pg1, packet_header, self.pg_packet_sizes, count + ) + ) + + # send packets and verify received packets + self.send_and_verify_pkts( + self.pg0, pkts, self.pg1, self.compare_rx_tx_packet_T_Encaps + ) + + # log the localsid counters + self.logger.info(self.vapi.cli("show sr localsid")) + + # remove SR steering + pol_steering.remove_vpp_config() + self.logger.info(self.vapi.cli("show sr steering-policies")) + + # remove SR Policies + self.sr_policy.remove_vpp_config() + self.logger.info(self.vapi.cli("show sr policies")) + + # remove FIB entries + # done by tearDown + + # cleanup interfaces + self.teardown_interfaces() + @unittest.skipUnless(0, "PC to fix") def test_SRv6_T_Insert(self): """Test SRv6 Transit.Insert behavior (IPv6 only).""" diff --git a/test/vpp_srv6.py b/test/vpp_srv6.py index d789105d7a7..1b09103f297 100644 --- a/test/vpp_srv6.py +++ b/test/vpp_srv6.py @@ -6,6 +6,7 @@ from vpp_object import VppObject from socket import inet_pton, inet_ntop, AF_INET, AF_INET6 +import copy class SRv6LocalSIDBehaviors: @@ -28,6 +29,7 @@ class SRv6PolicyType: # from src/vnet/srv6/sr.h SR_POLICY_TYPE_DEFAULT = 0 SR_POLICY_TYPE_SPRAY = 1 + SR_POLICY_TYPE_TEF = 2 class SRv6PolicySteeringTypes: @@ -148,6 +150,79 @@ class VppSRv6Policy(VppObject): ) +class VppSRv6PolicyV2(VppObject): + """ + SRv6 Policy + """ + + def __init__( + self, + test, + bsid, + is_encap, + sr_type, + weight, + fib_table, + segments, + encap_src, + source, + ): + self._test = test + self.bsid = bsid + self.is_encap = is_encap + self.sr_type = sr_type + self.weight = weight + self.fib_table = fib_table + self.segments = segments + self.encap_src = encap_src + self.n_segments = len(segments) + + # source not passed to API + # self.source = inet_pton(AF_INET6, source) + self.source = source + self._configured = False + + def add_vpp_config(self): + self._test.vapi.sr_policy_add_v2( + bsid_addr=self.bsid, + weight=self.weight, + is_encap=self.is_encap, + type=self.sr_type, + fib_table=self.fib_table, + encap_src=self.encap_src, + sids={ + "num_sids": self.n_segments, + "sids": self._get_fixed_segments(), + "weight": 1, + }, + ) + self._configured = True + + def remove_vpp_config(self): + self._test.vapi.sr_policy_del(self.bsid) + self._configured = False + + def query_vpp_config(self): + # no API to query SR Policies + # use _configured flag for now + return self._configured + + def object_id(self): + return "%d;%s-><%s>;%d" % ( + self.sr_type, + self.bsid, + ",".join(self.segments), + self.is_encap, + ) + + def _get_fixed_segments(self): + segs = copy.copy(self.segments) + # note: array expect size is 16 + for _ in range(16 - self.n_segments): + segs.append("") + return segs + + class VppSRv6Steering(VppObject): """ SRv6 Steering @@ -177,7 +252,7 @@ class VppSRv6Steering(VppObject): def add_vpp_config(self): self._test.vapi.sr_steering_add_del( is_del=0, - bsid=self.bsid, + bsid_addr=self.bsid, sr_policy_index=self.sr_policy_index, table_id=self.table_id, prefix={"address": self.prefix, "len": self.mask_width}, @@ -189,7 +264,7 @@ class VppSRv6Steering(VppObject): def remove_vpp_config(self): self._test.vapi.sr_steering_add_del( is_del=1, - bsid=self.bsid, + bsid_addr=self.bsid, sr_policy_index=self.sr_policy_index, table_id=self.table_id, prefix={"address": self.prefix, "len": self.mask_width}, |