arif/zaza-openstack-tests
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ea7336a67efe148ec707aa3d92cd20f5472bf995
zaza-openstack-tests/unit_tests/utilities/test_zaza_utilities_openstack.py
Frode Nordahl 2a566b5e05 Fix keystone tls detection and verification 
At present the check for which transport to use references a
configuration option that has been removed.

Update check to test for current configuration option plus
add a check for presence of `tls-certificates` relation.

Also remove insecure option and add verify option to allow
control of verification including providing own CA certificate
bundle.

Reference for verify option:
https://docs.openstack.org/keystoneauth/latest/api/keystoneauth1.html#keystoneauth1.session.Session
2018-07-04 13:17:25 +02:00

403 lines
17 KiB
Python
Raw Blame History

import copy
import mock
import tenacity
import unit_tests.utils as ut_utils
from zaza.utilities import openstack as openstack_utils
class TestOpenStackUtils(ut_utils.BaseTestCase):
def setUp(self):
super(TestOpenStackUtils, self).setUp()
self.port_name = "port_name"
self.net_uuid = "net_uuid"
self.project_id = "project_uuid"
self.ext_net = "ext_net"
self.private_net = "private_net"
self.port = {
"port": {"id": "port_id",
"name": self.port_name,
"network_id": self.net_uuid}}
self.ports = {"ports": [self.port["port"]]}
self.floatingip = {
"floatingip": {"id": "floatingip_id",
"floating_network_id": self.net_uuid,
"port_id": "port_id"}}
self.floatingips = {"floatingips": [self.floatingip["floatingip"]]}
self.address_scope_name = "address_scope_name"
self.address_scope = {
"address_scope": {"id": "address_scope_id",
"name": self.address_scope_name,
"shared": True,
"ip_version": 4,
"tenant_id": self.project_id}}
self.address_scopes = {
"address_scopes": [self.address_scope["address_scope"]]}
self.network = {
"network": {"id": "network_id",
"name": self.ext_net,
"tenant_id": self.project_id,
"router:external": True,
"provider:physical_network": "physnet1",
"provider:network_type": "flat"}}
self.networks = {
"networks": [self.network["network"]]}
self.neutronclient = mock.MagicMock()
self.neutronclient.list_ports.return_value = self.ports
self.neutronclient.create_port.return_value = self.port
self.neutronclient.list_floatingips.return_value = self.floatingips
self.neutronclient.create_floatingip.return_value = self.floatingip
self.neutronclient.list_address_scopes.return_value = (
self.address_scopes)
self.neutronclient.create_address_scope.return_value = (
self.address_scope)
self.neutronclient.list_networks.return_value = self.networks
self.neutronclient.create_network.return_value = self.network
def test_create_port(self):
self.patch_object(openstack_utils, "get_net_uuid")
self.get_net_uuid.return_value = self.net_uuid
# Already exists
port = openstack_utils.create_port(
self.neutronclient, self.port_name, self.private_net)
self.assertEqual(port, self.port["port"])
self.neutronclient.create_port.assert_not_called()
# Does not yet exist
self.neutronclient.list_ports.return_value = {"ports": []}
self.port["port"].pop("id")
port = openstack_utils.create_port(
self.neutronclient, self.port_name, self.private_net)
self.assertEqual(port, self.port["port"])
self.neutronclient.create_port.assert_called_once_with(self.port)
def test_create_floating_ip(self):
self.patch_object(openstack_utils, "get_net_uuid")
self.get_net_uuid.return_value = self.net_uuid
# Already exists
floatingip = openstack_utils.create_floating_ip(
self.neutronclient, self.ext_net, port=self.port["port"])
self.assertEqual(floatingip, self.floatingip["floatingip"])
self.neutronclient.create_floatingip.assert_not_called()
# Does not yet exist
self.neutronclient.list_floatingips.return_value = {"floatingips": []}
self.floatingip["floatingip"].pop("id")
floatingip = openstack_utils.create_floating_ip(
self.neutronclient, self.private_net, port=self.port["port"])
self.assertEqual(floatingip, self.floatingip["floatingip"])
self.neutronclient.create_floatingip.assert_called_once_with(
self.floatingip)
def test_create_address_scope(self):
self.patch_object(openstack_utils, "get_net_uuid")
self.get_net_uuid.return_value = self.net_uuid
# Already exists
address_scope = openstack_utils.create_address_scope(
self.neutronclient, self.project_id, self.address_scope_name)
self.assertEqual(address_scope, self.address_scope["address_scope"])
self.neutronclient.create_address_scope.assert_not_called()
# Does not yet exist
self.neutronclient.list_address_scopes.return_value = {
"address_scopes": []}
address_scope_msg = copy.deepcopy(self.address_scope)
address_scope_msg["address_scope"].pop("id")
address_scope = openstack_utils.create_address_scope(
self.neutronclient, self.project_id, self.address_scope_name)
self.assertEqual(address_scope, self.address_scope["address_scope"])
self.neutronclient.create_address_scope.assert_called_once_with(
address_scope_msg)
def test_create_external_network(self):
self.patch_object(openstack_utils, "get_net_uuid")
self.get_net_uuid.return_value = self.net_uuid
# Already exists
network = openstack_utils.create_external_network(
self.neutronclient, self.project_id, False)
self.assertEqual(network, self.network["network"])
self.neutronclient.create_network.assert_not_called()
# Does not yet exist
self.neutronclient.list_networks.return_value = {
"networks": []}
network_msg = copy.deepcopy(self.network)
network_msg["network"].pop("id")
network = openstack_utils.create_external_network(
self.neutronclient, self.project_id, False)
self.assertEqual(network, self.network["network"])
self.neutronclient.create_network.assert_called_once_with(
network_msg)
def test_get_keystone_scope(self):
self.patch_object(openstack_utils, "get_current_os_versions")
# <= Liberty
self.get_current_os_versions.return_value = {"keystone": "liberty"}
self.assertEqual(openstack_utils.get_keystone_scope(), "DOMAIN")
# > Liberty
self.get_current_os_versions.return_value = {"keystone": "mitaka"}
self.assertEqual(openstack_utils.get_keystone_scope(), "PROJECT")
def _test_get_overcloud_auth(self, tls_relation=False, ssl_cert=False,
v2_api=False):
self.patch_object(openstack_utils.model, 'get_relation_id')
self.patch_object(openstack_utils, 'get_application_config_option')
self.patch_object(openstack_utils, 'get_keystone_ip')
self.patch_object(openstack_utils, "get_current_os_versions")
self.get_keystone_ip.return_value = '127.0.0.1'
self.get_relation_id.return_value = None
self.get_application_config_option.return_value = None
if tls_relation or ssl_cert:
port = 35357
transport = 'https'
if tls_relation:
self.get_relation_id.return_value = 'tls-certificates:1'
if ssl_cert:
self.get_application_config_option.side_effect = [
'FAKECRTDATA',
None,
]
else:
port = 5000
transport = 'http'
if v2_api:
str_api = 'v2.0'
self.get_current_os_versions.return_value = {"keystone": "mitaka"}
expect = {
'OS_AUTH_URL': '{}://127.0.0.1:{}/{}'
.format(transport, port, str_api),
'OS_TENANT_NAME': 'admin',
'OS_USERNAME': 'admin',
'OS_PASSWORD': 'openstack',
'OS_REGION_NAME': 'RegionOne',
'API_VERSION': 2,
}
else:
str_api = 'v3'
self.get_current_os_versions.return_value = {"keystone": "queens"}
expect = {
'OS_AUTH_URL': '{}://127.0.0.1:{}/{}'
.format(transport, port, str_api),
'OS_USERNAME': 'admin',
'OS_PASSWORD': 'openstack',
'OS_REGION_NAME': 'RegionOne',
'OS_DOMAIN_NAME': 'admin_domain',
'OS_USER_DOMAIN_NAME': 'admin_domain',
'OS_PROJECT_NAME': 'admin',
'OS_PROJECT_DOMAIN_NAME': 'admin_domain',
'API_VERSION': 3,
}
self.assertEqual(openstack_utils.get_overcloud_auth(),
expect)
def test_get_overcloud_auth(self):
self._test_get_overcloud_auth()
def test_get_overcloud_auth_v2(self):
self._test_get_overcloud_auth(v2_api=True)
def test_get_overcloud_auth_tls_relation(self):
self._test_get_overcloud_auth(tls_relation=True)
def test_get_overcloud_auth_tls_relation_v2(self):
self._test_get_overcloud_auth(v2_api=True, tls_relation=True)
def test_get_overcloud_auth_ssl_cert(self):
self._test_get_overcloud_auth(ssl_cert=True)
def test_get_overcloud_auth_ssl_cert_v2(self):
self._test_get_overcloud_auth(v2_api=True, ssl_cert=True)
def test_get_overcloud_keystone_session(self):
self.patch_object(openstack_utils, "get_keystone_session")
self.patch_object(openstack_utils, "get_keystone_scope")
self.patch_object(openstack_utils, "get_overcloud_auth")
_auth = "FAKE_AUTH"
_scope = "PROJECT"
self.get_keystone_scope.return_value = _scope
self.get_overcloud_auth.return_value = _auth
openstack_utils.get_overcloud_keystone_session()
self.get_keystone_session.assert_called_once_with(_auth, scope=_scope,
verify=None)
def test_get_undercloud_keystone_session(self):
self.patch_object(openstack_utils, "get_keystone_session")
self.patch_object(openstack_utils, "get_keystone_scope")
self.patch_object(openstack_utils, "get_undercloud_auth")
_auth = "FAKE_AUTH"
_scope = "PROJECT"
self.get_keystone_scope.return_value = _scope
self.get_undercloud_auth.return_value = _auth
openstack_utils.get_undercloud_keystone_session()
self.get_keystone_session.assert_called_once_with(_auth, scope=_scope,
verify=None)
def test_get_urllib_opener(self):
self.patch_object(openstack_utils.urllib.request, "ProxyHandler")
self.patch_object(openstack_utils.urllib.request, "HTTPHandler")
self.patch_object(openstack_utils.urllib.request, "build_opener")
self.patch_object(openstack_utils.os, "getenv")
self.getenv.return_value = None
HTTPHandler_mock = mock.MagicMock()
self.HTTPHandler.return_value = HTTPHandler_mock
openstack_utils.get_urllib_opener()
self.build_opener.assert_called_once_with(HTTPHandler_mock)
self.HTTPHandler.assert_called_once_with()
def test_get_urllib_opener_proxy(self):
self.patch_object(openstack_utils.urllib.request, "ProxyHandler")
self.patch_object(openstack_utils.urllib.request, "HTTPHandler")
self.patch_object(openstack_utils.urllib.request, "build_opener")
self.patch_object(openstack_utils.os, "getenv")
self.getenv.return_value = 'http://squidy'
ProxyHandler_mock = mock.MagicMock()
self.ProxyHandler.return_value = ProxyHandler_mock
openstack_utils.get_urllib_opener()
self.build_opener.assert_called_once_with(ProxyHandler_mock)
self.ProxyHandler.assert_called_once_with({'http': 'http://squidy'})
def test_find_cirros_image(self):
urllib_opener_mock = mock.MagicMock()
self.patch_object(openstack_utils, "get_urllib_opener")
self.get_urllib_opener.return_value = urllib_opener_mock
urllib_opener_mock.open().read.return_value = b'12'
self.assertEqual(
openstack_utils.find_cirros_image('aarch64'),
'http://download.cirros-cloud.net/12/cirros-12-aarch64-disk.img')
def test_download_image(self):
urllib_opener_mock = mock.MagicMock()
self.patch_object(openstack_utils, "get_urllib_opener")
self.get_urllib_opener.return_value = urllib_opener_mock
self.patch_object(openstack_utils.urllib.request, "install_opener")
self.patch_object(openstack_utils.urllib.request, "urlretrieve")
openstack_utils.download_image('http://cirros/c.img', '/tmp/c1.img')
self.install_opener.assert_called_once_with(urllib_opener_mock)
self.urlretrieve.assert_called_once_with(
'http://cirros/c.img', '/tmp/c1.img')
def test_resource_reaches_status(self):
resource_mock = mock.MagicMock()
resource_mock.get.return_value = mock.MagicMock(status='available')
openstack_utils.resource_reaches_status(resource_mock, 'e01df65a')
def test_resource_reaches_status_fail(self):
openstack_utils.resource_reaches_status.retry.wait = \
tenacity.wait_none()
resource_mock = mock.MagicMock()
resource_mock.get.return_value = mock.MagicMock(status='unavailable')
with self.assertRaises(AssertionError):
openstack_utils.resource_reaches_status(
resource_mock,
'e01df65a')
def test_resource_reaches_status_bespoke(self):
resource_mock = mock.MagicMock()
resource_mock.get.return_value = mock.MagicMock(status='readyish')
openstack_utils.resource_reaches_status(
resource_mock,
'e01df65a',
'readyish')
def test_resource_reaches_status_bespoke_fail(self):
openstack_utils.resource_reaches_status.retry.wait = \
tenacity.wait_none()
resource_mock = mock.MagicMock()
resource_mock.get.return_value = mock.MagicMock(status='available')
with self.assertRaises(AssertionError):
openstack_utils.resource_reaches_status(
resource_mock,
'e01df65a',
'readyish')
def test_resource_removed(self):
resource_mock = mock.MagicMock()
resource_mock.list.return_value = [mock.MagicMock(id='ba8204b0')]
openstack_utils.resource_removed(resource_mock, 'e01df65a')
def test_resource_removed_fail(self):
openstack_utils.resource_reaches_status.retry.wait = \
tenacity.wait_none()
resource_mock = mock.MagicMock()
resource_mock.list.return_value = [mock.MagicMock(id='e01df65a')]
with self.assertRaises(AssertionError):
openstack_utils.resource_removed(resource_mock, 'e01df65a')
def test_delete_resource(self):
resource_mock = mock.MagicMock()
self.patch_object(openstack_utils, "resource_removed")
openstack_utils.delete_resource(resource_mock, 'e01df65a')
resource_mock.delete.assert_called_once_with('e01df65a')
self.resource_removed.assert_called_once_with(
resource_mock,
'e01df65a',
'resource')
def test_delete_image(self):
self.patch_object(openstack_utils, "delete_resource")
glance_mock = mock.MagicMock()
openstack_utils.delete_image(glance_mock, 'b46c2d83')
self.delete_resource.assert_called_once_with(
glance_mock.images,
'b46c2d83',
msg="glance image")
def test_upload_image_to_glance(self):
self.patch_object(openstack_utils, "resource_reaches_status")
glance_mock = mock.MagicMock()
image_mock = mock.MagicMock(id='9d1125af')
glance_mock.images.create.return_value = image_mock
m = mock.mock_open()
with mock.patch('zaza.utilities.openstack.open', m, create=False) as f:
openstack_utils.upload_image_to_glance(
glance_mock,
'/tmp/im1.img',
'bob')
glance_mock.images.create.assert_called_once_with(
name='bob',
disk_format='qcow2',
visibility='public',
container_format='bare')
glance_mock.images.upload.assert_called_once_with(
'9d1125af',
f(),
)
self.resource_reaches_status.assert_called_once_with(
glance_mock.images,
'9d1125af',
expected_status='active',
msg='Image status wait')
def test_create_image(self):
glance_mock = mock.MagicMock()
self.patch_object(openstack_utils.os.path, "exists")
self.patch_object(openstack_utils, "download_image")
self.patch_object(openstack_utils, "upload_image_to_glance")
openstack_utils.create_image(
glance_mock,
'http://cirros/c.img',
'bob')
self.exists.return_value = False
self.download_image.assert_called_once_with(
'http://cirros/c.img',
'tests/c.img')
self.upload_image_to_glance.assert_called_once_with(
glance_mock,
'tests/c.img',
'bob')
Reference in New Issue View Git Blame Copy Permalink