Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Neutron gateway actions #611

Merged
merged 12 commits into from
Sep 13, 2021
73 changes: 38 additions & 35 deletions zaza/openstack/charm_tests/neutron/tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -274,18 +274,9 @@ def setUpClass(cls, application_name='neutron-gateway', model_alias=None):
current_release = openstack_utils.get_os_release()
bionic_train = openstack_utils.get_os_release('bionic_train')
xenial_mitaka = openstack_utils.get_os_release('xenial_mitaka')
cls.SKIP_LBAAS_TESTS = (xenial_mitaka <= current_release or
cls.SKIP_LBAAS_TESTS = (current_release <= xenial_mitaka or
current_release >= bionic_train)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

An alternative option/syntax:

 cls.SKIP_LBAAS_TESTS = not (xenial_mitaka > current_release < bionic_train)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

thanks, that does look better.


def tearDown(self):
"""Cleanup loadbalancers if there are any left over."""
super(NeutronGatewayStatusActionsTest, self).tearDown()
if not self.SKIP_LBAAS_TESTS:
load_balancers = self.neutron_client.list_loadbalancers().get(
'loadbalancers', [])
for lbaas in load_balancers:
self.neutron_client.delete_loadbalancer(lbaas['id'])

def _assert_result_match(self, action_result, resource_list,
resource_name):
"""Assert that action_result contains same data as resource_list."""
Expand Down Expand Up @@ -352,33 +343,45 @@ def test_get_status_load_balancers(self):
if self.SKIP_LBAAS_TESTS:
self.skipTest('LBaasV2 is not supported in this version.')

# create LBaasV2 for the purpose of this test
lbaas_name = 'test_lbaas'

subnet_list = self.neutron_client.list_subnets(
name='private_subnet').get('subnets', [])

if not subnet_list:
raise RuntimeError('Expected subnet "private_subnet" is not '
'configured.')
loadbalancer_id = None

subnet = subnet_list[0]
loadbalancer_data = {'loadbalancer': {'name': lbaas_name,
'vip_subnet_id': subnet['id']}}
self.neutron_client.create_loadbalancer(body=loadbalancer_data)

# test that client and action report same data
ngw_unit = zaza.model.get_units(self.application_name,
model_name=self.model_name)[0]
lbaas_from_client = self.neutron_client.list_loadbalancers().get(
'loadbalancers', [])

result = zaza.model.run_action(ngw_unit.entity_id,
'get-status-lb',
model_name=self.model_name,
action_params={"format": "json"})
try:
# create LBaasV2 for the purpose of this test
lbaas_name = 'test_lbaas'
subnet_list = self.neutron_client.list_subnets(
name='private_subnet').get('subnets', [])

if not subnet_list:
raise RuntimeError('Expected subnet "private_subnet" is not '
'configured.')

subnet = subnet_list[0]
loadbalancer_data = {'loadbalancer': {'name': lbaas_name,
'vip_subnet_id': subnet['id']
}
}
loadbalancer = self.neutron_client.create_loadbalancer(
body=loadbalancer_data)
loadbalancer_id = loadbalancer['loadbalancer']['id']

# test that client and action report same data
ngw_unit = zaza.model.get_units(self.application_name,
model_name=self.model_name)[0]
lbaas_from_client = self.neutron_client.list_loadbalancers().get(
'loadbalancers', [])

self._assert_result_match(result, lbaas_from_client, 'load-balancers')
result = zaza.model.run_action(ngw_unit.entity_id,
'get-status-lb',
model_name=self.model_name,
action_params={"format": "json"})

self._assert_result_match(result, lbaas_from_client,
'load-balancers')
except Exception as exc:
raise exc
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This isn't needed; the exception with happen if no exceptions are caught which is essentially the same thing.

finally:
if loadbalancer_id:
self.neutron_client.delete_loadbalancer(loadbalancer_id)


class NeutronCreateNetworkTest(test_utils.OpenStackBaseTest):
Expand Down