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_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_addr,
  44                  end_psp, sw_if_index, vlan_index, fib_table):
  45         self._test = test
  46         self.localsid = localsid
  47         self.behavior = behavior
  48         self.nh_addr = nh_addr
  49         self.end_psp = end_psp
  50         self.sw_if_index = sw_if_index
  51         self.vlan_index = vlan_index
  52         self.fib_table = fib_table
  53         self._configured = False
  54
  55     def add_vpp_config(self):
  56         self._test.vapi.sr_localsid_add_del(
  57             localsid=self.localsid,
  58             behavior=self.behavior,
  59             nh_addr=self.nh_addr,
  60             is_del=0,
  61             end_psp=self.end_psp,
  62             sw_if_index=self.sw_if_index,
  63             vlan_index=self.vlan_index,
  64             fib_table=self.fib_table)
  65         self._configured = True
  66
  67     def remove_vpp_config(self):
  68         self._test.vapi.sr_localsid_add_del(
  69             localsid=self.localsid,
  70             behavior=self.behavior,
  71             nh_addr=self.nh_addr,
  72             is_del=1,
  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 = False
  78
  79     def query_vpp_config(self):
  80         # sr_localsids_dump API is disabled
  81         # use _configured flag for now
  82         return self._configured
  83
  84     def object_id(self):
  85         return ("%d;%s,%d"
  86                 % (self.fib_table,
  87                    self.localsid,
  88                    self.behavior))
  89
  90
  91 class VppSRv6Policy(VppObject):
  92     """
  93     SRv6 Policy
  94     """
  95
  96     def __init__(self, test, bsid,
  97                  is_encap, sr_type, weight, fib_table,
  98                  segments, source):
  99         self._test = test
 100         self.bsid = bsid
 101         self.is_encap = is_encap
 102         self.sr_type = sr_type
 103         self.weight = weight
 104         self.fib_table = fib_table
 105         self.segments = segments
 106         self.n_segments = len(segments)
 107         # source not passed to API
 108         # self.source = inet_pton(AF_INET6, source)
 109         self.source = source
 110         self._configured = False
 111
 112     def add_vpp_config(self):
 113         self._test.vapi.sr_policy_add(
 114                      bsid=self.bsid,
 115                      weight=self.weight,
 116                      is_encap=self.is_encap,
 117                      is_spray=self.sr_type,
 118                      fib_table=self.fib_table,
 119                      sids={'num_sids': self.n_segments, 'sids': self.segments})
 120         self._configured = True
 121
 122     def remove_vpp_config(self):
 123         self._test.vapi.sr_policy_del(
 124                      self.bsid)
 125         self._configured = False
 126
 127     def query_vpp_config(self):
 128         # no API to query SR Policies
 129         # use _configured flag for now
 130         return self._configured
 131
 132     def object_id(self):
 133         return ("%d;%s-><%s>;%d"
 134                 % (self.sr_type,
 135                    self.bsid,
 136                    ','.join(self.segments),
 137                    self.is_encap))
 138
 139
 140 class VppSRv6Steering(VppObject):
 141     """
 142     SRv6 Steering
 143     """
 144
 145     def __init__(self, test,
 146                  bsid,
 147                  prefix,
 148                  mask_width,
 149                  traffic_type,
 150                  sr_policy_index,
 151                  table_id,
 152                  sw_if_index):
 153         self._test = test
 154         self.bsid = bsid
 155         self.prefix = prefix
 156         self.mask_width = mask_width
 157         self.traffic_type = traffic_type
 158         self.sr_policy_index = sr_policy_index
 159         self.sw_if_index = sw_if_index
 160         self.table_id = table_id
 161         self._configured = False
 162
 163     def add_vpp_config(self):
 164         self._test.vapi.sr_steering_add_del(
 165                      is_del=0,
 166                      bsid=self.bsid,
 167                      sr_policy_index=self.sr_policy_index,
 168                      table_id=self.table_id,
 169                      prefix={'address': self.prefix, 'len':  self.mask_width},
 170                      sw_if_index=self.sw_if_index,
 171                      traffic_type=self.traffic_type)
 172         self._configured = True
 173
 174     def remove_vpp_config(self):
 175         self._test.vapi.sr_steering_add_del(
 176                      is_del=1,
 177                      bsid=self.bsid,
 178                      sr_policy_index=self.sr_policy_index,
 179                      table_id=self.table_id,
 180                      prefix={'address': self.prefix, 'len':  self.mask_width},
 181                      sw_if_index=self.sw_if_index,
 182                      traffic_type=self.traffic_type)
 183         self._configured = False
 184
 185     def query_vpp_config(self):
 186         # no API to query steering entries
 187         # use _configured flag for now
 188         return self._configured
 189
 190     def object_id(self):
 191         return ("%d;%d;%s/%d->%s"
 192                 % (self.table_id,
 193                    self.traffic_type,
 194                    self.prefix,
 195                    self.mask_width,
 196                    self.bsid))