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_addr, behavior, nh_addr, end_psp,
  44                  sw_if_index, vlan_index, fib_table):
  45         self._test = test
  46         self.localsid_addr = localsid_addr
  47         # keep binary format in _localsid_addr
  48         self._localsid_addr = inet_pton(AF_INET6, self.localsid_addr)
  49         self.behavior = behavior
  50         self.nh_addr = nh_addr
  51         # keep binary format in _nh_addr
  52         if ':' in nh_addr:
  53             # IPv6
  54             self._nh_addr = inet_pton(AF_INET6, nh_addr)
  55         else:
  56             # IPv4
  57             # API expects 16 octets (128 bits)
  58             # last 4 octets are used for IPv4
  59             # --> prepend 12 octets
  60             self._nh_addr = ('\x00' * 12) + inet_pton(AF_INET, nh_addr)
  61         self.end_psp = end_psp
  62         self.sw_if_index = sw_if_index
  63         self.vlan_index = vlan_index
  64         self.fib_table = fib_table
  65         self._configured = False
  66
  67     def add_vpp_config(self):
  68         self._test.vapi.sr_localsid_add_del(
  69             self._localsid_addr,
  70             self.behavior,
  71             self._nh_addr,
  72             is_del=0,
  73             end_psp=self.end_psp,
  74             sw_if_index=self.sw_if_index,
  75             vlan_index=self.vlan_index,
  76             fib_table=self.fib_table)
  77         self._configured = True
  78
  79     def remove_vpp_config(self):
  80         self._test.vapi.sr_localsid_add_del(
  81             self._localsid_addr,
  82             self.behavior,
  83             self._nh_addr,
  84             is_del=1,
  85             end_psp=self.end_psp,
  86             sw_if_index=self.sw_if_index,
  87             vlan_index=self.vlan_index,
  88             fib_table=self.fib_table)
  89         self._configured = False
  90
  91     def query_vpp_config(self):
  92         # sr_localsids_dump API is disabled
  93         # use _configured flag for now
  94         return self._configured
  95
  96     def __str__(self):
  97         return self.object_id()
  98
  99     def object_id(self):
 100         return ("%d;%s,%d"
 101                 % (self.fib_table,
 102                    self.localsid_addr,
 103                    self.behavior))
 104
 105
 106 class VppSRv6Policy(VppObject):
 107     """
 108     SRv6 Policy
 109     """
 110
 111     def __init__(self, test, bsid,
 112                  is_encap, sr_type, weight, fib_table,
 113                  segments, source):
 114         self._test = test
 115         self.bsid = bsid
 116         # keep binary format in _bsid
 117         self._bsid = inet_pton(AF_INET6, bsid)
 118         self.is_encap = is_encap
 119         self.sr_type = sr_type
 120         self.weight = weight
 121         self.fib_table = fib_table
 122         self.segments = segments
 123         # keep binary format in _segments
 124         self._segments = []
 125         for seg in segments:
 126             self._segments.extend(inet_pton(AF_INET6, seg))
 127         self.n_segments = len(segments)
 128         # source not passed to API
 129         # self.source = inet_pton(AF_INET6, source)
 130         self.source = source
 131         self._configured = False
 132
 133     def add_vpp_config(self):
 134         self._test.vapi.sr_policy_add(
 135                      self._bsid,
 136                      self.weight,
 137                      self.is_encap,
 138                      self.sr_type,
 139                      self.fib_table,
 140                      self.n_segments,
 141                      self._segments)
 142         self._configured = True
 143
 144     def remove_vpp_config(self):
 145         self._test.vapi.sr_policy_del(
 146                      self._bsid)
 147         self._configured = False
 148
 149     def query_vpp_config(self):
 150         # no API to query SR Policies
 151         # use _configured flag for now
 152         return self._configured
 153
 154     def __str__(self):
 155         return self.object_id()
 156
 157     def object_id(self):
 158         return ("%d;%s-><%s>;%d"
 159                 % (self.sr_type,
 160                    self.bsid,
 161                    ','.join(self.segments),
 162                    self.is_encap))
 163
 164
 165 class VppSRv6Steering(VppObject):
 166     """
 167     SRv6 Steering
 168     """
 169
 170     def __init__(self, test,
 171                  bsid,
 172                  prefix,
 173                  mask_width,
 174                  traffic_type,
 175                  sr_policy_index,
 176                  table_id,
 177                  sw_if_index):
 178         self._test = test
 179         self.bsid = bsid
 180         # keep binary format in _bsid
 181         self._bsid = inet_pton(AF_INET6, bsid)
 182         self.prefix = prefix
 183         # keep binary format in _prefix
 184         if ':' in prefix:
 185             # IPv6
 186             self._prefix = inet_pton(AF_INET6, prefix)
 187         else:
 188             # IPv4
 189             # API expects 16 octets (128 bits)
 190             # last 4 octets are used for IPv4
 191             # --> prepend 12 octets
 192             self._prefix = ('\x00' * 12) + inet_pton(AF_INET, prefix)
 193         self.mask_width = mask_width
 194         self.traffic_type = traffic_type
 195         self.sr_policy_index = sr_policy_index
 196         self.sw_if_index = sw_if_index
 197         self.table_id = table_id
 198         self._configured = False
 199
 200     def add_vpp_config(self):
 201         self._test.vapi.sr_steering_add_del(
 202                      0,
 203                      self._bsid,
 204                      self.sr_policy_index,
 205                      self.table_id,
 206                      self._prefix,
 207                      self.mask_width,
 208                      self.sw_if_index,
 209                      self.traffic_type)
 210         self._configured = True
 211
 212     def remove_vpp_config(self):
 213         self._test.vapi.sr_steering_add_del(
 214                      1,
 215                      self._bsid,
 216                      self.sr_policy_index,
 217                      self.table_id,
 218                      self._prefix,
 219                      self.mask_width,
 220                      self.sw_if_index,
 221                      self.traffic_type)
 222         self._configured = False
 223
 224     def query_vpp_config(self):
 225         # no API to query steering entries
 226         # use _configured flag for now
 227         return self._configured
 228
 229     def __str__(self):
 230         return self.object_id()
 231
 232     def object_id(self):
 233         return ("%d;%d;%s/%d->%s"
 234                 % (self.table_id,
 235                    self.traffic_type,
 236                    self.prefix,
 237                    self.mask_width,
 238                    self.bsid))