test/vpp_srv6.py

   1 """
   2   SRv6 LocalSIDs
   3
   4   object abstractions for representing SRv6 localSIDs in VPP
   5 """
   6
   7 from vpp_object import VppObject
   8 from socket import inet_pton, inet_ntop, AF_INET, AF_INET6
   9 import copy
  10
  11
  12 class SRv6LocalSIDBehaviors:
  13     # from src/vnet/srv6/sr.h
  14     SR_BEHAVIOR_END = 1
  15     SR_BEHAVIOR_X = 2
  16     SR_BEHAVIOR_T = 3
  17     SR_BEHAVIOR_D_FIRST = 4  # Unused. Separator in between regular and D
  18     SR_BEHAVIOR_DX2 = 5
  19     SR_BEHAVIOR_DX6 = 6
  20     SR_BEHAVIOR_DX4 = 7
  21     SR_BEHAVIOR_DT6 = 8
  22     SR_BEHAVIOR_DT4 = 9
  23     SR_BEHAVIOR_END_UN_PERF = 10
  24     SR_BEHAVIOR_END_UN = 11
  25     SR_BEHAVIOR_LAST = 12  # Must always be the last one
  26
  27
  28 class SRv6PolicyType:
  29     # from src/vnet/srv6/sr.h
  30     SR_POLICY_TYPE_DEFAULT = 0
  31     SR_POLICY_TYPE_SPRAY = 1
  32     SR_POLICY_TYPE_TEF = 2
  33
  34
  35 class SRv6PolicySteeringTypes:
  36     # from src/vnet/srv6/sr.h
  37     SR_STEER_L2 = 2
  38     SR_STEER_IPV4 = 4
  39     SR_STEER_IPV6 = 6
  40
  41
  42 class VppSRv6LocalSID(VppObject):
  43     """
  44     SRv6 LocalSID
  45     """
  46
  47     def __init__(
  48         self,
  49         test,
  50         localsid,
  51         behavior,
  52         nh_addr,
  53         end_psp,
  54         sw_if_index,
  55         vlan_index,
  56         fib_table,
  57     ):
  58         self._test = test
  59         self.localsid = localsid
  60         self.behavior = behavior
  61         self.nh_addr = nh_addr
  62         self.end_psp = end_psp
  63         self.sw_if_index = sw_if_index
  64         self.vlan_index = vlan_index
  65         self.fib_table = fib_table
  66         self._configured = False
  67
  68     def add_vpp_config(self):
  69         self._test.vapi.sr_localsid_add_del(
  70             localsid=self.localsid,
  71             behavior=self.behavior,
  72             nh_addr=self.nh_addr,
  73             is_del=0,
  74             end_psp=self.end_psp,
  75             sw_if_index=self.sw_if_index,
  76             vlan_index=self.vlan_index,
  77             fib_table=self.fib_table,
  78         )
  79         self._configured = True
  80
  81     def remove_vpp_config(self):
  82         self._test.vapi.sr_localsid_add_del(
  83             localsid=self.localsid,
  84             behavior=self.behavior,
  85             nh_addr=self.nh_addr,
  86             is_del=1,
  87             end_psp=self.end_psp,
  88             sw_if_index=self.sw_if_index,
  89             vlan_index=self.vlan_index,
  90             fib_table=self.fib_table,
  91         )
  92         self._configured = False
  93
  94     def query_vpp_config(self):
  95         # sr_localsids_dump API is disabled
  96         # use _configured flag for now
  97         return self._configured
  98
  99     def object_id(self):
 100         return "%d;%s,%d" % (self.fib_table, self.localsid, self.behavior)
 101
 102
 103 class VppSRv6Policy(VppObject):
 104     """
 105     SRv6 Policy
 106     """
 107
 108     def __init__(
 109         self, test, bsid, is_encap, sr_type, weight, fib_table, segments, source
 110     ):
 111         self._test = test
 112         self.bsid = bsid
 113         self.is_encap = is_encap
 114         self.sr_type = sr_type
 115         self.weight = weight
 116         self.fib_table = fib_table
 117         self.segments = segments
 118         self.n_segments = len(segments)
 119         # source not passed to API
 120         # self.source = inet_pton(AF_INET6, source)
 121         self.source = source
 122         self._configured = False
 123
 124     def add_vpp_config(self):
 125         self._test.vapi.sr_policy_add(
 126             bsid=self.bsid,
 127             weight=self.weight,
 128             is_encap=self.is_encap,
 129             is_spray=self.sr_type,
 130             fib_table=self.fib_table,
 131             sids={"num_sids": self.n_segments, "sids": self.segments},
 132         )
 133         self._configured = True
 134
 135     def remove_vpp_config(self):
 136         self._test.vapi.sr_policy_del(self.bsid)
 137         self._configured = False
 138
 139     def query_vpp_config(self):
 140         # no API to query SR Policies
 141         # use _configured flag for now
 142         return self._configured
 143
 144     def object_id(self):
 145         return "%d;%s-><%s>;%d" % (
 146             self.sr_type,
 147             self.bsid,
 148             ",".join(self.segments),
 149             self.is_encap,
 150         )
 151
 152
 153 class VppSRv6PolicyV2(VppObject):
 154     """
 155     SRv6 Policy
 156     """
 157
 158     def __init__(
 159         self,
 160         test,
 161         bsid,
 162         is_encap,
 163         sr_type,
 164         weight,
 165         fib_table,
 166         segments,
 167         encap_src,
 168         source,
 169     ):
 170         self._test = test
 171         self.bsid = bsid
 172         self.is_encap = is_encap
 173         self.sr_type = sr_type
 174         self.weight = weight
 175         self.fib_table = fib_table
 176         self.segments = segments
 177         self.encap_src = encap_src
 178         self.n_segments = len(segments)
 179
 180         # source not passed to API
 181         # self.source = inet_pton(AF_INET6, source)
 182         self.source = source
 183         self._configured = False
 184
 185     def add_vpp_config(self):
 186         self._test.vapi.sr_policy_add_v2(
 187             bsid_addr=self.bsid,
 188             weight=self.weight,
 189             is_encap=self.is_encap,
 190             type=self.sr_type,
 191             fib_table=self.fib_table,
 192             encap_src=self.encap_src,
 193             sids={
 194                 "num_sids": self.n_segments,
 195                 "sids": self._get_fixed_segments(),
 196                 "weight": 1,
 197             },
 198         )
 199         self._configured = True
 200
 201     def remove_vpp_config(self):
 202         self._test.vapi.sr_policy_del(self.bsid)
 203         self._configured = False
 204
 205     def query_vpp_config(self):
 206         # no API to query SR Policies
 207         # use _configured flag for now
 208         return self._configured
 209
 210     def object_id(self):
 211         return "%d;%s-><%s>;%d" % (
 212             self.sr_type,
 213             self.bsid,
 214             ",".join(self.segments),
 215             self.is_encap,
 216         )
 217
 218     def _get_fixed_segments(self):
 219         segs = copy.copy(self.segments)
 220         # note: array expect size is 16
 221         for _ in range(16 - self.n_segments):
 222             segs.append("")
 223         return segs
 224
 225
 226 class VppSRv6Steering(VppObject):
 227     """
 228     SRv6 Steering
 229     """
 230
 231     def __init__(
 232         self,
 233         test,
 234         bsid,
 235         prefix,
 236         mask_width,
 237         traffic_type,
 238         sr_policy_index,
 239         table_id,
 240         sw_if_index,
 241     ):
 242         self._test = test
 243         self.bsid = bsid
 244         self.prefix = prefix
 245         self.mask_width = mask_width
 246         self.traffic_type = traffic_type
 247         self.sr_policy_index = sr_policy_index
 248         self.sw_if_index = sw_if_index
 249         self.table_id = table_id
 250         self._configured = False
 251
 252     def add_vpp_config(self):
 253         self._test.vapi.sr_steering_add_del(
 254             is_del=0,
 255             bsid_addr=self.bsid,
 256             sr_policy_index=self.sr_policy_index,
 257             table_id=self.table_id,
 258             prefix={"address": self.prefix, "len": self.mask_width},
 259             sw_if_index=self.sw_if_index,
 260             traffic_type=self.traffic_type,
 261         )
 262         self._configured = True
 263
 264     def remove_vpp_config(self):
 265         self._test.vapi.sr_steering_add_del(
 266             is_del=1,
 267             bsid_addr=self.bsid,
 268             sr_policy_index=self.sr_policy_index,
 269             table_id=self.table_id,
 270             prefix={"address": self.prefix, "len": self.mask_width},
 271             sw_if_index=self.sw_if_index,
 272             traffic_type=self.traffic_type,
 273         )
 274         self._configured = False
 275
 276     def query_vpp_config(self):
 277         # no API to query steering entries
 278         # use _configured flag for now
 279         return self._configured
 280
 281     def object_id(self):
 282         return "%d;%d;%s/%d->%s" % (
 283             self.table_id,
 284             self.traffic_type,
 285             self.prefix,
 286             self.mask_width,
 287             self.bsid,
 288         )