Fix aggregate functional test for id restriction

Nova just landed a change that validates aggregate ids are ints.
This had a side-effect of causing our aggregate test to fail,
because we were sending a delete by name not id. It was failing
open for us - the delete would have likely not done anything
before.

Change-Id: I1e151454f39d51db52cdc7a23dbdcbf9b28c9381
This commit is contained in:
Monty Taylor 2020-03-10 08:17:10 -05:00
parent 1ad44c8fa1
commit a0e68c318f
3 changed files with 32 additions and 2 deletions

View File

@ -1633,6 +1633,16 @@ class ComputeCloudMixin(_normalize.Normalizer):
:raises: OpenStackCloudException on operation error. :raises: OpenStackCloudException on operation error.
""" """
if (
isinstance(name_or_id, six.string_types + (six.binary_type,))
and not name_or_id.isdigit()
):
aggregate = self.get_aggregate(name_or_id)
if not aggregate:
self.log.debug(
"Aggregate %s not found for deleting", name_or_id)
return False
name_or_id = aggregate.id
try: try:
self.compute.delete_aggregate(name_or_id, ignore_missing=False) self.compute.delete_aggregate(name_or_id, ignore_missing=False)
return True return True

View File

@ -50,9 +50,11 @@ class TestAggregate(base.BaseFunctionalTest):
) )
self.assertNotIn('key', aggregate['metadata']) self.assertNotIn('key', aggregate['metadata'])
self.operator_cloud.delete_aggregate(aggregate_name) # Validate that we can delete by name
self.assertTrue(
self.operator_cloud.delete_aggregate(aggregate_name))
def cleanup(self, aggregate_name): def cleanup(self, aggregate_name):
aggregate = self.operator_cloud.get_aggregate(aggregate_name) aggregate = self.operator_cloud.get_aggregate(aggregate_name)
if aggregate: if aggregate:
self.operator_cloud.delete_aggregate(aggregate_name) self.operator_cloud.delete_aggregate(aggregate['id'])

View File

@ -78,6 +78,24 @@ class TestAggregate(base.TestCase):
self.assert_calls() self.assert_calls()
def test_delete_aggregate_by_name(self):
self.register_uris([
dict(
method='GET',
uri=self.get_mock_url(
'compute', 'public', append=['os-aggregates']
),
json={'aggregates': [self.fake_aggregate]},
),
dict(method='DELETE',
uri=self.get_mock_url(
'compute', 'public', append=['os-aggregates', '1'])),
])
self.assertTrue(self.cloud.delete_aggregate(self.aggregate_name))
self.assert_calls()
def test_update_aggregate_set_az(self): def test_update_aggregate_set_az(self):
self.register_uris([ self.register_uris([
dict(method='GET', dict(method='GET',