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 *
   8 from socket import inet_pton, inet_ntop, AF_INET, AF_INET6
   9
  10
  11 class SRv6LocalSIDBehaviors():
  12     # from src/vnet/srv6/sr.h
  13     SR_BEHAVIOR_END = 1
  14     SR_BEHAVIOR_X = 2
  15     SR_BEHAVIOR_T = 3
  16     SR_BEHAVIOR_D_FIRST = 4   # Unused. Separator in between regular and D
  17     SR_BEHAVIOR_DX2 = 5
  18     SR_BEHAVIOR_DX6 = 6
  19     SR_BEHAVIOR_DX4 = 7
  20     SR_BEHAVIOR_DT6 = 8
  21     SR_BEHAVIOR_DT4 = 9
  22     SR_BEHAVIOR_LAST = 10      # Must always be the last one
  23
  24
  25 class SRv6PolicyType():
  26     # from src/vnet/srv6/sr.h
  27     SR_POLICY_TYPE_DEFAULT = 0
  28     SR_POLICY_TYPE_SPRAY = 1
  29
  30
  31 class SRv6PolicySteeringTypes():
  32     # from src/vnet/srv6/sr.h
  33     SR_STEER_L2 = 2
  34     SR_STEER_IPV4 = 4
  35     SR_STEER_IPV6 = 6
  36
  37
  38 class VppSRv6LocalSID(VppObject):
  39     """
  40     SRv6 LocalSID
  41     """
  42
  43     def __init__(self, test, localsid, behavior, nh_addr4, nh_addr6,
  44                  end_psp, sw_if_index, vlan_index, fib_table):
  45         self._test = test
  46         self.localsid = localsid
  47         # keep binary format in _localsid
  48         self.localsid["addr"] = inet_pton(AF_INET6, self.localsid["addr"])
  49         self.behavior = behavior
  50         self.nh_addr4 = inet_pton(AF_INET, nh_addr4)
  51         self.nh_addr6 = inet_pton(AF_INET6, nh_addr6)
  52         self.end_psp = end_psp
  53         self.sw_if_index = sw_if_index
  54         self.vlan_index = vlan_index
  55         self.fib_table = fib_table
  56         self._configured = False
  57
  58     def add_vpp_config(self):
  59         self._test.vapi.sr_localsid_add_del(
  60             self.localsid,
  61             self.behavior,
  62             self.nh_addr4,
  63             self.nh_addr6,
  64             is_del=0,
  65             end_psp=self.end_psp,
  66             sw_if_index=self.sw_if_index,
  67             vlan_index=self.vlan_index,
  68             fib_table=self.fib_table)
  69         self._configured = True
  70
  71     def remove_vpp_config(self):
  72         self._test.vapi.sr_localsid_add_del(
  73             self.localsid,
  74             self.behavior,
  75             self.nh_addr4,
  76             self.nh_addr6,
  77             is_del=1,
  78             end_psp=self.end_psp,
  79             sw_if_index=self.sw_if_index,
  80             vlan_index=self.vlan_index,
  81             fib_table=self.fib_table)
  82         self._configured = False
  83
  84     def query_vpp_config(self):
  85         # sr_localsids_dump API is disabled
  86         # use _configured flag for now
  87         return self._configured
  88
  89     def __str__(self):
  90         return self.object_id()
  91
  92     def object_id(self):
  93         return ("%d;%s,%d"
  94                 % (self.fib_table,
  95                    self.localsid,
  96                    self.behavior))
  97
  98
  99 class VppSRv6Policy(VppObject):
 100     """
 101     SRv6 Policy
 102     """
 103
 104     def __init__(self, test, bsid,
 105                  is_encap, sr_type, weight, fib_table,
 106                  segments, source):
 107         self._test = test
 108         self.bsid = bsid
 109         # keep binary format in _bsid
 110         self._bsid = inet_pton(AF_INET6, bsid)
 111         self.is_encap = is_encap
 112         self.sr_type = sr_type
 113         self.weight = weight
 114         self.fib_table = fib_table
 115         self.segments = segments
 116         # keep binary format in _segments
 117         self._segments = []
 118         for seg in segments:
 119             self._segments.extend(inet_pton(AF_INET6, seg))
 120         self.n_segments = len(segments)
 121         # source not passed to API
 122         # self.source = inet_pton(AF_INET6, source)
 123         self.source = source
 124         self._configured = False
 125
 126     def add_vpp_config(self):
 127         self._test.vapi.sr_policy_add(
 128                      self._bsid,
 129                      self.weight,
 130                      self.is_encap,
 131                      self.sr_type,
 132                      self.fib_table,
 133                      self.n_segments,
 134                      self._segments)
 135         self._configured = True
 136
 137     def remove_vpp_config(self):
 138         self._test.vapi.sr_policy_del(
 139                      self._bsid)
 140         self._configured = False
 141
 142     def query_vpp_config(self):
 143         # no API to query SR Policies
 144         # use _configured flag for now
 145         return self._configured
 146
 147     def __str__(self):
 148         return self.object_id()
 149
 150     def object_id(self):
 151         return ("%d;%s-><%s>;%d"
 152                 % (self.sr_type,
 153                    self.bsid,
 154                    ','.join(self.segments),
 155                    self.is_encap))
 156
 157
 158 class VppSRv6Steering(VppObject):
 159     """
 160     SRv6 Steering
 161     """
 162
 163     def __init__(self, test,
 164                  bsid,
 165                  prefix,
 166                  mask_width,
 167                  traffic_type,
 168                  sr_policy_index,
 169                  table_id,
 170                  sw_if_index):
 171         self._test = test
 172         self.bsid = bsid
 173         # keep binary format in _bsid
 174         self._bsid = inet_pton(AF_INET6, bsid)
 175         self.prefix = prefix
 176         # keep binary format in _prefix
 177         if ':' in prefix:
 178             # IPv6
 179             self._prefix = inet_pton(AF_INET6, prefix)
 180         else:
 181             # IPv4
 182             # API expects 16 octets (128 bits)
 183             # last 4 octets are used for IPv4
 184             # --> prepend 12 octets
 185             self._prefix = ('\x00' * 12) + inet_pton(AF_INET, prefix)
 186         self.mask_width = mask_width
 187         self.traffic_type = traffic_type
 188         self.sr_policy_index = sr_policy_index
 189         self.sw_if_index = sw_if_index
 190         self.table_id = table_id
 191         self._configured = False
 192
 193     def add_vpp_config(self):
 194         self._test.vapi.sr_steering_add_del(
 195                      0,
 196                      self._bsid,
 197                      self.sr_policy_index,
 198                      self.table_id,
 199                      self._prefix,
 200                      self.mask_width,
 201                      self.sw_if_index,
 202                      self.traffic_type)
 203         self._configured = True
 204
 205     def remove_vpp_config(self):
 206         self._test.vapi.sr_steering_add_del(
 207                      1,
 208                      self._bsid,
 209                      self.sr_policy_index,
 210                      self.table_id,
 211                      self._prefix,
 212                      self.mask_width,
 213                      self.sw_if_index,
 214                      self.traffic_type)
 215         self._configured = False
 216
 217     def query_vpp_config(self):
 218         # no API to query steering entries
 219         # use _configured flag for now
 220         return self._configured
 221
 222     def __str__(self):
 223         return self.object_id()
 224
 225     def object_id(self):
 226         return ("%d;%d;%s/%d->%s"
 227                 % (self.table_id,
 228                    self.traffic_type,
 229                    self.prefix,
 230                    self.mask_width,
 231                    self.bsid))