KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use delete_path method in Airtest
Best Python code snippet using Airtest
bgp_neighbors.py
Source:bgp_neighbors.py 
1#2# -*- coding: utf-8 -*-3# Copyright 2019 Red Hat4# GNU General Public License v3.0+5# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)6"""7The sonic_bgp_neighbors class8It is in this file where the current configuration (as dict)9is compared to the provided configuration (as dict) and the command set10necessary to bring the current configuration to it's desired end-state is11created12"""13from __future__ import absolute_import, division, print_function14__metaclass__ = type15from ansible_collections.ansible.netcommon.plugins.module_utils.network.common.cfg.base import (16 ConfigBase,17)18from ansible_collections.ansible.netcommon.plugins.module_utils.network.common.utils import (19 to_list,20)21from ansible_collections.dellemc.enterprise_sonic.plugins.module_utils.network.sonic.facts.facts import Facts22from ansible_collections.dellemc.enterprise_sonic.plugins.module_utils.network.sonic.sonic import (23 to_request,24 edit_config25)26from ansible_collections.dellemc.enterprise_sonic.plugins.module_utils.network.sonic.utils.utils import (27 update_states,28 get_diff,29)30from ansible_collections.dellemc.enterprise_sonic.plugins.module_utils.network.sonic.utils.bgp_utils import (31 validate_bgps,32 normalize_neighbors_interface_name,33)34from ansible_collections.dellemc.enterprise_sonic.plugins.module_utils.network.sonic.sonic import to_request35from ansible.module_utils.connection import ConnectionError36PATCH = 'patch'37DELETE = 'delete'38TEST_KEYS = [39 {'config': {'vrf_name': '', 'bgp_as': ''}},40 {'neighbors': {'neighbor': ''}},41 {'peer_group': {'name': ''}},42 {'afis': {'afi': '', 'safi': ''}},43]44class Bgp_neighbors(ConfigBase):45 """46 The sonic_bgp_neighbors class47 """48 gather_subset = [49 '!all',50 '!min',51 ]52 gather_network_resources = [53 'bgp_neighbors',54 ]55 network_instance_path = '/data/openconfig-network-instance:network-instances/network-instance'56 protocol_bgp_path = 'protocols/protocol=BGP,bgp/bgp'57 neighbor_path = 'neighbors/neighbor'58 def __init__(self, module):59 super(Bgp_neighbors, self).__init__(module)60 def get_bgp_neighbors_facts(self):61 """ Get the 'facts' (the current configuration)62 :rtype: A dictionary63 :returns: The current configuration as a dictionary64 """65 facts, _warnings = Facts(self._module).get_facts(self.gather_subset, self.gather_network_resources)66 bgp_facts = facts['ansible_network_resources'].get('bgp_neighbors')67 if not bgp_facts:68 bgp_facts = []69 return bgp_facts70 def execute_module(self):71 """ Execute the module72 :rtype: A dictionary73 :returns: The result from module execution74 """75 result = {'changed': False}76 warnings = list()77 existing_bgp_facts = self.get_bgp_neighbors_facts()78 commands, requests = self.set_config(existing_bgp_facts)79 if commands and len(requests) > 0:80 if not self._module.check_mode:81 try:82 edit_config(self._module, to_request(self._module, requests))83 except ConnectionError as exc:84 self._module.fail_json(msg=str(exc), code=exc.code)85 result['changed'] = True86 result['commands'] = commands87 changed_bgp_facts = self.get_bgp_neighbors_facts()88 result['before'] = existing_bgp_facts89 if result['changed']:90 result['after'] = changed_bgp_facts91 result['warnings'] = warnings92 return result93 def set_config(self, existing_bgp_facts):94 """ Collect the configuration from the args passed to the module,95 collect the current configuration (as a dict from facts)96 :rtype: A list97 :returns: the commands necessary to migrate the current configuration98 to the desired configuration99 """100 want = self._module.params['config']101 normalize_neighbors_interface_name(want, self._module)102 have = existing_bgp_facts103 resp = self.set_state(want, have)104 return to_list(resp)105 def set_state(self, want, have):106 """ Select the appropriate function based on the state provided107 :param want: the desired configuration as a dictionary108 :param have: the current configuration as a dictionary109 :rtype: A list110 :returns: the commands necessary to migrate the current configuration111 to the desired configuration112 """113 commands = []114 requests = []115 state = self._module.params['state']116 diff = get_diff(want, have, TEST_KEYS)117 if state == 'deleted':118 commands, requests = self._state_deleted(want, have, diff)119 elif state == 'merged':120 commands, requests = self._state_merged(want, have, diff)121 return commands, requests122 def _state_merged(self, want, have, diff):123 """ The command generator when state is merged124 :param want: the additive configuration as a dictionary125 :param have: the current configuration as a dictionary126 :rtype: A list127 :returns: the commands necessary to merge the provided into128 the current configuration129 """130 commands = []131 requests = []132 commands = diff133 validate_bgps(self._module, commands, have)134 requests = self.get_modify_bgp_requests(commands, have)135 if commands and len(requests) > 0:136 commands = update_states(commands, "merged")137 else:138 commands = []139 return commands, requests140 def _state_deleted(self, want, have, diff):141 """ The command generator when state is deleted142 :param want: the objects from which the configuration should be removed143 :param have: the current configuration as a dictionary144 :rtype: A list145 :returns: the commands necessary to remove the current configuration146 of the provided objects147 """148 is_delete_all = False149 if not want:150 is_delete_all = True151 if is_delete_all:152 commands = have153 new_have = have154 else:155 new_have = self.remove_default_entries(have)156 d_diff = get_diff(want, new_have, TEST_KEYS, is_skeleton=True)157 delete_diff = get_diff(want, d_diff, TEST_KEYS, is_skeleton=True)158 commands = delete_diff159 requests = self.get_delete_bgp_neighbor_requests(commands, new_have, want, is_delete_all)160 if commands and len(requests) > 0:161 commands = update_states(commands, "deleted")162 else:163 commands = []164 return commands, requests165 def remove_default_entries(self, data):166 new_data = []167 if not data:168 return new_data169 for conf in data:170 new_conf = {}171 as_val = conf['bgp_as']172 vrf_name = conf['vrf_name']173 new_conf['bgp_as'] = as_val174 new_conf['vrf_name'] = vrf_name175 peergroup = conf.get('peer_group', None)176 new_peergroups = []177 if peergroup is not None:178 for pg in peergroup:179 new_pg = {}180 pg_val = pg.get('name', None)181 new_pg['name'] = pg_val182 remote_as = pg.get('remote_as', None)183 new_remote = {}184 if remote_as:185 peer_as = remote_as.get('peer_as', None)186 peer_type = remote_as.get('peer_type', None)187 if peer_as is not None:188 new_remote['peer_as'] = peer_as189 if peer_type is not None:190 new_remote['peer_type'] = peer_type191 if new_remote:192 new_pg['remote_as'] = new_remote193 timers = pg.get('timers', None)194 new_timers = {}195 if timers:196 keepalive = timers.get('keepalive', None)197 holdtime = timers.get('holdtime', None)198 connect_retry = timers.get('connect_retry', None)199 if keepalive is not None and keepalive != 60:200 new_timers['keepalive'] = keepalive201 if holdtime is not None and holdtime != 180:202 new_timers['holdtime'] = holdtime203 if connect_retry is not None and connect_retry != 30:204 new_timers['connect_retry'] = connect_retry205 if new_timers:206 new_pg['timers'] = new_timers207 advertisement_interval = pg.get('advertisement_interval', None)208 if advertisement_interval is not None and advertisement_interval != 30:209 new_pg['advertisement_interval'] = advertisement_interval210 bfd = pg.get('bfd', None)211 if bfd is not None:212 new_pg['bfd'] = bfd213 capability = pg.get('capability', None)214 if capability is not None:215 new_pg['capability'] = capability216 afi = []217 address_family = pg.get('address_family', None)218 if address_family:219 if address_family.get('afis', None):220 for each in address_family['afis']:221 if each:222 tmp = {}223 if each.get('afi', None) is not None:224 tmp['afi'] = each['afi']225 if each.get('safi', None) is not None:226 tmp['safi'] = each['safi']227 if each.get('activate', None) is not None and each['activate'] is not False:228 tmp['activate'] = each['activate']229 if each.get('allowas_in', None) is not None:230 tmp['allowas_in'] = each['allowas_in']231 afi.append(tmp)232 if afi and len(afi) > 0:233 afis = {}234 afis.update({'afis': afi})235 new_pg['address_family'] = afis236 if new_pg:237 new_peergroups.append(new_pg)238 if new_peergroups:239 new_conf['peer_group'] = new_peergroups240 neighbors = conf.get('neighbors', None)241 new_neighbors = []242 if neighbors is not None:243 for neighbor in neighbors:244 new_neighbor = {}245 neighbor_val = neighbor.get('neighbor', None)246 new_neighbor['neighbor'] = neighbor_val247 remote_as = neighbor.get('remote_as', None)248 new_remote = {}249 if remote_as:250 peer_as = remote_as.get('peer_as', None)251 peer_type = remote_as.get('peer_type', None)252 if peer_as is not None:253 new_remote['peer_as'] = peer_as254 if peer_type is not None:255 new_remote['peer_type'] = peer_type256 if new_remote:257 new_neighbor['remote_as'] = new_remote258 peer_group = neighbor.get('peer_group', None)259 if peer_group:260 new_neighbor['peer_group'] = peer_group261 timers = neighbor.get('timers', None)262 new_timers = {}263 if timers:264 keepalive = timers.get('keepalive', None)265 holdtime = timers.get('holdtime', None)266 connect_retry = timers.get('connect_retry', None)267 if keepalive is not None and keepalive != 60:268 new_timers['keepalive'] = keepalive269 if holdtime is not None and holdtime != 180:270 new_timers['holdtime'] = holdtime271 if connect_retry is not None and connect_retry != 30:272 new_timers['connect_retry'] = connect_retry273 if new_timers:274 new_neighbor['timers'] = new_timers275 advertisement_interval = neighbor.get('advertisement_interval', None)276 if advertisement_interval is not None and advertisement_interval != 30:277 new_neighbor['advertisement_interval'] = advertisement_interval278 bfd = neighbor.get('bfd', None)279 if bfd is not None:280 new_neighbor['bfd'] = bfd281 capability = neighbor.get('capability', None)282 if capability is not None:283 new_neighbor['capability'] = capability284 if new_neighbor:285 new_neighbors.append(new_neighbor)286 if new_neighbors:287 new_conf['neighbors'] = new_neighbors288 if new_conf:289 new_data.append(new_conf)290 return new_data291 def build_bgp_peer_groups_payload(self, cmd, have, bgp_as, vrf_name):292 requests = []293 bgp_peer_group_list = []294 for peer_group in cmd:295 if peer_group:296 bgp_peer_group = {}297 peer_group_cfg = {}298 tmp_bfd = {}299 tmp_ebgp = {}300 tmp_timers = {}301 tmp_capability = {}302 tmp_remote = {}303 tmp_transport = {}304 afi = []305 if peer_group.get('name', None) is not None:306 peer_group_cfg.update({'peer-group-name': peer_group['name']})307 bgp_peer_group.update({'peer-group-name': peer_group['name']})308 if peer_group.get('bfd', None) is not None:309 if peer_group['bfd'].get('enabled', None) is not None:310 tmp_bfd.update({'enabled': peer_group['bfd']['enabled']})311 if peer_group['bfd'].get('check_failure', None) is not None:312 tmp_bfd.update({'check-control-plane-failure': peer_group['bfd']['check_failure']})313 if peer_group['bfd'].get('profile', None) is not None:314 tmp_bfd.update({'bfd-profile': peer_group['bfd']['profile']})315 if peer_group.get('auth_pwd', None) is not None:316 if (peer_group['auth_pwd'].get('pwd', None) is not None and317 peer_group['auth_pwd'].get('encrypted', None) is not None):318 bgp_peer_group.update({'auth-password': {'config': {'password': peer_group['auth_pwd']['pwd'],319 'encrypted': peer_group['auth_pwd']['encrypted']}}})320 if peer_group.get('ebgp_multihop', None) is not None:321 if peer_group['ebgp_multihop'].get('enabled', None) is not None:322 tmp_ebgp.update({'enabled': peer_group['ebgp_multihop']['enabled']})323 if peer_group['ebgp_multihop'].get('multihop_ttl', None) is not None:324 tmp_ebgp.update({'multihop-ttl': peer_group['ebgp_multihop']['multihop_ttl']})325 if peer_group.get('timers', None) is not None:326 if peer_group['timers'].get('holdtime', None) is not None:327 tmp_timers.update({'hold-time': peer_group['timers']['holdtime']})328 if peer_group['timers'].get('keepalive', None) is not None:329 tmp_timers.update({'keepalive-interval': peer_group['timers']['keepalive']})330 if peer_group['timers'].get('connect_retry', None) is not None:331 tmp_timers.update({'connect-retry': peer_group['timers']['connect_retry']})332 if peer_group.get('capability', None) is not None:333 if peer_group['capability'].get('dynamic', None) is not None:334 tmp_capability.update({'capability-dynamic': peer_group['capability']['dynamic']})335 if peer_group['capability'].get('extended_nexthop', None) is not None:336 tmp_capability.update({'capability-extended-nexthop': peer_group['capability']['extended_nexthop']})337 if peer_group.get('pg_description', None) is not None:338 peer_group_cfg.update({'description': peer_group['pg_description']})339 if peer_group.get('disable_connected_check', None) is not None:340 peer_group_cfg.update({'disable-ebgp-connected-route-check': peer_group['disable_connected_check']})341 if peer_group.get('dont_negotiate_capability', None) is not None:342 peer_group_cfg.update({'dont-negotiate-capability': peer_group['dont_negotiate_capability']})343 if peer_group.get('enforce_first_as', None) is not None:344 peer_group_cfg.update({'enforce-first-as': peer_group['enforce_first_as']})345 if peer_group.get('enforce_multihop', None) is not None:346 peer_group_cfg.update({'enforce-multihop': peer_group['enforce_multihop']})347 if peer_group.get('override_capability', None) is not None:348 peer_group_cfg.update({'override-capability': peer_group['override_capability']})349 if peer_group.get('shutdown_msg', None) is not None:350 peer_group_cfg.update({'shutdown-message': peer_group['shutdown_msg']})351 if peer_group.get('solo', None) is not None:352 peer_group_cfg.update({'solo-peer': peer_group['solo']})353 if peer_group.get('strict_capability_match', None) is not None:354 peer_group_cfg.update({'strict-capability-match': peer_group['strict_capability_match']})355 if peer_group.get('ttl_security', None) is not None:356 peer_group_cfg.update({'ttl-security-hops': peer_group['ttl_security']})357 if peer_group.get('local_as', None) is not None:358 if peer_group['local_as'].get('as', None) is not None:359 peer_group_cfg.update({'local-as': peer_group['local_as']['as']})360 if peer_group['local_as'].get('no_prepend', None) is not None:361 peer_group_cfg.update({'local-as-no-prepend': peer_group['local_as']['no_prepend']})362 if peer_group['local_as'].get('replace_as', None) is not None:363 peer_group_cfg.update({'local-as-replace-as': peer_group['local_as']['replace_as']})364 if peer_group.get('local_address', None) is not None:365 tmp_transport.update({'local-address': peer_group['local_address']})366 if peer_group.get('passive', None) is not None:367 tmp_transport.update({'passive-mode': peer_group['passive']})368 if peer_group.get('advertisement_interval', None) is not None:369 tmp_timers.update({'minimum-advertisement-interval': peer_group['advertisement_interval']})370 if peer_group.get('remote_as', None) is not None:371 have_nei = self.find_pg(have, bgp_as, vrf_name, peer_group)372 if peer_group['remote_as'].get('peer_as', None) is not None:373 if have_nei:374 if have_nei.get("remote_as", None) is not None:375 if have_nei["remote_as"].get("peer_type", None) is not None:376 del_nei = {}377 del_nei.update({'name': have_nei['name']})378 del_nei.update({'remote_as': have_nei['remote_as']})379 requests.extend(self.delete_specific_peergroup_param_request(vrf_name, del_nei))380 tmp_remote.update({'peer-as': peer_group['remote_as']['peer_as']})381 if peer_group['remote_as'].get('peer_type', None) is not None:382 if have_nei:383 if have_nei.get("remote_as", None) is not None:384 if have_nei["remote_as"].get("peer_as", None) is not None:385 del_nei = {}386 del_nei.update({'name': have_nei['name']})387 del_nei.update({'remote_as': have_nei['remote_as']})388 requests.extend(self.delete_specific_peergroup_param_request(vrf_name, del_nei))389 tmp_remote.update({'peer-type': peer_group['remote_as']['peer_type'].upper()})390 if peer_group.get('address_family', None) is not None:391 if peer_group['address_family'].get('afis', None) is not None:392 for each in peer_group['address_family']['afis']:393 samp = {}394 if each.get('afi', None) is not None and each.get('safi', None) is not None:395 afi_safi = each['afi'].upper() + "_" + each['safi'].upper()396 if afi_safi is not None:397 afi_safi_name = 'openconfig-bgp-types:' + afi_safi398 if afi_safi_name is not None:399 samp.update({'afi-safi-name': afi_safi_name})400 samp.update({'config': {'afi-safi-name': afi_safi_name}})401 if each.get('activate', None) is not None:402 enabled = each['activate']403 if enabled is not None:404 samp.update({'config': {'enabled': enabled}})405 if each.get('allowas_in', None) is not None:406 have_pg_af = self.find_af(have, bgp_as, vrf_name, peer_group, each['afi'], each['safi'])407 if each['allowas_in'].get('origin', None) is not None:408 if have_pg_af:409 if have_pg_af.get('allowas_in', None) is not None:410 if have_pg_af['allowas_in'].get('value', None) is not None:411 del_nei = {}412 del_nei.update({'name': peer_group['name']})413 afis_list = []414 temp_cfg = {'afi': each['afi'], 'safi': each['safi']}415 temp_cfg['allowas_in'] = {'value': have_pg_af['allowas_in']['value']}416 afis_list.append(temp_cfg)417 del_nei.update({'address_family': {'afis': afis_list}})418 requests.extend(self.delete_specific_peergroup_param_request(vrf_name, del_nei))419 origin = each['allowas_in']['origin']420 samp.update({'allow-own-as': {'config': {'origin': origin, "enabled": bool("true")}}})421 if each['allowas_in'].get('value', None) is not None:422 if have_pg_af:423 if have_pg_af.get('allowas_in', None) is not None:424 if have_pg_af['allowas_in'].get('origin', None) is not None:425 del_nei = {}426 del_nei.update({'name': peer_group['name']})427 afis_list = []428 temp_cfg = {'afi': each['afi'], 'safi': each['safi']}429 temp_cfg['allowas_in'] = {'origin': have_pg_af['allowas_in']['origin']}430 afis_list.append(temp_cfg)431 del_nei.update({'address_family': {'afis': afis_list}})432 requests.extend(self.delete_specific_peergroup_param_request(vrf_name, del_nei))433 as_count = each['allowas_in']['value']434 samp.update({'allow-own-as': {'config': {'as-count': as_count, "enabled": bool("true")}}})435 if samp:436 afi.append(samp)437 if tmp_bfd:438 bgp_peer_group.update({'enable-bfd': {'config': tmp_bfd}})439 if tmp_ebgp:440 bgp_peer_group.update({'ebgp-multihop': {'config': tmp_ebgp}})441 if tmp_timers:442 bgp_peer_group.update({'timers': {'config': tmp_timers}})443 if tmp_transport:444 bgp_peer_group.update({'transport': {'config': tmp_transport}})445 if afi and len(afi) > 0:446 bgp_peer_group.update({'afi-safis': {'afi-safi': afi}})447 if tmp_capability:448 peer_group_cfg.update(tmp_capability)449 if tmp_remote:450 peer_group_cfg.update(tmp_remote)451 if peer_group_cfg:452 bgp_peer_group.update({'config': peer_group_cfg})453 if bgp_peer_group:454 bgp_peer_group_list.append(bgp_peer_group)455 payload = {'openconfig-network-instance:peer-groups': {'peer-group': bgp_peer_group_list}}456 return payload, requests457 def find_pg(self, have, bgp_as, vrf_name, peergroup):458 mat_dict = next((m_peer for m_peer in have if m_peer['bgp_as'] == bgp_as and m_peer['vrf_name'] == vrf_name), None)459 if mat_dict and mat_dict.get("peer_group", None) is not None:460 mat_pg = next((m for m in mat_dict['peer_group'] if m["name"] == peergroup['name']), None)461 return mat_pg462 def find_af(self, have, bgp_as, vrf_name, peergroup, afi, safi):463 mat_pg = self.find_pg(have, bgp_as, vrf_name, peergroup)464 if mat_pg and mat_pg['address_family'].get('afis', None) is not None:465 mat_af = next((af for af in mat_pg['address_family']['afis'] if af['afi'] == afi and af['safi'] == safi), None)466 return mat_af467 def find_nei(self, have, bgp_as, vrf_name, neighbor):468 mat_dict = next((m_neighbor for m_neighbor in have if m_neighbor['bgp_as'] == bgp_as and m_neighbor['vrf_name'] == vrf_name), None)469 if mat_dict and mat_dict.get("neighbors", None) is not None:470 mat_neighbor = next((m for m in mat_dict['neighbors'] if m["neighbor"] == neighbor['neighbor']), None)471 return mat_neighbor472 def build_bgp_neighbors_payload(self, cmd, have, bgp_as, vrf_name):473 bgp_neighbor_list = []474 requests = []475 for neighbor in cmd:476 if neighbor:477 bgp_neighbor = {}478 neighbor_cfg = {}479 tmp_bfd = {}480 tmp_ebgp = {}481 tmp_timers = {}482 tmp_capability = {}483 tmp_remote = {}484 tmp_transport = {}485 if neighbor.get('bfd', None) is not None:486 if neighbor['bfd'].get('enabled', None) is not None:487 tmp_bfd.update({'enabled': neighbor['bfd']['enabled']})488 if neighbor['bfd'].get('check_failure', None) is not None:489 tmp_bfd.update({'check-control-plane-failure': neighbor['bfd']['check_failure']})490 if neighbor['bfd'].get('profile', None) is not None:491 tmp_bfd.update({'bfd-profile': neighbor['bfd']['profile']})492 if neighbor.get('auth_pwd', None) is not None:493 if (neighbor['auth_pwd'].get('pwd', None) is not None and494 neighbor['auth_pwd'].get('encrypted', None) is not None):495 bgp_neighbor.update({'auth-password': {'config': {'password': neighbor['auth_pwd']['pwd'],496 'encrypted': neighbor['auth_pwd']['encrypted']}}})497 if neighbor.get('ebgp_multihop', None) is not None:498 if neighbor['ebgp_multihop'].get('enabled', None) is not None:499 tmp_ebgp.update({'enabled': neighbor['ebgp_multihop']['enabled']})500 if neighbor['ebgp_multihop'].get('multihop_ttl', None) is not None:501 tmp_ebgp.update({'multihop-ttl': neighbor['ebgp_multihop']['multihop_ttl']})502 if neighbor.get('timers', None) is not None:503 if neighbor['timers'].get('holdtime', None) is not None:504 tmp_timers.update({'hold-time': neighbor['timers']['holdtime']})505 if neighbor['timers'].get('keepalive', None) is not None:506 tmp_timers.update({'keepalive-interval': neighbor['timers']['keepalive']})507 if neighbor['timers'].get('connect_retry', None) is not None:508 tmp_timers.update({'connect-retry': neighbor['timers']['connect_retry']})509 if neighbor.get('capability', None) is not None:510 if neighbor['capability'].get('dynamic', None) is not None:511 tmp_capability.update({'capability-dynamic': neighbor['capability']['dynamic']})512 if neighbor['capability'].get('extended_nexthop', None) is not None:513 tmp_capability.update({'capability-extended-nexthop': neighbor['capability']['extended_nexthop']})514 if neighbor.get('advertisement_interval', None) is not None:515 tmp_timers.update({'minimum-advertisement-interval': neighbor['advertisement_interval']})516 if neighbor.get('neighbor', None) is not None:517 bgp_neighbor.update({'neighbor-address': neighbor['neighbor']})518 neighbor_cfg.update({'neighbor-address': neighbor['neighbor']})519 if neighbor.get('peer_group', None) is not None:520 neighbor_cfg.update({'peer-group': neighbor['peer_group']})521 if neighbor.get('nbr_description', None) is not None:522 neighbor_cfg.update({'description': neighbor['nbr_description']})523 if neighbor.get('disable_connected_check', None) is not None:524 neighbor_cfg.update({'disable-ebgp-connected-route-check': neighbor['disable_connected_check']})525 if neighbor.get('dont_negotiate_capability', None) is not None:526 neighbor_cfg.update({'dont-negotiate-capability': neighbor['dont_negotiate_capability']})527 if neighbor.get('enforce_first_as', None) is not None:528 neighbor_cfg.update({'enforce-first-as': neighbor['enforce_first_as']})529 if neighbor.get('enforce_multihop', None) is not None:530 neighbor_cfg.update({'enforce-multihop': neighbor['enforce_multihop']})531 if neighbor.get('override_capability', None) is not None:532 neighbor_cfg.update({'override-capability': neighbor['override_capability']})533 if neighbor.get('port', None) is not None:534 neighbor_cfg.update({'peer-port': neighbor['port']})535 if neighbor.get('shutdown_msg', None) is not None:536 neighbor_cfg.update({'shutdown-message': neighbor['shutdown_msg']})537 if neighbor.get('solo', None) is not None:538 neighbor_cfg.update({'solo-peer': neighbor['solo']})539 if neighbor.get('strict_capability_match', None) is not None:540 neighbor_cfg.update({'strict-capability-match': neighbor['strict_capability_match']})541 if neighbor.get('ttl_security', None) is not None:542 neighbor_cfg.update({'ttl-security-hops': neighbor['ttl_security']})543 if neighbor.get('v6only', None) is not None:544 neighbor_cfg.update({'openconfig-bgp-ext:v6only': neighbor['v6only']})545 if neighbor.get('local_as', None) is not None:546 if neighbor['local_as'].get('as', None) is not None:547 neighbor_cfg.update({'local-as': neighbor['local_as']['as']})548 if neighbor['local_as'].get('no_prepend', None) is not None:549 neighbor_cfg.update({'local-as-no-prepend': neighbor['local_as']['no_prepend']})550 if neighbor['local_as'].get('replace_as', None) is not None:551 neighbor_cfg.update({'local-as-replace-as': neighbor['local_as']['replace_as']})552 if neighbor.get('local_address', None) is not None:553 tmp_transport.update({'local-address': neighbor['local_address']})554 if neighbor.get('passive', None) is not None:555 tmp_transport.update({'passive-mode': neighbor['passive']})556 if neighbor.get('remote_as', None) is not None:557 have_nei = self.find_nei(have, bgp_as, vrf_name, neighbor)558 if neighbor['remote_as'].get('peer_as', None) is not None:559 if have_nei:560 if have_nei.get("remote_as", None) is not None:561 if have_nei["remote_as"].get("peer_type", None) is not None:562 del_nei = {}563 del_nei.update({'neighbor': have_nei['neighbor']})564 del_nei.update({'remote_as': have_nei['remote_as']})565 requests.extend(self.delete_specific_param_request(vrf_name, del_nei))566 tmp_remote.update({'peer-as': neighbor['remote_as']['peer_as']})567 if neighbor['remote_as'].get('peer_type', None) is not None:568 if have_nei:569 if have_nei.get("remote_as", None) is not None:570 if have_nei["remote_as"].get("peer_as", None) is not None:571 del_nei = {}572 del_nei.update({'neighbor': have_nei['neighbor']})573 del_nei.update({'remote_as': have_nei['remote_as']})574 requests.extend(self.delete_specific_param_request(vrf_name, del_nei))575 tmp_remote.update({'peer-type': neighbor['remote_as']['peer_type'].upper()})576 if tmp_bfd:577 bgp_neighbor.update({'enable-bfd': {'config': tmp_bfd}})578 if tmp_ebgp:579 bgp_neighbor.update({'ebgp-multihop': {'config': tmp_ebgp}})580 if tmp_timers:581 bgp_neighbor.update({'timers': {'config': tmp_timers}})582 if tmp_transport:583 bgp_neighbor.update({'transport': {'config': tmp_transport}})584 if tmp_capability:585 neighbor_cfg.update(tmp_capability)586 if tmp_remote:587 neighbor_cfg.update(tmp_remote)588 if neighbor_cfg:589 bgp_neighbor.update({'config': neighbor_cfg})590 if bgp_neighbor:591 bgp_neighbor_list.append(bgp_neighbor)592 payload = {'openconfig-network-instance:neighbors': {'neighbor': bgp_neighbor_list}}593 return payload, requests594 def get_modify_bgp_requests(self, commands, have):595 requests = []596 if not commands:597 return requests598 for cmd in commands:599 edit_path = '%s=%s/%s' % (self.network_instance_path, cmd['vrf_name'], self.protocol_bgp_path)600 if 'peer_group' in cmd and cmd['peer_group']:601 edit_peer_groups_payload, edit_requests = self.build_bgp_peer_groups_payload(cmd['peer_group'], have, cmd['bgp_as'], cmd['vrf_name'])602 edit_peer_groups_path = edit_path + '/peer-groups'603 if edit_requests:604 requests.extend(edit_requests)605 requests.append({'path': edit_peer_groups_path, 'method': PATCH, 'data': edit_peer_groups_payload})606 if 'neighbors' in cmd and cmd['neighbors']:607 edit_neighbors_payload, edit_requests = self.build_bgp_neighbors_payload(cmd['neighbors'], have, cmd['bgp_as'], cmd['vrf_name'])608 edit_neighbors_path = edit_path + '/neighbors'609 if edit_requests:610 requests.extend(edit_requests)611 requests.append({'path': edit_neighbors_path, 'method': PATCH, 'data': edit_neighbors_payload})612 return requests613 def get_delete_specific_bgp_peergroup_param_request(self, vrf_name, cmd, want_match):614 requests = []615 want_peer_group = want_match.get('peer_group', None)616 for each in cmd['peer_group']:617 if each:618 name = each.get('name', None)619 remote_as = each.get('remote_as', None)620 timers = each.get('timers', None)621 advertisement_interval = each.get('advertisement_interval', None)622 bfd = each.get('bfd', None)623 capability = each.get('capability', None)624 address_family = each.get('address_family', None)625 if name and not remote_as and not timers and not advertisement_interval and not bfd and not capability and not address_family:626 want_pg_match = None627 if want_peer_group:628 want_pg_match = next((cfg for cfg in want_peer_group if cfg['name'] == name), None)629 if want_pg_match:630 keys = ['remote_as', 'timers', 'advertisement_interval', 'bfd', 'capability', 'address_family']631 if not any(want_pg_match.get(key, None) for key in keys):632 requests.append(self.get_delete_vrf_specific_peergroup_request(vrf_name, name))633 else:634 requests.extend(self.delete_specific_peergroup_param_request(vrf_name, each))635 return requests636 def delete_specific_peergroup_param_request(self, vrf_name, cmd):637 requests = []638 delete_static_path = '%s=%s/%s' % (self.network_instance_path, vrf_name, self.protocol_bgp_path)639 delete_static_path = delete_static_path + '/peer-groups/peer-group=%s' % (cmd['name'])640 if cmd.get('remote_as', None) is not None:641 if cmd['remote_as'].get('peer_as', None) is not None:642 delete_path = delete_static_path + '/config/peer-as'643 requests.append({'path': delete_path, 'method': DELETE})644 elif cmd['remote_as'].get('peer_type', None) is not None:645 delete_path = delete_static_path + '/config/peer-type'646 requests.append({'path': delete_path, 'method': DELETE})647 if cmd.get('advertisement_interval', None) is not None:648 delete_path = delete_static_path + '/timers/config/minimum-advertisement-interval'649 requests.append({'path': delete_path, 'method': DELETE})650 if cmd.get('timers', None) is not None:651 if cmd['timers'].get('holdtime', None) is not None:652 delete_path = delete_static_path + '/timers/config/hold-time'653 requests.append({'path': delete_path, 'method': DELETE})654 if cmd['timers'].get('keepalive', None) is not None:655 delete_path = delete_static_path + '/timers/config/keepalive-interval'656 requests.append({'path': delete_path, 'method': DELETE})657 if cmd['timers'].get('connect_retry', None) is not None:658 delete_path = delete_static_path + '/timers/config/connect-retry'659 requests.append({'path': delete_path, 'method': DELETE})660 if cmd.get('capability', None) is not None:661 if cmd['capability'].get('dynamic', None) is not None:662 delete_path = delete_static_path + '/config/capability-dynamic'663 requests.append({'path': delete_path, 'method': DELETE})664 if cmd['capability'].get('extended_nexthop', None) is not None:665 delete_path = delete_static_path + '/config/capability-extended-nexthop'666 requests.append({'path': delete_path, 'method': DELETE})667 if cmd.get('pg_description', None) is not None:668 delete_path = delete_static_path + '/config/description'669 requests.append({'path': delete_path, 'method': DELETE})670 if cmd.get('disable_connected_check', None) is not None:671 delete_path = delete_static_path + '/config/disable-ebgp-connected-route-check'672 requests.append({'path': delete_path, 'method': DELETE})673 if cmd.get('dont_negotiate_capability', None) is not None:674 delete_path = delete_static_path + '/config/dont-negotiate-capability'675 requests.append({'path': delete_path, 'method': DELETE})676 if cmd.get('enforce_first_as', None) is not None:677 delete_path = delete_static_path + '/config/enforce-first-as'678 requests.append({'path': delete_path, 'method': DELETE})679 if cmd.get('enforce_multihop', None) is not None:680 delete_path = delete_static_path + '/config/enforce-multihop'681 requests.append({'path': delete_path, 'method': DELETE})682 if cmd.get('override_capability', None) is not None:683 delete_path = delete_static_path + '/config/override-capability'684 requests.append({'path': delete_path, 'method': DELETE})685 if cmd.get('shutdown_msg', None) is not None:686 delete_path = delete_static_path + '/config/shutdown-message'687 requests.append({'path': delete_path, 'method': DELETE})688 if cmd.get('solo', None) is not None:689 delete_path = delete_static_path + '/config/solo-peer'690 requests.append({'path': delete_path, 'method': DELETE})691 if cmd.get('strict_capability_match', None) is not None:692 delete_path = delete_static_path + '/config/strict-capability-match'693 requests.append({'path': delete_path, 'method': DELETE})694 if cmd.get('ttl_security', None) is not None:695 delete_path = delete_static_path + '/config/ttl-security-hops'696 requests.append({'path': delete_path, 'method': DELETE})697 if cmd.get('local_as', None) is not None:698 if cmd['local_as'].get('as', None) is not None:699 delete_path = delete_static_path + '/config/local-as'700 requests.append({'path': delete_path, 'method': DELETE})701 if cmd['local_as'].get('no_prepend', None) is not None:702 delete_path = delete_static_path + '/config/local-as-no-prepend'703 requests.append({'path': delete_path, 'method': DELETE})704 if cmd['local_as'].get('replace_as', None) is not None:705 delete_path = delete_static_path + '/config/local-as-replace-as'706 requests.append({'path': delete_path, 'method': DELETE})707 if cmd.get('local_address', None) is not None:708 delete_path = delete_static_path + '/transport/config/local-address'709 requests.append({'path': delete_path, 'method': DELETE})710 if cmd.get('passive', None) is not None:711 delete_path = delete_static_path + '/transport/config/passive-mode'712 requests.append({'path': delete_path, 'method': DELETE})713 if cmd.get('bfd', None) is not None:714 if cmd['bfd'].get('enabled', None) is not None:715 delete_path = delete_static_path + '/enable-bfd/config/enabled'716 requests.append({'path': delete_path, 'method': DELETE})717 if cmd['bfd'].get('check_failure', None) is not None:718 delete_path = delete_static_path + '/enable-bfd/config/check-control-plane-failure'719 requests.append({'path': delete_path, 'method': DELETE})720 if cmd['bfd'].get('profile', None) is not None:721 delete_path = delete_static_path + '/enable-bfd/config/bfd-profile'722 requests.append({'path': delete_path, 'method': DELETE})723 if cmd.get('auth_pwd', None) is not None:724 if cmd['auth_pwd'].get('pwd', None) is not None:725 delete_path = delete_static_path + '/auth-password/config/password'726 requests.append({'path': delete_path, 'method': DELETE})727 if cmd['auth_pwd'].get('encrypted', None) is not None:728 delete_path = delete_static_path + '/auth-password/config/encrypted'729 requests.append({'path': delete_path, 'method': DELETE})730 if cmd.get('ebgp_multihop', None) is not None:731 if cmd['ebgp_multihop'].get('enabled', None) is not None:732 delete_path = delete_static_path + '/ebgp-multihop/config/enabled'733 requests.append({'path': delete_path, 'method': DELETE})734 if cmd['ebgp_multihop'].get('multihop_ttl', None) is not None:735 delete_path = delete_static_path + '/ebgp-multihop/config/multihop_ttl'736 requests.append({'path': delete_path, 'method': DELETE})737 if cmd.get('address_family', None) is not None:738 if cmd['address_family'].get('afis', None) is None:739 delete_path = delete_static_path + '/afi-safis/afi-safi'740 requests.append({'path': delete_path, 'method': DELETE})741 else:742 for each in cmd['address_family']['afis']:743 afi = each.get('afi', None)744 safi = each.get('safi', None)745 activate = each.get('activate', None)746 allowas_in = each.get('allowas_in', None)747 afi_safi = afi.upper() + '_' + safi.upper()748 afi_safi_name = 'openconfig-bgp-types:' + afi_safi749 if afi and safi and not activate and not allowas_in:750 delete_path = delete_static_path + '/afi-safis/afi-safi=%s' % (afi_safi_name)751 requests.append({'path': delete_path, 'method': DELETE})752 else:753 if activate:754 delete_path = delete_static_path + '/afi-safis/afi-safi=%s/config/enabled' % (afi_safi_name)755 requests.append({'path': delete_path, 'method': DELETE})756 if allowas_in:757 if allowas_in.get('origin', None):758 delete_path = delete_static_path + '/afi-safis/afi-safi=%s/allow-own-as/config/origin' % (afi_safi_name)759 requests.append({'path': delete_path, 'method': DELETE})760 if allowas_in.get('value', None):761 delete_path = delete_static_path + '/afi-safis/afi-safi=%s/allow-own-as/config/as-count' % (afi_safi_name)762 requests.append({'path': delete_path, 'method': DELETE})763 return requests764 def get_delete_specific_bgp_param_request(self, vrf_name, cmd, want_match):765 requests = []766 want_neighbors = want_match.get('neighbors', None)767 for each in cmd['neighbors']:768 if each:769 neighbor = each.get('neighbor', None)770 remote_as = each.get('remote_as', None)771 peer_group = each.get('peer_group', None)772 timers = each.get('timers', None)773 advertisement_interval = each.get('advertisement_interval', None)774 bfd = each.get('bfd', None)775 capability = each.get('capability', None)776 if neighbor and not remote_as and not peer_group and not timers and not advertisement_interval and not bfd and not capability:777 want_nei_match = None778 if want_neighbors:779 want_nei_match = next(cfg for cfg in want_neighbors if cfg['neighbor'] == neighbor)780 if want_nei_match:781 keys = ['remote_as', 'peer_group', 'timers', 'advertisement_interval', 'bfd', 'capability']782 if not any(want_nei_match.get(key, None) for key in keys):783 requests.append(self.delete_neighbor_whole_request(vrf_name, neighbor))784 else:785 requests.extend(self.delete_specific_param_request(vrf_name, each))786 return requests787 def delete_neighbor_whole_request(self, vrf_name, neighbor):788 requests = []789 url = '%s=%s/%s/%s=%s/' % (self.network_instance_path, vrf_name, self.protocol_bgp_path, self.neighbor_path, neighbor)790 return({'path': url, 'method': DELETE})791 def delete_specific_param_request(self, vrf_name, cmd):792 requests = []793 delete_static_path = '%s=%s/%s' % (self.network_instance_path, vrf_name, self.protocol_bgp_path)794 delete_static_path = delete_static_path + '/neighbors/neighbor=%s' % (cmd['neighbor'])795 if cmd.get('remote_as', None) is not None:796 if cmd['remote_as'].get('peer_as', None) is not None:797 delete_path = delete_static_path + '/config/peer-as'798 requests.append({'path': delete_path, 'method': DELETE})799 elif cmd['remote_as'].get('peer_type', None) is not None:800 delete_path = delete_static_path + '/config/peer-type'801 requests.append({'path': delete_path, 'method': DELETE})802 if cmd.get('peer_group', None) is not None:803 delete_path = delete_static_path + '/config/peer-group'804 requests.append({'path': delete_path, 'method': DELETE})805 if cmd.get('nbr_description', None) is not None:806 delete_path = delete_static_path + '/config/description'807 requests.append({'path': delete_path, 'method': DELETE})808 if cmd.get('disable_connected_check', None) is not None:809 delete_path = delete_static_path + '/config/disable-ebgp-connected-route-check'810 requests.append({'path': delete_path, 'method': DELETE})811 if cmd.get('dont_negotiate_capability', None) is not None:812 delete_path = delete_static_path + '/config/dont-negotiate-capability'813 requests.append({'path': delete_path, 'method': DELETE})814 if cmd.get('enforce_first_as', None) is not None:815 delete_path = delete_static_path + '/config/enforce-first-as'816 requests.append({'path': delete_path, 'method': DELETE})817 if cmd.get('enforce_multihop', None) is not None:818 delete_path = delete_static_path + '/config/enforce-multihop'819 requests.append({'path': delete_path, 'method': DELETE})820 if cmd.get('override_capability', None) is not None:821 delete_path = delete_static_path + '/config/override-capability'822 requests.append({'path': delete_path, 'method': DELETE})823 if cmd.get('port', None) is not None:824 delete_path = delete_static_path + '/config/peer-port'825 requests.append({'path': delete_path, 'method': DELETE})826 if cmd.get('shutdown_msg', None) is not None:827 delete_path = delete_static_path + '/config/shutdown-message'828 requests.append({'path': delete_path, 'method': DELETE})829 if cmd.get('solo', None) is not None:830 delete_path = delete_static_path + '/config/solo-peer'831 requests.append({'path': delete_path, 'method': DELETE})832 if cmd.get('strict_capability_match', None) is not None:833 delete_path = delete_static_path + '/config/strict-capability-match'834 requests.append({'path': delete_path, 'method': DELETE})835 if cmd.get('ttl_security', None) is not None:836 delete_path = delete_static_path + '/config/ttl-security-hops'837 requests.append({'path': delete_path, 'method': DELETE})838 if cmd.get('v6only', None) is not None:839 delete_path = delete_static_path + '/config/openconfig-bgp-ext:v6only'840 requests.append({'path': delete_path, 'method': DELETE})841 if cmd.get('local_as', None) is not None:842 if cmd['local_as'].get('as', None) is not None:843 delete_path = delete_static_path + '/config/local-as'844 requests.append({'path': delete_path, 'method': DELETE})845 if cmd['local_as'].get('no_prepend', None) is not None:846 delete_path = delete_static_path + '/config/local-as-no-prepend'847 requests.append({'path': delete_path, 'method': DELETE})848 if cmd['local_as'].get('replace_as', None) is not None:849 delete_path = delete_static_path + '/config/local-as-replace-as'850 requests.append({'path': delete_path, 'method': DELETE})851 if cmd.get('local_address', None) is not None:852 delete_path = delete_static_path + '/transport/config/local-address'853 requests.append({'path': delete_path, 'method': DELETE})854 if cmd.get('passive', None) is not None:855 delete_path = delete_static_path + '/transport/config/passive-mode'856 requests.append({'path': delete_path, 'method': DELETE})857 if cmd.get('advertisement_interval', None) is not None:858 delete_path = delete_static_path + '/timers/config/minimum-advertisement-interval'859 requests.append({'path': delete_path, 'method': DELETE})860 if cmd.get('timers', None) is not None:861 if cmd['timers'].get('holdtime', None) is not None:862 delete_path = delete_static_path + '/timers/config/hold-time'863 requests.append({'path': delete_path, 'method': DELETE})864 if cmd['timers'].get('keepalive', None) is not None:865 delete_path = delete_static_path + '/timers/config/keepalive-interval'866 requests.append({'path': delete_path, 'method': DELETE})867 if cmd['timers'].get('connect_retry', None) is not None:868 delete_path = delete_static_path + '/timers/config/connect-retry'869 requests.append({'path': delete_path, 'method': DELETE})870 if cmd.get('capability', None) is not None:871 if cmd['capability'].get('dynamic', None) is not None:872 delete_path = delete_static_path + '/config/capability-dynamic'873 requests.append({'path': delete_path, 'method': DELETE})874 if cmd['capability'].get('extended_nexthop', None) is not None:875 delete_path = delete_static_path + '/config/capability-extended-nexthop'876 requests.append({'path': delete_path, 'method': DELETE})877 if cmd.get('bfd', None) is not None:878 if cmd['bfd'].get('enabled', None) is not None:879 delete_path = delete_static_path + '/enable-bfd/config/enabled'880 requests.append({'path': delete_path, 'method': DELETE})881 if cmd['bfd'].get('check_failure', None) is not None:882 delete_path = delete_static_path + '/enable-bfd/config/check-control-plane-failure'883 requests.append({'path': delete_path, 'method': DELETE})884 if cmd['bfd'].get('profile', None) is not None:885 delete_path = delete_static_path + '/enable-bfd/config/bfd-profile'886 requests.append({'path': delete_path, 'method': DELETE})887 if cmd.get('auth_pwd', None) is not None:888 if cmd['auth_pwd'].get('pwd', None) is not None:889 delete_path = delete_static_path + '/auth-password/config/password'890 requests.append({'path': delete_path, 'method': DELETE})891 if cmd['auth_pwd'].get('encrypted', None) is not None:892 delete_path = delete_static_path + '/auth-password/config/encrypted'893 requests.append({'path': delete_path, 'method': DELETE})894 if cmd.get('ebgp_multihop', None) is not None:895 if cmd['ebgp_multihop'].get('enabled', None) is not None:896 delete_path = delete_static_path + '/ebgp-multihop/config/enabled'897 requests.append({'path': delete_path, 'method': DELETE})898 if cmd['ebgp_multihop'].get('multihop_ttl', None) is not None:899 delete_path = delete_static_path + '/ebgp-multihop/config/multihop_ttl'900 requests.append({'path': delete_path, 'method': DELETE})901 return requests902 def get_delete_vrf_specific_neighbor_request(self, vrf_name, have):903 requests = []904 for each in have:905 if each.get('neighbor', None):906 requests.append(self.delete_neighbor_whole_request(vrf_name, each['neighbor']))907 return requests908 def get_delete_vrf_specific_peergroup_request(self, vrf_name, peergroup_name):909 requests = []910 delete_neighbor_path = '%s=%s/%s/peer-groups/peer-group=%s' % (self.network_instance_path, vrf_name, self.protocol_bgp_path, peergroup_name)911 return ({'path': delete_neighbor_path, 'method': DELETE})912 def get_delete_all_bgp_neighbor_requests(self, commands):913 requests = []914 for cmd in commands:915 if cmd.get('neighbors', None):916 requests.extend(self.get_delete_vrf_specific_neighbor_request(cmd['vrf_name'], cmd['neighbors']))917 if 'peer_group' in cmd and cmd['peer_group']:918 for each in cmd['peer_group']:919 requests.append(self.get_delete_vrf_specific_peergroup_request(cmd['vrf_name'], each['name']))920 return requests921 def get_delete_bgp_neighbor_requests(self, commands, have, want, is_delete_all):922 requests = []923 if is_delete_all:924 requests = self.get_delete_all_bgp_neighbor_requests(commands)925 else:926 for cmd in commands:927 vrf_name = cmd['vrf_name']928 as_val = cmd['bgp_as']929 neighbors = cmd.get('neighbors', None)930 peer_group = cmd.get('peer_group', None)931 want_match = next((cfg for cfg in want if vrf_name == cfg['vrf_name'] and as_val == cfg['bgp_as']), None)932 want_neighbors = want_match.get('neighbors', None)933 want_peer_group = want_match.get('peer_group', None)934 if neighbors is None and peer_group is None and want_neighbors is None and want_peer_group is None:935 new_cmd = {}936 for each in have:937 if vrf_name == each['vrf_name'] and as_val == each['bgp_as']:938 new_neighbors = []939 new_pg = []940 if each.get('neighbors', None):941 new_neighbors = [{'neighbor': i['neighbor']} for i in each.get('neighbors', None)]942 if each.get('peer_group', None):943 new_pg = [{'name': i['name']} for i in each.get('peer_group', None)]944 if new_neighbors:945 new_cmd['neighbors'] = new_neighbors946 requests.extend(self.get_delete_vrf_specific_neighbor_request(vrf_name, new_cmd['neighbors']))947 if new_pg:948 new_cmd['name'] = new_pg949 for each in new_cmd['name']:950 requests.append(self.get_delete_vrf_specific_peergroup_request(vrf_name, each['name']))951 break952 else:953 if neighbors:954 requests.extend(self.get_delete_specific_bgp_param_request(vrf_name, cmd, want_match))955 if peer_group:956 requests.extend(self.get_delete_specific_bgp_peergroup_param_request(vrf_name, cmd, want_match))...
delete.py
Source:delete.py
1import os2from toolbox_Chinese.merge import merge_files3from toolbox_Chinese.estimate_picth import *4import shutil5def delete_files(ori_path,singer_name,song_name):6 # ori_path = 'C:/DataBaker_songs'7 # singer_names = ['DB-DM-001-F-001','DB-DM-002-F-002','DB-DM-003-F-003','DB-DM-004-M-001',8 # 'DB-DM-005-M-002','DB-DM-006-F-004','DB-DM-007-M-003','DB-DM-008-M-004']9 #10 # todelete_files = ['nosilence.wav','small.wav','speech_']11 # for singer_name in singer_names:12 # second_path = ori_path + os.sep + singer_name + os.sep + 'Vox'13 # f_list = os.listdir(second_path)14 # Vox_list = []15 # for f in f_list:16 # if os.path.splitext(f)[1] == '.wav': # seperate with extension name17 # Vox_list.append(f.split('.')[0])18 # for song_name in Vox_list:19 delete_path = ori_path + os.sep + singer_name + os.sep + 'badsample_changepitch' + os.sep + song_name +os.sep20 delete_file1 = 'nosilence.wav'21 delete_file2 = 'small.wav'22 delete_file3 ='speech_' + song_name +'.wav'23 delete_dir1 = 'notescut'24 if os.path.exists(delete_path + delete_file1):25 os.remove(delete_path + delete_file1)26 if os.path.exists(delete_path + delete_file2):27 os.remove(delete_path +delete_file2)28 if os.path.exists(delete_path + delete_file3):29 os.remove(delete_path +delete_file3)30 if os.path.exists(delete_path + delete_dir1):31 shutil.rmtree(delete_path + delete_dir1)32 ###################################################33 # ori_path = 'C:/Dataset/NHSS_Database/Data'34 # singer_names = ['F01','F02','F03','F04','F05','M01','M02','M03','M04','M05']35 #36 # for singer_name in singer_names:37 # second_path = ori_path + os.sep + singer_name38 # f_list = os.listdir(second_path)39 # song_names = []40 #41 # for f in f_list:42 # if f.startswith('S'):43 # song_names.append(f)44 #45 # for song_name in song_names:46 # delete_path = ori_path + os.sep + singer_name + os.sep + song_name +os.sep47 # delete_file1 = 'song_nosilence.wav'48 # delete_file2 = 'small.wav'49 # delete_file3 ='speech.wav'50 #51 # if os.path.exists(delete_path + delete_file1):52 # os.remove(delete_path + delete_file1)53 # if os.path.exists(delete_path + delete_file2):54 # os.remove(delete_path +delete_file2)55 # if os.path.exists(delete_path + delete_file3):...
deletenotescut.py
Source:deletenotescut.py
1import os2import shutil3# from toolbox_Chinese.merge import merge_files4# from toolbox_Chinese.estimate_picth import *5def delete_files(ori_path,singer_name,song_name):6 delete_path = ori_path + os.sep + singer_name + os.sep + 'badsample_changerhythm' + os.sep + song_name +os.sep7 delete_file1 = 'nosilence.wav'8 # delete_file2 = 'small.wav'9 # delete_file3 ='speech_' + song_name +'.wav'10 delete_dir1 = 'notescut'11 if os.path.exists(delete_path + delete_file1):12 os.remove(delete_path + delete_file1)13 # if os.path.exists(delete_path + delete_file2):14 # os.remove(delete_path +delete_file2)15 # if os.path.exists(delete_path + delete_file3):16 # os.remove(delete_path +delete_file3)17 if os.path.exists(delete_path + delete_dir1):18 shutil.rmtree(delete_path + delete_dir1)19 ###################################################20 # ori_path = 'C:/Dataset/NHSS_Database/Data'21 # singer_names = ['F01','F02','F03','F04','F05','M01','M02','M03','M04','M05']22 #23 # for singer_name in singer_names:24 # second_path = ori_path + os.sep + singer_name25 # f_list = os.listdir(second_path)26 # song_names = []27 #28 # for f in f_list:29 # if f.startswith('S'):30 # song_names.append(f)31 #32 # for song_name in song_names:33 # delete_path = ori_path + os.sep + singer_name + os.sep + song_name +os.sep34 # delete_file1 = 'song_nosilence.wav'35 # delete_file2 = 'small.wav'36 # delete_file3 ='speech.wav'37 #38 # if os.path.exists(delete_path + delete_file1):39 # os.remove(delete_path + delete_file1)40 # if os.path.exists(delete_path + delete_file2):41 # os.remove(delete_path +delete_file2)42 # if os.path.exists(delete_path + delete_file3):...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run Airtest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
