Add openstack client tests

We need to test openstack client as it may differ from the openstack
API. Added some new openstack client tests together with the module
to execute openstack commands on the undercloud node.

Change-Id: Ia4960385c7f2b7565a5e9a5888a9cb5613f45520
This commit is contained in:
Alex Katz 2020-10-22 18:11:03 +03:00 committed by Federico Ressi
parent ad0881a25d
commit e759386c53
12 changed files with 626 additions and 2 deletions

View File

@ -1,2 +1,3 @@
pandas # BSD pandas # BSD
python-openstackclient # APACHE-2.0
validations-libs # APACHE-2.0 validations-libs # APACHE-2.0

View File

@ -0,0 +1,63 @@
# Copyright (c) 2020 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
from tobiko.openstack.openstackclient import _exception
from tobiko.openstack.openstackclient import _client
from tobiko.openstack.openstackclient import _network
from tobiko.openstack.openstackclient import _port
from tobiko.openstack.openstackclient import _security_group
from tobiko.openstack.openstackclient import _security_group_rule
from tobiko.openstack.openstackclient import _subnet
OSPCliError = _exception.OSPCliAuthError
OSPCliAuthError = _exception.OSPCliAuthError
execute = _client.execute
network_list = _network.network_list
network_show = _network.network_show
network_create = _network.network_create
network_delete = _network.network_delete
network_set = _network.network_set
network_unset = _network.network_unset
port_list = _port.port_list
port_show = _port.port_show
port_create = _port.port_create
port_delete = _port.port_delete
port_set = _port.port_set
port_unset = _port.port_unset
security_group_list = _security_group.security_group_list
security_group_show = _security_group.security_group_show
security_group_create = _security_group.security_group_create
security_group_delete = _security_group.security_group_delete
security_group_set = _security_group.security_group_set
security_group_unset = _security_group.security_group_unset
security_group_rule_list = _security_group_rule.security_group_rule_list
security_group_rule_show = _security_group_rule.security_group_rule_show
security_group_rule_create = _security_group_rule.security_group_rule_create
security_group_rule_delete = _security_group_rule.security_group_rule_delete
subnet_list = _subnet.subnet_list
subnet_show = _subnet.subnet_show
subnet_create = _subnet.subnet_create
subnet_delete = _subnet.subnet_delete
subnet_set = _subnet.subnet_set
subnet_unset = _subnet.subnet_unset

View File

@ -0,0 +1,77 @@
# Copyright (c) 2020 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
import json
from oslo_log import log
from tobiko.openstack import keystone
from tobiko.openstack.openstackclient import _exception
from tobiko.shell import sh
import tobiko.tripleo
LOG = log.getLogger(__name__)
def execute(cmd, *args, **kwargs):
arg_list = _param_list(*args, **kwargs)
cmd_to_exec = cmd.format(params=' '.join(arg_list))
if tobiko.tripleo.has_undercloud():
ssh_client = tobiko.tripleo.undercloud_ssh_client()
else:
ssh_client = None
try:
LOG.debug(f'Command to be executed:\n{cmd_to_exec}')
result = sh.execute(cmd_to_exec, ssh_client=ssh_client)
except sh.ShellCommandFailed as ex:
if ex.exit_status == 1:
raise _exception.OSPCliApiError(message=f'{ex.stderr}')
elif ex.exit_status == 2:
raise _exception.OSPCliClientError(message=f'{ex.stderr}')
else:
raise
output_format = kwargs.pop('format', '')
if output_format == 'json':
return json.loads(result.stdout)
else:
return dict()
def _param_list(*args, **kwargs):
if not any(param in kwargs for param in ['os-token', 'os-username']):
credentials = keystone.get_keystone_credentials()
kwargs['os-auth-url'] = credentials.auth_url
kwargs['os-password'] = credentials.password
kwargs['os-username'] = credentials.username
kwargs['os-project-name'] = credentials.project_name
if credentials.api_version == 3:
kwargs['os-user-domain-name'] = credentials.user_domain_name
kwargs['os-project-domain-name'] = credentials.project_domain_name
kwargs['os-identity-api-version'] = credentials.api_version
arg_list = []
for arg in args:
if len(arg) == 1:
arg_list.append(f'-{arg}')
else:
arg_list.append(f'--{arg}')
for arg, value in kwargs.items():
if len(arg) == 1:
arg_list.append(f'-{arg} {value}')
else:
arg_list.append(f'--{arg} {value}')
return arg_list

View File

@ -0,0 +1,34 @@
# Copyright (c) 2020 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
import tobiko
class OSPCliError(tobiko.TobikoException):
pass
class OSPCliAuthError(OSPCliError):
pass
class OSPCliClientError(OSPCliError):
pass
class OSPCliApiError(OSPCliError):
pass

View File

@ -0,0 +1,51 @@
# Copyright (c) 2020 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
from tobiko.openstack.openstackclient import _client
def network_list(*args, **kwargs):
cmd = 'openstack network list {params}'
kwargs['format'] = 'json'
return _client.execute(cmd, *args, **kwargs)
def network_show(network, *args, **kwargs):
cmd = f'openstack network show {{params}} {network}'
kwargs['format'] = 'json'
return _client.execute(cmd, *args, **kwargs)
def network_create(net_name, *args, **kwargs):
cmd = f'openstack network create {{params}} {net_name}'
kwargs['format'] = 'json'
return _client.execute(cmd, *args, **kwargs)
def network_delete(networks, *args, **kwargs):
cmd = f'openstack network delete {{params}} {" ".join(networks)}'
return _client.execute(cmd, *args, **kwargs)
def network_set(network, *args, **kwargs):
cmd = f'openstack network set {{params}} {network}'
return _client.execute(cmd, *args, **kwargs)
def network_unset(network, *args, **kwargs):
cmd = f'openstack network unset {{params}} {network}'
return _client.execute(cmd, *args, **kwargs)

View File

@ -0,0 +1,52 @@
# Copyright (c) 2020 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
from tobiko.openstack.openstackclient import _client
def port_list(*args, **kwargs):
cmd = 'openstack port list {params}'
kwargs['format'] = 'json'
return _client.execute(cmd, *args, **kwargs)
def port_show(port, *args, **kwargs):
cmd = f'openstack port show {{params}} {port}'
kwargs['format'] = 'json'
return _client.execute(cmd, *args, **kwargs)
def port_create(port_name, network_name, *args, **kwargs):
cmd = f'openstack port create {{params}} --network {network_name} '\
f'{port_name}'
kwargs['format'] = 'json'
return _client.execute(cmd, *args, **kwargs)
def port_delete(ports, *args, **kwargs):
cmd = f'openstack port delete {{params}} {" ".join(ports)}'
return _client.execute(cmd, *args, **kwargs)
def port_set(port, *args, **kwargs):
cmd = f'openstack port set {{params}} {port}'
return _client.execute(cmd, *args, **kwargs)
def port_unset(port, *args, **kwargs):
cmd = f'openstack port unset {{params}} {port}'
return _client.execute(cmd, *args, **kwargs)

View File

@ -0,0 +1,51 @@
# Copyright (c) 2020 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
from tobiko.openstack.openstackclient import _client
def security_group_list(*args, **kwargs):
cmd = 'openstack security group list {params}'
kwargs['format'] = 'json'
return _client.execute(cmd, *args, **kwargs)
def security_group_show(group, *args, **kwargs):
cmd = f'openstack security group show {{params}} {group}'
kwargs['format'] = 'json'
return _client.execute(cmd, *args, **kwargs)
def security_group_create(group_name, *args, **kwargs):
cmd = f'openstack security group create {{params}} {group_name}'
kwargs['format'] = 'json'
return _client.execute(cmd, *args, **kwargs)
def security_group_delete(groups, *args, **kwargs):
cmd = f'openstack security group delete {{params}} {" ".join(groups)}'
return _client.execute(cmd, *args, **kwargs)
def security_group_set(group, *args, **kwargs):
cmd = f'openstack security group set {{params}} {group}'
return _client.execute(cmd, *args, **kwargs)
def security_group_unset(group, *args, **kwargs):
cmd = f'openstack security group unset {{params}} {group}'
return _client.execute(cmd, *args, **kwargs)

View File

@ -0,0 +1,42 @@
# Copyright (c) 2020 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
from tobiko.openstack.openstackclient import _client
def security_group_rule_list(*args, **kwargs):
group = kwargs.pop('group', '')
cmd = f'openstack security group rule list {{params}} {group}'
kwargs['format'] = 'json'
return _client.execute(cmd, *args, **kwargs)
def security_group_rule_show(rule, *args, **kwargs):
cmd = f'openstack security group rule show {{params}} {rule}'
kwargs['format'] = 'json'
return _client.execute(cmd, *args, **kwargs)
def security_group_rule_create(group, *args, **kwargs):
cmd = f'openstack security group rule create {{params}} {group}'
kwargs['format'] = 'json'
return _client.execute(cmd, *args, **kwargs)
def security_group_rule_delete(rules, *args, **kwargs):
cmd = f'openstack security group rule delete {{params}} {" ".join(rules)}'
return _client.execute(cmd, *args, **kwargs)

View File

@ -0,0 +1,52 @@
# Copyright (c) 2020 Red Hat, Inc.
#
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
from tobiko.openstack.openstackclient import _client
def subnet_list(*args, **kwargs):
cmd = 'openstack subnet list {params}'
kwargs['format'] = 'json'
return _client.execute(cmd, *args, **kwargs)
def subnet_show(subnet, *args, **kwargs):
cmd = f'openstack subnet show {{params}} {subnet}'
kwargs['format'] = 'json'
return _client.execute(cmd, *args, **kwargs)
def subnet_create(subnet_name, network_name, *args, **kwargs):
cmd = f'openstack subnet create {{params}} --network '\
f'{network_name} {subnet_name}'
kwargs['format'] = 'json'
return _client.execute(cmd, *args, **kwargs)
def subnet_delete(subnets, *args, **kwargs):
cmd = f'openstack subnet delete {{params}} {" ".join(subnets)}'
return _client.execute(cmd, *args, **kwargs)
def subnet_set(subnet, *args, **kwargs):
cmd = f'openstack subnet set {{params}} {subnet}'
return _client.execute(cmd, *args, **kwargs)
def subnet_unset(subnet, *args, **kwargs):
cmd = f'openstack subnet unset {{params}} {subnet}'
return _client.execute(cmd, *args, **kwargs)

View File

@ -54,6 +54,7 @@ LocalShellProcessFixture = _local.LocalShellProcessFixture
LocalExecutePathFixture = _local.LocalExecutePathFixture LocalExecutePathFixture = _local.LocalExecutePathFixture
process = _process.process process = _process.process
str_from_stream = _process.str_from_stream
ShellProcessFixture = _process.ShellProcessFixture ShellProcessFixture = _process.ShellProcessFixture
PsError = _ps.PsError PsError = _ps.PsError

View File

@ -0,0 +1,199 @@
# Copyright (c) 2019 Red Hat
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import absolute_import
import random
import testtools
from oslo_log import log
from tobiko.openstack import neutron
from tobiko.openstack import openstackclient
LOG = log.getLogger(__name__)
class BaseCliTest(testtools.TestCase):
def setUp(self):
super(BaseCliTest, self).setUp()
self.api = neutron.get_neutron_client()
def api_network_delete(self, network):
nets = self.api.list_networks()['networks']
for net in nets:
if net['name'] == network:
self.api.delete_network(net['id'])
break
if net['id'] == network:
self.api.delete_network(network)
break
def api_subnet_delete(self, subnet_name):
subnets = self.api.list_subnets()['subnets']
for subnet in subnets:
if subnet['name'] == subnet_name:
self.api.delete_subnet(subnet['id'])
break
if subnet['id'] == subnet_name:
self.api.delete_subnet(subnet_name)
break
def api_port_delete(self, port_name):
ports = self.api.list_ports()['ports']
for port in ports:
if port['name'] == port_name:
self.api.delete_port(port['id'])
break
if port['id'] == port_name:
self.api.delete_port(port_name)
break
def api_random_port_create(self):
net_name = self.random_name()
port_name = self.random_name()
network = self.api.create_network({'network': {'name': net_name}})
self.addCleanup(self.api_network_delete, net_name)
network_id = network['network']['id']
self.api.create_port({'port': {'name': port_name,
'network_id': network_id}})
self.addCleanup(self.api_port_delete, port_name)
return port_name
def api_random_subnet_create(self):
net_name = self.random_name()
subnet_name = self.random_name()
network = self.api.create_network({'network': {'name': net_name}})
self.addCleanup(self.api_network_delete, net_name)
network_id = network['network']['id']
self.api.create_subnet({'subnet': {'name': subnet_name,
'network_id': network_id,
'ip_version': 4,
'cidr': '123.123.123.0/24'}})
return subnet_name
def api_random_network_create(self):
name = self.random_name()
self.api.create_network({'network': {'name': name}})
self.addCleanup(self.api_network_delete, name)
return name
def random_name(self, length=16):
letters = 'abcdefghijklmnopqrstuvwxyz'
random_string = ''.join(random.choice(letters) for i in range(length))
return f'{self.__class__.__name__}-{random_string}'
class NeutronCliNetwork(BaseCliTest):
def test_network_creation(self):
net_name = self.random_name()
output = openstackclient.network_create(net_name)
self.addCleanup(self.api_network_delete, net_name)
self.assertEqual(output['name'], net_name) # pylint: disable=E1126
self.assertEqual(output['status'], 'ACTIVE') # pylint: disable=E1126
def test_network_deletion(self):
net_name_1 = self.api_random_network_create()
net_name_2 = self.api_random_network_create()
openstackclient.network_delete([net_name_1, net_name_2])
nets = self.api.list_networks()['networks']
for net in nets:
self.assertNotEqual(net['name'], net_name_1)
self.assertNotEqual(net['name'], net_name_2)
def test_network_list(self):
net_name = self.api_random_network_create()
nets = openstackclient.network_list()
found = False
for net in nets:
if net['Name'] == net_name:
found = True
break
self.assertTrue(found)
def test_network_show(self):
net_name = self.api_random_network_create()
net = openstackclient.network_show(net_name)
self.assertEqual(net['name'], net_name) # pylint: disable=E1126
class NeutronCliSubnet(BaseCliTest):
def test_subnet_creation(self):
subnet_name = self.random_name()
net_name = self.api_random_network_create()
output = openstackclient.subnet_create(
subnet_name, net_name, **{'subnet-range': '123.123.123.0/24'})
self.assertEqual(output['name'], subnet_name) # pylint: disable=E1126
def test_subnet_deletion(self):
subnet_name_1 = self.api_random_subnet_create()
subnet_name_2 = self.api_random_subnet_create()
openstackclient.subnet_delete([subnet_name_1, subnet_name_2])
subnets = self.api.list_subnets()['subnets']
for subnet in subnets:
self.assertNotEqual(subnet['name'], subnet_name_1)
self.assertNotEqual(subnet['name'], subnet_name_2)
def test_subnet_list(self):
subnet_name = self.api_random_subnet_create()
subnets = openstackclient.subnet_list()
found = False
for subnet in subnets:
if subnet['Name'] == subnet_name:
found = True
break
self.assertTrue(found)
def test_subnet_show(self):
subnet_name = self.api_random_subnet_create()
subnet = openstackclient.subnet_show(subnet_name)
self.assertEqual(subnet['name'], subnet_name) # pylint: disable=E1126
class NeutronCliPort(BaseCliTest):
def test_port_creation(self):
port_name = self.random_name()
net_name = self.api_random_network_create()
output = openstackclient.port_create(port_name, net_name)
self.addCleanup(self.api_port_delete, port_name)
self.assertEqual(output['name'], port_name) # pylint: disable=E1126
def test_port_deletion(self):
port_name_1 = self.api_random_port_create()
port_name_2 = self.api_random_port_create()
openstackclient.port_delete([port_name_1, port_name_2])
ports = self.api.list_ports()['ports']
for port in ports:
self.assertNotEqual(port['name'], port_name_1)
self.assertNotEqual(port['name'], port_name_2)
def test_port_list(self):
port_name = self.api_random_port_create()
ports = openstackclient.port_list()
found = False
for port in ports:
if port['Name'] == port_name:
found = True
break
self.assertTrue(found)
def test_port_show(self):
port_name = self.api_random_port_create()
port = openstackclient.port_show(port_name)
self.assertEqual(port['name'], port_name) # pylint: disable=E1126

View File

@ -38,6 +38,7 @@ skip_if_missing_overcloud = overcloud.skip_if_missing_overcloud
TripleoTopology = topology.TripleoTopology TripleoTopology = topology.TripleoTopology
load_undercloud_rcfile = undercloud.load_undercloud_rcfile load_undercloud_rcfile = undercloud.load_undercloud_rcfile
has_undercloud = undercloud.has_undercloud
skip_if_missing_undercloud = undercloud.skip_if_missing_undercloud skip_if_missing_undercloud = undercloud.skip_if_missing_undercloud
undercloud_host_config = undercloud.undercloud_host_config undercloud_host_config = undercloud.undercloud_host_config
undercloud_keystone_client = undercloud.undercloud_keystone_client undercloud_keystone_client = undercloud.undercloud_keystone_client