test/vpp_bier.py

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