Juju Charm - Ironic Conductor
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

192 lines
7.2 KiB

# Copyright 2020 Canonical Ltd
#
# 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.
import mock
import charms_openstack.test_utils as test_utils
from keystoneauth1 import exceptions as ks_exc
import charm.openstack.ironic.api_utils as api_utils
class _NotFoundException(Exception):
pass
class TestGetKeystoneSession(test_utils.PatchHelper):
def setUp(self):
super().setUp()
self.ks_int = mock.MagicMock()
self.ks_int.credentials_username.return_value = "ironic"
self.ks_int.credentials_password.return_value = "super_secret"
self.ks_int.credentials_project.return_value = "services"
self.ks_int.auth_protocol.return_value = "https"
self.ks_int.auth_host.return_value = "example.com"
self.ks_int.credentials_port.return_value = "5000"
self.ks_int.api_version.return_value = "3"
self.ks_int.credentials_project_domain_name.return_value = "default"
self.ks_int.credentials_user_domain_name.return_value = "default"
self.ks_expect = {
"username": "ironic",
"password": "super_secret",
"auth_url": "https://example.com:5000",
"project_name": "services",
"project_domain_name": "default",
"user_domain_name": "default",
}
def test_create_keystone_session(self):
self.patch_object(api_utils, 'loading')
self.patch_object(api_utils, 'ks_session')
loader = mock.MagicMock()
auth = mock.MagicMock()
loader.load_from_options.return_value = auth
self.loading.get_plugin_loader.return_value = loader
api_utils.create_keystone_session(self.ks_int)
self.loading.get_plugin_loader.assert_called_with("v3password")
loader.load_from_options.assert_called_with(**self.ks_expect)
self.ks_session.Session.assert_called_with(
auth=auth, verify=api_utils.SYSTEM_CA_BUNDLE)
def test_create_keystone_session_v2(self):
self.patch_object(api_utils, 'loading')
self.patch_object(api_utils, 'ks_session')
loader = mock.MagicMock()
auth = mock.MagicMock()
loader.load_from_options.return_value = auth
self.loading.get_plugin_loader.return_value = loader
self.ks_int.api_version.return_value = "v2.0"
api_utils.create_keystone_session(self.ks_int)
del self.ks_expect["project_domain_name"]
del self.ks_expect["user_domain_name"]
self.loading.get_plugin_loader.assert_called_with("password")
loader.load_from_options.assert_called_with(**self.ks_expect)
self.ks_session.Session.assert_called_with(
auth=auth, verify=api_utils.SYSTEM_CA_BUNDLE)
class TestOSClients(test_utils.PatchHelper):
def setUp(self):
super().setUp()
self.session = mock.MagicMock()
self.stores = {
"stores": [
{"id": "swift"},
{"id": "local"},
{"id": "ceph", "default": True}
]
}
self.patch_object(
api_utils.glanceclient,
'Client', name="glance_client")
self.patch_object(
api_utils.swiftclient,
'Connection', name="swift_con")
self.patch_object(
api_utils.keystoneclient.v3,
'Client', name="ks_client")
self.mocked_glance = mock.MagicMock()
self.glance_client.return_value = self.mocked_glance
self.mocked_glance.images.get_stores_info.return_value = self.stores
self.mocked_swift = mock.MagicMock()
self.swift_con.return_value = self.mocked_swift
self.mocked_ks = mock.MagicMock()
self.ks_client.return_value = self.mocked_ks
self.target = api_utils.OSClients(self.session)
self.glance_client.assert_called_with(session=self.session, version=2)
self.swift_con.assert_called_with(
session=self.session, cacert=api_utils.SYSTEM_CA_BUNDLE)
self.ks_client.assert_called_with(session=self.session)
def test_stores_info(self):
self.assertEqual(self.target._stores_info, self.stores["stores"])
def test_glance_stores(self):
self.assertEqual(
self.target.glance_stores,
[i["id"] for i in self.stores["stores"]])
def test_default_glance_store(self):
self.assertEqual(self.target.get_default_glance_store(), "ceph")
del self.stores["stores"][2]["default"]
self.stores["stores"][1]["default"] = True
self.assertEqual(self.target.get_default_glance_store(), "local")
def test_get_object_account_properties(self):
props = {
"x-account-meta-fakeprop": "hi there",
"x-account-meta-fakeprop2": "bye there",
"bogus": "won't be here in result"
}
expected_result = {
"fakeprop": "hi there",
"fakeprop2": "bye there",
}
self.mocked_swift.get_account.return_value = (props, "")
result = self.target.get_object_account_properties()
self.assertEqual(result, expected_result)
def test_set_object_account_property(self):
props = {
"x-account-meta-fakeprop": "hi there",
}
self.mocked_swift.get_account.return_value = (props, "")
self.target.set_object_account_property("FaKePrOp", "hi there")
self.mocked_swift.post_account.assert_not_called()
self.target.set_object_account_property("FaKePrOp2", "bye there")
self.mocked_swift.post_account.assert_called_with(
{"x-account-meta-fakeprop2": "bye there"})
def test_delete_object_account_property(self):
self.target.delete_object_account_property("FaKePrOp2")
self.mocked_swift.post_account.assert_called_with(
{"x-account-meta-fakeprop2": ""})
def test_has_service_type(self):
mocked_svc_find = mock.MagicMock()
mocked_svc_find.id = "fakeid"
self.ks_client.services.find.return_value = mocked_svc_find
self.target._ks = self.ks_client
result = self.target._has_service_type(
"object-store", interface="public")
self.ks_client.services.find.assert_called_with(
type="object-store")
self.ks_client.endpoints.find.assert_called_with(
service_id="fakeid", interface="public")
self.assertTrue(result)
def test_does_not_have_service_type(self):
ks_exc.http.NotFound = _NotFoundException
self.ks_client.services.find.side_effect = ks_exc.http.NotFound()
self.target._ks = self.ks_client
result = self.target._has_service_type(
"object-store", interface="public")
self.ks_client.services.find.assert_called_with(
type="object-store")
self.ks_client.endpoints.find.assert_not_called()
self.assertFalse(result)