| """ |
| BIER Tables and Routes |
| """ |
| |
| import socket |
| from vpp_object import VppObject |
| |
| |
| 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 |
| self._test.registry.register(self, self._test.logger) |
| |
| def remove_vpp_config(self): |
| self._test.vapi.bier_imp_del(self.bi_index) |
| |
| def object_id(self): |
| return "bier-imp;[%d:%d:%d:%d]" % ( |
| self.tbl_id.set_id, |
| self.tbl_id.sub_domain_id, |
| self.tbl_id.hdr_len_id, |
| self.src, |
| ) |
| |
| def query_vpp_config(self): |
| return find_bier_imp(self._test, self.tbl_id, self.src) |
| |
| |
| class VppBierDispTable(VppObject): |
| """ |
| BIER Disposition Table |
| """ |
| |
| def __init__(self, test, id): |
| self._test = test |
| self.id = id |
| |
| def add_vpp_config(self): |
| self._test.vapi.bier_disp_table_add_del(self.id, is_add=1) |
| self._test.registry.register(self, self._test.logger) |
| |
| def remove_vpp_config(self): |
| self._test.vapi.bier_disp_table_add_del(self.id, is_add=0) |
| |
| def object_id(self): |
| return "bier-disp-table;[%d]" % (self.id) |
| |
| def query_vpp_config(self): |
| return find_bier_disp_table(self._test, self.id) |
| |
| |
| class VppBierDispEntry(VppObject): |
| """ |
| BIER Disposition Entry |
| """ |
| |
| def __init__( |
| self, test, tbl_id, bp, payload_proto, nh_proto, nh, nh_tbl, rpf_id=~0 |
| ): |
| self._test = test |
| self.tbl_id = tbl_id |
| self.nh_tbl = nh_tbl |
| self.nh_proto = nh_proto |
| self.bp = bp |
| self.payload_proto = payload_proto |
| self.rpf_id = rpf_id |
| self.nh = socket.inet_pton(socket.AF_INET, nh) |
| |
| def add_vpp_config(self): |
| self._test.vapi.bier_disp_entry_add_del( |
| self.tbl_id, |
| self.bp, |
| self.payload_proto, |
| self.nh_proto, |
| self.nh, |
| self.nh_tbl, |
| self.rpf_id, |
| is_add=1, |
| ) |
| self._test.registry.register(self, self._test.logger) |
| |
| def remove_vpp_config(self): |
| self._test.vapi.bier_disp_entry_add_del( |
| self.tbl_id, |
| self.bp, |
| self.payload_proto, |
| self.nh_proto, |
| self.nh, |
| self.nh_tbl, |
| self.rpf_id, |
| is_add=0, |
| ) |
| |
| def object_id(self): |
| return "bier-disp-entry;[%d:%d]" % (self.tbl_id, self.bp) |
| |
| def query_vpp_config(self): |
| return find_bier_disp_entry(self._test, self.tbl_id, self.bp) |