Support force update logical port

Sometimes, user want to use X-Allow-Overwrite header set to true to
force update the logical port

Change-Id: I3a82f73d67766b2105a6ffa4f4c465fb7b4a56e7
This commit is contained in:
Xiaopei Liu 2019-06-25 01:41:24 +08:00
parent f4731dd9a5
commit eadaef448e
2 changed files with 34 additions and 2 deletions

View File

@ -801,6 +801,34 @@ class LogicalPortTestCase(BaseTestResource):
data=jsonutils.dumps(fake_port, sort_keys=True),
headers=self.default_headers())
def test_update_port_with_force_true(self):
"""Test modifying tags using tags attribute with force true"""
fake_port = copy.deepcopy(test_constants.FAKE_PORT)
orig_tags = [{'scope': 'a1', 'tag': 'b1'},
{'scope': 'a2', 'tag': 'b2'}]
fake_port['tags'] = orig_tags
# Add a new tag
new_tags = [{'scope': 'a3', 'tag': 'b3'}]
mocked_resource = self.get_mocked_resource()
def get_fake_port(*args, **kwargs):
return copy.copy(fake_port)
mocked_resource.client.get = get_fake_port
mocked_resource.update(
fake_port['id'],
fake_port['attachment']['id'],
tags=new_tags, force=True)
# update expected result:
fake_port['tags'] = new_tags
headers = self.default_headers()
headers['X-Allow-Overwrite'] = 'true'
test_client.assert_json_call(
'put', mocked_resource,
'https://1.2.3.4/api/v1/logical-ports/%s' % fake_port['id'],
data=jsonutils.dumps(fake_port, sort_keys=True),
headers=headers)
class LogicalRouterTestCase(BaseTestResource):

View File

@ -195,7 +195,7 @@ class LogicalPort(utils.NsxLibApiBase):
vif_type=None, app_id=None,
allocate_addresses=nsx_constants.ALLOCATE_ADDRESS_NONE,
description=None, tn_uuid=None,
extra_configs=None):
extra_configs=None, force=False):
# Do not allow tags & tags_update at the same call
if tags_update and tags:
raise exceptions.ManagerError(
@ -219,8 +219,12 @@ class LogicalPort(utils.NsxLibApiBase):
extra_configs=extra_configs,
tags=tags))
headers = None
if force:
headers = {'X-Allow-Overwrite': 'true'}
return self._update_resource(
self.get_path(lport_id), lport, retry=True)
self.get_path(lport_id), lport, headers=headers, retry=True)
def get_by_attachment(self, attachment_type, attachment_id):
"""Return all logical port matching the attachment type and Id"""