Browse Source

Add force delete option for IP block subnet

If IP block is created by Policy API, we need
{'X-Allow-Overwrite': 'true'} in header to delete subnet

Change-Id: Ib05437fef32d2ea206ee5af67b441f710d6f5016
changes/30/746830/2
Danting Liu 11 months ago
parent
commit
be35d1dc11
  1. 23
      vmware_nsxlib/tests/unit/v3/test_load_balancer.py
  2. 3
      vmware_nsxlib/tests/unit/v3/test_qos_switching_profile.py
  3. 11
      vmware_nsxlib/tests/unit/v3/test_resources.py
  4. 5
      vmware_nsxlib/tests/unit/v3/test_security.py
  5. 8
      vmware_nsxlib/v3/core_resources.py
  6. 8
      vmware_nsxlib/v3/utils.py

23
vmware_nsxlib/tests/unit/v3/test_load_balancer.py

@ -152,7 +152,8 @@ class TestApplicationProfile(nsxlib_testcase.NsxClientTestCase):
self.nsxlib.load_balancer.application_profile.delete(
fake_profile['id'])
delete.assert_called_with(
'loadbalancer/application-profiles/%s' % fake_profile['id'])
'loadbalancer/application-profiles/%s' % fake_profile['id'],
headers=None)
class TestPersistenceProfile(nsxlib_testcase.NsxClientTestCase):
@ -193,7 +194,8 @@ class TestPersistenceProfile(nsxlib_testcase.NsxClientTestCase):
self.nsxlib.load_balancer.persistence_profile.delete(
fake_profile['id'])
delete.assert_called_with(
'loadbalancer/persistence-profiles/%s' % fake_profile['id'])
'loadbalancer/persistence-profiles/%s' % fake_profile['id'],
headers=None)
class TestRule(nsxlib_testcase.NsxClientTestCase):
@ -227,7 +229,7 @@ class TestRule(nsxlib_testcase.NsxClientTestCase):
fake_rule = consts.FAKE_RULE.copy()
self.nsxlib.load_balancer.rule.delete(fake_rule['id'])
delete.assert_called_with(
'loadbalancer/rules/%s' % fake_rule['id'])
'loadbalancer/rules/%s' % fake_rule['id'], headers=None)
class TestClientSslProfile(nsxlib_testcase.NsxClientTestCase):
@ -265,7 +267,8 @@ class TestClientSslProfile(nsxlib_testcase.NsxClientTestCase):
self.nsxlib.load_balancer.client_ssl_profile.delete(
fake_profile['id'])
delete.assert_called_with(
'loadbalancer/client-ssl-profiles/%s' % fake_profile['id'])
'loadbalancer/client-ssl-profiles/%s' % fake_profile['id'],
headers=None)
class TestServerSslProfile(nsxlib_testcase.NsxClientTestCase):
@ -303,7 +306,8 @@ class TestServerSslProfile(nsxlib_testcase.NsxClientTestCase):
self.nsxlib.load_balancer.server_ssl_profile.delete(
fake_profile['id'])
delete.assert_called_with(
'loadbalancer/server-ssl-profiles/%s' % fake_profile['id'])
'loadbalancer/server-ssl-profiles/%s' % fake_profile['id'],
headers=None)
class TestMonitor(nsxlib_testcase.NsxClientTestCase):
@ -341,7 +345,7 @@ class TestMonitor(nsxlib_testcase.NsxClientTestCase):
fake_monitor = consts.FAKE_MONITOR.copy()
self.nsxlib.load_balancer.monitor.delete(fake_monitor['id'])
delete.assert_called_with(
'loadbalancer/monitors/%s' % fake_monitor['id'])
'loadbalancer/monitors/%s' % fake_monitor['id'], headers=None)
class TestPool(nsxlib_testcase.NsxClientTestCase):
@ -378,7 +382,7 @@ class TestPool(nsxlib_testcase.NsxClientTestCase):
fake_profile = consts.FAKE_POOL.copy()
self.nsxlib.load_balancer.pool.delete(fake_profile['id'])
delete.assert_called_with(
'loadbalancer/pools/%s' % fake_profile['id'])
'loadbalancer/pools/%s' % fake_profile['id'], headers=None)
def test_remove_monitor_from_pool(self):
fake_pool = consts.FAKE_POOL.copy()
@ -463,7 +467,8 @@ class TestVirtualServer(nsxlib_testcase.NsxClientTestCase):
self.nsxlib.load_balancer.virtual_server.delete(
fake_virtual_server['id'])
delete.assert_called_with(
'loadbalancer/virtual-servers/%s' % fake_virtual_server['id'])
'loadbalancer/virtual-servers/%s' % fake_virtual_server['id'],
headers=None)
def test_add_rule(self):
fake_virtual_server = consts.FAKE_VIRTUAL_SERVER.copy()
@ -633,7 +638,7 @@ class TestService(nsxlib_testcase.NsxClientTestCase):
fake_service = consts.FAKE_SERVICE.copy()
self.nsxlib.load_balancer.service.delete(fake_service['id'])
delete.assert_called_with(
'loadbalancer/services/%s' % fake_service['id'])
'loadbalancer/services/%s' % fake_service['id'], headers=None)
def test_get_usage(self):
with mock.patch.object(self.nsxlib.client, 'get') as get:

3
vmware_nsxlib/tests/unit/v3/test_qos_switching_profile.py

@ -223,7 +223,8 @@ class NsxLibQosTestCase(nsxlib_testcase.NsxClientTestCase):
test_constants.FAKE_QOS_PROFILE['id'])
delete.assert_called_with(
'switching-profiles/%s'
% test_constants.FAKE_QOS_PROFILE['id'])
% test_constants.FAKE_QOS_PROFILE['id'],
headers=None)
def test_qos_switching_profile_set_shaping(self):
"""Test updating a qos-switching profile

11
vmware_nsxlib/tests/unit/v3/test_resources.py

@ -2349,6 +2349,17 @@ class NsxLibIpBlockSubnetTestCase(BaseTestResource):
(mocked_resource.uri_segment, block_id),
headers=self.default_headers())
def test_force_delete(self):
mocked_resource = self.get_mocked_resource()
subnet_id = 'subnet1'
headers = self.default_headers()
headers['X-Allow-Overwrite'] = 'true'
mocked_resource.delete(subnet_id, allow_overwrite=True)
test_client.assert_json_call(
'delete', mocked_resource,
'https://1.2.3.4/api/v1/pools/ip-subnets/%s' % subnet_id,
headers=headers)
class NsxLibIpBlockTestCase(BaseTestResource):

5
vmware_nsxlib/tests/unit/v3/test_security.py

@ -432,7 +432,8 @@ class TestNsxLibIPSet(nsxlib_testcase.NsxClientTestCase):
with mock.patch.object(self.nsxlib.client, 'delete') as delete:
fake_ip_set = test_constants.FAKE_IP_SET.copy()
self.nsxlib.ip_set.delete(fake_ip_set['id'])
delete.assert_called_with('ip-sets/%s' % fake_ip_set['id'])
delete.assert_called_with('ip-sets/%s' % fake_ip_set['id'],
headers=None)
def test_update_ip_set(self):
fake_ip_set = test_constants.FAKE_IP_SET.copy()
@ -593,7 +594,7 @@ class TestNsxLibNSGroup(nsxlib_testcase.NsxClientTestCase):
'delete') as del_mock:
self.nsxlib.ns_group.delete(ns_group_id)
del_mock.assert_called_with(
'ns-groups/%s?force=true' % ns_group_id)
'ns-groups/%s?force=true' % ns_group_id, headers=None)
def test_update_nsgroup_and_section(self):
name = 'name'

8
vmware_nsxlib/v3/core_resources.py

@ -1086,9 +1086,13 @@ class NsxLibIpBlockSubnet(utils.NsxLibApiBase):
headers = {'X-Allow-Overwrite': 'true'}
return self.client.create(self.get_path(), body, headers=headers)
def delete(self, subnet_id):
def delete(self, subnet_id, allow_overwrite=False):
"""Delete a IP block subnet on the backend."""
self._delete_with_retry(subnet_id)
headers = None
if allow_overwrite:
# Need to force delete subnet if IpBlock is created by Policy API
headers = {'X-Allow-Overwrite': 'true'}
self._delete_with_retry(subnet_id, headers=headers)
def list(self, ip_block_id):
resource = '%s?block_id=%s' % (self.get_path(), ip_block_id)

8
vmware_nsxlib/v3/utils.py

@ -477,15 +477,15 @@ class NsxLibApiBase(object):
action_params=action_params,
update_payload_cbk=update_payload_cbk)
def _delete_with_retry(self, resource):
self._delete_by_path_with_retry(self.get_path(resource))
def _delete_with_retry(self, resource, headers=None):
self._delete_by_path_with_retry(self.get_path(resource), headers)
def _delete_by_path_with_retry(self, path):
def _delete_by_path_with_retry(self, path, headers=None):
# Using internal method so we can access max_attempts in the decorator
@retry_upon_exception(nsxlib_exc.StaleRevision,
max_attempts=self.max_attempts)
def _do_delete():
self.client.delete(path)
self.client.delete(path, headers=headers)
_do_delete()

Loading…
Cancel
Save