src/plugins/srv6-ad/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
  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_END_UN_PERF = 10
  23     SR_BEHAVIOR_END_UN = 11
  24     SR_BEHAVIOR_LAST = 12      # Must always be the last one
  25
  26
  27 class SRv6PolicyType():
  28     # from src/vnet/srv6/sr.h
  29     SR_POLICY_TYPE_DEFAULT = 0
  30     SR_POLICY_TYPE_SPRAY = 1
  31
  32
  33 class SRv6PolicySteeringTypes():
  34     # from src/vnet/srv6/sr.h
  35     SR_STEER_L2 = 2
  36     SR_STEER_IPV4 = 4
  37     SR_STEER_IPV6 = 6
  38
  39
  40 class VppSRv6LocalSID(VppObject):
  41     """
  42     SRv6 LocalSID
  43     """
  44
  45     def __init__(self, test, localsid, behavior, nh_addr,
  46                  end_psp, sw_if_index, vlan_index, fib_table):
  47         self._test = test
  48         self.localsid = localsid
  49         self.behavior = behavior
  50         self.nh_addr = nh_addr
  51         self.end_psp = end_psp
  52         self.sw_if_index = sw_if_index
  53         self.vlan_index = vlan_index
  54         self.fib_table = fib_table
  55         self._configured = False
  56
  57     def add_vpp_config(self):
  58         self._test.vapi.sr_localsid_add_del(
  59             localsid=self.localsid,
  60             behavior=self.behavior,
  61             nh_addr=self.nh_addr,
  62             is_del=0,
  63             end_psp=self.end_psp,
  64             sw_if_index=self.sw_if_index,
  65             vlan_index=self.vlan_index,
  66             fib_table=self.fib_table)
  67         self._configured = True
  68
  69     def remove_vpp_config(self):
  70         self._test.vapi.sr_localsid_add_del(
  71             localsid=self.localsid,
  72             behavior=self.behavior,
  73             nh_addr=self.nh_addr,
  74             is_del=1,
  75             end_psp=self.end_psp,
  76             sw_if_index=self.sw_if_index,
  77             vlan_index=self.vlan_index,
  78             fib_table=self.fib_table)
  79         self._configured = False
  80
  81     def query_vpp_config(self):
  82         # sr_localsids_dump API is disabled
  83         # use _configured flag for now
  84         return self._configured
  85
  86     def object_id(self):
  87         return ("%d;%s,%d"
  88                 % (self.fib_table,
  89                    self.localsid,
  90                    self.behavior))
  91
  92
  93 class VppSRv6Policy(VppObject):
  94     """
  95     SRv6 Policy
  96     """
  97
  98     def __init__(self, test, bsid,
  99                  is_encap, sr_type, weight, fib_table,
 100                  segments, source):
 101         self._test = test
 102         self.bsid = bsid
 103         self.is_encap = is_encap
 104         self.sr_type = sr_type
 105         self.weight = weight
 106         self.fib_table = fib_table
 107         self.segments = segments
 108         self.n_segments = len(segments)
 109         # source not passed to API
 110         # self.source = inet_pton(AF_INET6, source)
 111         self.source = source
 112         self._configured = False
 113
 114     def add_vpp_config(self):
 115         self._test.vapi.sr_policy_add(
 116                      bsid=self.bsid,
 117                      weight=self.weight,
 118                      is_encap=self.is_encap,
 119                      is_spray=self.sr_type,
 120                      fib_table=self.fib_table,
 121                      sids={'num_sids': self.n_segments, 'sids': self.segments})
 122         self._configured = True
 123
 124     def remove_vpp_config(self):
 125         self._test.vapi.sr_policy_del(
 126                      self.bsid)
 127         self._configured = False
 128
 129     def query_vpp_config(self):
 130         # no API to query SR Policies
 131         # use _configured flag for now
 132         return self._configured
 133
 134     def object_id(self):
 135         return ("%d;%s-><%s>;%d"
 136                 % (self.sr_type,
 137                    self.bsid,
 138                    ','.join(self.segments),
 139                    self.is_encap))
 140
 141
 142 class VppSRv6Steering(VppObject):
 143     """
 144     SRv6 Steering
 145     """
 146
 147     def __init__(self, test,
 148                  bsid,
 149                  prefix,
 150                  mask_width,
 151                  traffic_type,
 152                  sr_policy_index,
 153                  table_id,
 154                  sw_if_index):
 155         self._test = test
 156         self.bsid = bsid
 157         self.prefix = prefix
 158         self.mask_width = mask_width
 159         self.traffic_type = traffic_type
 160         self.sr_policy_index = sr_policy_index
 161         self.sw_if_index = sw_if_index
 162         self.table_id = table_id
 163         self._configured = False
 164
 165     def add_vpp_config(self):
 166         self._test.vapi.sr_steering_add_del(
 167                      is_del=0,
 168                      bsid=self.bsid,
 169                      sr_policy_index=self.sr_policy_index,
 170                      table_id=self.table_id,
 171                      prefix={'address': self.prefix, 'len':  self.mask_width},
 172                      sw_if_index=self.sw_if_index,
 173                      traffic_type=self.traffic_type)
 174         self._configured = True
 175
 176     def remove_vpp_config(self):
 177         self._test.vapi.sr_steering_add_del(
 178                      is_del=1,
 179                      bsid=self.bsid,
 180                      sr_policy_index=self.sr_policy_index,
 181                      table_id=self.table_id,
 182                      prefix={'address': self.prefix, 'len':  self.mask_width},
 183                      sw_if_index=self.sw_if_index,
 184                      traffic_type=self.traffic_type)
 185         self._configured = False
 186
 187     def query_vpp_config(self):
 188         # no API to query steering entries
 189         # use _configured flag for now
 190         return self._configured
 191
 192     def object_id(self):
 193         return ("%d;%d;%s/%d->%s"
 194                 % (self.table_id,
 195                    self.traffic_type,
 196                    self.prefix,
 197                    self.mask_width,
 198                    self.bsid))