""" BIER Tables and Routes """ import socket from vpp_object import VppObject from vpp_ip_route import MPLS_LABEL_INVALID, VppRoutePath, VppMplsLabel class BIER_HDR_PAYLOAD: BIER_HDR_PROTO_MPLS_DOWN_STREAM = 1 BIER_HDR_PROTO_MPLS_UP_STREAM = 2 BIER_HDR_PROTO_ETHERNET = 3 BIER_HDR_PROTO_IPV4 = 4 BIER_HDR_PROTO_IPV6 = 5 BIER_HDR_PROTO_VXLAN = 6 BIER_HDR_PROTO_CTRL = 7 BIER_HDR_PROTO_OAM = 8 class VppBierTableID(): def __init__(self, sub_domain_id, set_id, hdr_len_id): self.set_id = set_id self.sub_domain_id = sub_domain_id self.hdr_len_id = hdr_len_id def find_bier_table(test, bti): tables = test.vapi.bier_table_dump() for t in tables: if bti.set_id == t.bt_tbl_id.bt_set \ and bti.sub_domain_id == t.bt_tbl_id.bt_sub_domain \ and bti.hdr_len_id == t.bt_tbl_id.bt_hdr_len_id: return True return False def find_bier_route(test, bti, bp): routes = test.vapi.bier_route_dump(bti) for r in routes: if bti.set_id == r.br_route.br_tbl_id.bt_set \ and bti.sub_domain_id == r.br_route.br_tbl_id.bt_sub_domain \ and bti.hdr_len_id == r.br_route.br_tbl_id.bt_hdr_len_id \ and bp == r.br_route.br_bp: return True return False def find_bier_disp_table(test, bdti): tables = test.vapi.bier_disp_table_dump() for t in tables: if bdti == t.bdt_tbl_id: return True return False def find_bier_disp_entry(test, bdti, bp): entries = test.vapi.bier_disp_entry_dump(bdti) for e in entries: if bp == e.bde_bp \ and bdti == e.bde_tbl_id: return True return False def find_bier_imp(test, bti, bp): imps = test.vapi.bier_imp_dump() for i in imps: if bti.set_id == i.bi_tbl_id.bt_set \ and bti.sub_domain_id == i.bi_tbl_id.bt_sub_domain \ and bti.hdr_len_id == i.bi_tbl_id.bt_hdr_len_id \ and bp == i.bi_src: return True return False class VppBierTable(VppObject): """ BIER Table """ def __init__(self, test, id, mpls_label): self._test = test self.id = id self.mpls_label = mpls_label def add_vpp_config(self): self._test.vapi.bier_table_add_del( self.id, self.mpls_label, is_add=1) self._test.registry.register(self, self._test.logger) def remove_vpp_config(self): self._test.vapi.bier_table_add_del( self.id, self.mpls_label, is_add=0) def object_id(self): return "bier-table;[%d:%d:%d]" % (self.id.set_id, self.id.sub_domain_id, self.id.hdr_len_id) def query_vpp_config(self): return find_bier_table(self._test, self.id) class VppBierRoute(VppObject): """ BIER route """ def __init__(self, test, tbl_id, bp, paths): self._test = test self.tbl_id = tbl_id self.bp = bp self.paths = paths self.encoded_paths = [] for path in self.paths: self.encoded_paths.append(path.encode()) def add_vpp_config(self): self._test.vapi.bier_route_add_del( self.tbl_id, self.bp, self.encoded_paths, is_add=1) self._test.registry.register(self, self._test.logger) def remove_vpp_config(self): self._test.vapi.bier_route_add_del( self.tbl_id, self.bp, self.encoded_paths, is_add=0) def update_paths(self, paths): self.paths = paths self.encoded_paths = [] for path in self.paths: self.encoded_paths.append(path.encode()) self._test.vapi.bier_route_add_del( self.tbl_id, self.bp, self.encoded_paths, is_replace=1) def add_path(self, path): self.encoded_paths.append(path.encode()) self._test.vapi.bier_route_add_del( self.tbl_id, self.bp, [path.encode()], is_add=1, is_replace=0) self.paths.append(path) self._test.registry.register(self, self._test.logger) def remove_path(self, path): self.encoded_paths.remove(path.encode()) self._test.vapi.bier_route_add_del( self.tbl_id, self.bp, [path.encode()], is_add=0, is_replace=0) self.paths.remove(path) def remove_all_paths(self): self._test.vapi.bier_route_add_del( self.tbl_id, self.bp, [], is_add=0, is_replace=1) self.paths = [] def object_id(self): return "bier-route;[%d:%d:%d:%d]" % (self.tbl_id.set_id, self.tbl_id.sub_domain_id, self.tbl_id.hdr_len_id, self.bp) def query_vpp_config(self): return find_bier_route(self._test, self.tbl_id, self.bp) class VppBierImp(VppObject): """ BIER route """ def __init__(self, test, tbl_id, src, ibytes): self._test = test self.tbl_id = tbl_id self.ibytes = ibytes self.src = src def add_vpp_config(self): res = self._test.vapi.bier_imp_add( self.tbl_id, self.src, self.ibytes) self.bi_index = res.bi_index