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