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