test/vpp_bier.py

   1 """
   2   BIER Tables and Routes
   3 """
   4
   5 import socket
   6 from vpp_object import VppObject
   7 from vpp_ip_route import MPLS_LABEL_INVALID, VppRoutePath, VppMplsLabel
   8
   9
  10 class BIER_HDR_PAYLOAD:
  11     BIER_HDR_PROTO_MPLS_DOWN_STREAM = 1
  12     BIER_HDR_PROTO_MPLS_UP_STREAM = 2
  13     BIER_HDR_PROTO_ETHERNET = 3
  14     BIER_HDR_PROTO_IPV4 = 4
  15     BIER_HDR_PROTO_IPV6 = 5
  16     BIER_HDR_PROTO_VXLAN = 6
  17     BIER_HDR_PROTO_CTRL = 7
  18     BIER_HDR_PROTO_OAM = 8
  19
  20
  21 class VppBierTableID():
  22     def __init__(self, sub_domain_id, set_id, hdr_len_id):
  23         self.set_id = set_id
  24         self.sub_domain_id = sub_domain_id
  25         self.hdr_len_id = hdr_len_id
  26
  27
  28 def find_bier_table(test, bti):
  29     tables = test.vapi.bier_table_dump()
  30     for t in tables:
  31         if bti.set_id == t.bt_tbl_id.bt_set \
  32            and bti.sub_domain_id == t.bt_tbl_id.bt_sub_domain \
  33            and bti.hdr_len_id == t.bt_tbl_id.bt_hdr_len_id:
  34             return True
  35     return False
  36
  37
  38 def find_bier_route(test, bti, bp):
  39     routes = test.vapi.bier_route_dump(bti)
  40     for r in routes:
  41         if bti.set_id == r.br_route.br_tbl_id.bt_set \
  42            and bti.sub_domain_id == r.br_route.br_tbl_id.bt_sub_domain \
  43            and bti.hdr_len_id == r.br_route.br_tbl_id.bt_hdr_len_id \
  44            and bp == r.br_route.br_bp:
  45             return True
  46     return False
  47
  48
  49 def find_bier_disp_table(test, bdti):
  50     tables = test.vapi.bier_disp_table_dump()
  51     for t in tables:
  52         if bdti == t.bdt_tbl_id:
  53             return True
  54     return False
  55
  56
  57 def find_bier_disp_entry(test, bdti, bp):
  58     entries = test.vapi.bier_disp_entry_dump(bdti)
  59     for e in entries:
  60         if bp == e.bde_bp \
  61            and bdti == e.bde_tbl_id:
  62             return True
  63     return False
  64
  65
  66 def find_bier_imp(test, bti, bp):
  67     imps = test.vapi.bier_imp_dump()
  68     for i in imps:
  69         if bti.set_id == i.bi_tbl_id.bt_set \
  70            and bti.sub_domain_id == i.bi_tbl_id.bt_sub_domain \
  71            and bti.hdr_len_id == i.bi_tbl_id.bt_hdr_len_id \
  72            and bp == i.bi_src:
  73             return True
  74     return False
  75
  76
  77 class VppBierTable(VppObject):
  78     """
  79     BIER Table
  80     """
  81
  82     def __init__(self, test, id, mpls_label):
  83         self._test = test
  84         self.id = id
  85         self.mpls_label = mpls_label
  86
  87     def add_vpp_config(self):
  88         self._test.vapi.bier_table_add_del(
  89             self.id,
  90             self.mpls_label,
  91             is_add=1)
  92         self._test.registry.register(self, self._test.logger)
  93
  94     def remove_vpp_config(self):
  95         self._test.vapi.bier_table_add_del(
  96             self.id,
  97             self.mpls_label,
  98             is_add=0)
  99
 100     def object_id(self):
 101         return "bier-table;[%d:%d:%d]" % (self.id.set_id,
 102                                           self.id.sub_domain_id,
 103                                           self.id.hdr_len_id)
 104
 105     def query_vpp_config(self):
 106         return find_bier_table(self._test, self.id)
 107
 108
 109 class VppBierRoute(VppObject):
 110     """
 111     BIER route
 112     """
 113
 114     def __init__(self, test, tbl_id, bp, paths):
 115         self._test = test
 116         self.tbl_id = tbl_id
 117         self.bp = bp
 118         self.paths = paths
 119         self.encoded_paths = []
 120         for path in self.paths:
 121             self.encoded_paths.append(path.encode())
 122
 123     def add_vpp_config(self):
 124         self._test.vapi.bier_route_add_del(
 125             self.tbl_id,
 126             self.bp,
 127             self.encoded_paths,
 128             is_add=1)
 129         self._test.registry.register(self, self._test.logger)
 130
 131     def remove_vpp_config(self):
 132         self._test.vapi.bier_route_add_del(
 133             self.tbl_id,
 134             self.bp,
 135             self.encoded_paths,
 136             is_add=0)
 137
 138     def update_paths(self, paths):
 139         self.paths = paths
 140         self.encoded_paths = []
 141         for path in self.paths:
 142             self.encoded_paths.append(path.encode())
 143         self._test.vapi.bier_route_add_del(
 144             self.tbl_id,
 145             self.bp,
 146             self.encoded_paths,
 147             is_replace=1)
 148
 149     def add_path(self, path):
 150         self.encoded_paths.append(path.encode())
 151         self._test.vapi.bier_route_add_del(
 152             self.tbl_id,
 153             self.bp,
 154             [path.encode()],
 155             is_add=1,
 156             is_replace=0)
 157         self.paths.append(path)
 158         self._test.registry.register(self, self._test.logger)
 159
 160     def remove_path(self, path):
 161         self.encoded_paths.remove(path.encode())
 162         self._test.vapi.bier_route_add_del(
 163             self.tbl_id,
 164             self.bp,
 165             [path.encode()],
 166             is_add=0,
 167             is_replace=0)
 168         self.paths.remove(path)
 169
 170     def remove_all_paths(self):
 171         self._test.vapi.bier_route_add_del(
 172             self.tbl_id,
 173             self.bp,
 174             [],
 175             is_add=0,
 176             is_replace=1)
 177         self.paths = []
 178
 179     def object_id(self):
 180         return "bier-route;[%d:%d:%d:%d]" % (self.tbl_id.set_id,
 181                                              self.tbl_id.sub_domain_id,
 182                                              self.tbl_id.hdr_len_id,
 183                                              self.bp)
 184
 185     def query_vpp_config(self):
 186         return find_bier_route(self._test, self.tbl_id, self.bp)
 187
 188
 189 class VppBierImp(VppObject):
 190     """
 191     BIER route
 192     """
 193
 194     def __init__(self, test, tbl_id, src, ibytes):
 195         self._test = test
 196         self.tbl_id = tbl_id
 197         self.ibytes = ibytes
 198         self.src = src
 199
 200     def add_vpp_config(self):
 201         res = self._test.vapi.bier_imp_add(
 202             self.tbl_id,
 203             self.src,
 204             self.ibytes)
 205         self.bi_index = res.bi_index
 206         self._test.registry.register(self, self._test.logger)
 207
 208     def remove_vpp_config(self):
 209         self._test.vapi.bier_imp_del(
 210             self.bi_index)
 211
 212     def object_id(self):
 213         return "bier-imp;[%d:%d:%d:%d]" % (self.tbl_id.set_id,
 214                                            self.tbl_id.sub_domain_id,
 215                                            self.tbl_id.hdr_len_id,
 216                                            self.src)
 217
 218     def query_vpp_config(self):
 219         return find_bier_imp(self._test, self.tbl_id, self.src)
 220
 221
 222 class VppBierDispTable(VppObject):
 223     """
 224     BIER Disposition Table
 225     """
 226
 227     def __init__(self, test, id):
 228         self._test = test
 229         self.id = id
 230
 231     def add_vpp_config(self):
 232         self._test.vapi.bier_disp_table_add_del(
 233             self.id,
 234             is_add=1)
 235         self._test.registry.register(self, self._test.logger)
 236
 237     def remove_vpp_config(self):
 238         self._test.vapi.bier_disp_table_add_del(
 239             self.id,
 240             is_add=0)
 241
 242     def object_id(self):
 243         return "bier-disp-table;[%d]" % (self.id)
 244
 245     def query_vpp_config(self):
 246         return find_bier_disp_table(self._test, self.id)
 247
 248
 249 class VppBierDispEntry(VppObject):
 250     """
 251     BIER Disposition Entry
 252     """
 253
 254     def __init__(self, test, tbl_id, bp, payload_proto, nh_proto,
 255                  nh, nh_tbl, rpf_id=~0):
 256         self._test = test
 257         self.tbl_id = tbl_id
 258         self.nh_tbl = nh_tbl
 259         self.nh_proto = nh_proto
 260         self.bp = bp
 261         self.payload_proto = payload_proto
 262         self.rpf_id = rpf_id
 263         self.nh = socket.inet_pton(socket.AF_INET, nh)
 264
 265     def add_vpp_config(self):
 266         self._test.vapi.bier_disp_entry_add_del(
 267             self.tbl_id,
 268             self.bp,
 269             self.payload_proto,
 270             self.nh_proto,
 271             self.nh,
 272             self.nh_tbl,
 273             self.rpf_id,
 274             is_add=1)
 275         self._test.registry.register(self, self._test.logger)
 276
 277     def remove_vpp_config(self):
 278         self._test.vapi.bier_disp_entry_add_del(
 279             self.tbl_id,
 280             self.bp,
 281             self.payload_proto,
 282             self.nh_proto,
 283             self.nh,
 284             self.nh_tbl,
 285             self.rpf_id,
 286             is_add=0)
 287
 288     def object_id(self):
 289         return "bier-disp-entry;[%d:%d]" % (self.tbl_id,
 290                                             self.bp)
 291
 292     def query_vpp_config(self):
 293         return find_bier_disp_entry(self._test, self.tbl_id, self.bp)