# Copyright 2014: Mirantis 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 unittest import mock import ddt from rally.common import cfg from rally_openstack.common import osclients from rally_openstack.verification.tempest import config from tests.unit import fakes from tests.unit import test CONF = cfg.CONF CRED = { "username": "admin", "tenant_name": "admin", "password": "admin-12345", "auth_url": "http://test:5000/v2.0/", "region_name": "test", "https_insecure": False, "https_cacert": "/path/to/cacert/file", "user_domain_name": "admin", "project_domain_name": "admin" } PATH = "rally_openstack.verification.tempest.config" @ddt.ddt class TempestConfigfileManagerTestCase(test.TestCase): def setUp(self): super(TempestConfigfileManagerTestCase, self).setUp() env = fakes.FakeEnvironment( env_uuid="fake_env", data={ "platforms": { "openstack": { "platform_data": { "admin": CRED } } } } ) with mock.patch("%s.credential.OpenStackCredential" % PATH, return_value=fakes.FakeCredential(**CRED)): self.tempest = config.TempestConfigfileManager(env) def test__configure_auth(self): self.tempest.conf.add_section("auth") self.tempest._configure_auth() expected = ( ("admin_username", CRED["username"]), ("admin_password", CRED["password"]), ("admin_project_name", CRED["tenant_name"]), ("admin_domain_name", CRED["user_domain_name"])) result = self.tempest.conf.items("auth") for item in expected: self.assertIn(item, result) @ddt.data("data_processing", "data-processing") def test__configure_data_processing(self, service_type): self.tempest.available_services = ["sahara"] self.tempest.clients.services.return_value = { service_type: "sahara"} self.tempest.conf.add_section("data-processing") self.tempest._configure_data_processing() self.assertEqual(service_type, self.tempest.conf.get("data-processing", "catalog_type")) @ddt.data( # The prefix "ex_" is abbreviation of "expected" # case #1: both versions are discoverable; version is in the auth_url {"auth_url": "http://example.com/v2.0", "data": [{"version": (3, 0), "url": "foo3.com"}, {"version": (2, 0), "url": "foo2.com"}], "ex_uri": "http://example.com/v2.0", "ex_auth_version": "v2", "ex_uri_v3": "http://example.com/v3"}, # case #2: the same case, but v3 is in the url {"auth_url": "http://example.com/v3", "data": [{"version": (3, 0), "url": "foo3.com"}, {"version": (2, 0), "url": "foo2.com"}], "ex_uri": "http://example.com/v2.0", "ex_auth_version": "v3", "ex_uri_v3": "http://example.com/v3"}, # case #3: both versions are discoverable; version is not in auth_url {"auth_url": "http://example.com", "data": [{"version": (3, 0), "url": "foo3.com"}, {"version": (2, 0), "url": "foo2.com"}], "ex_uri": "foo2.com", "ex_uri_v3": "foo3.com", "ex_auth_version": "v3"}, # case #4: the same case, but data in the another sort. {"auth_url": "http://example.com", "data": [{"version": (2, 0), "url": "foo2.com"}, {"version": (3, 0), "url": "foo3.com"}], "ex_uri": "foo2.com", "ex_uri_v3": "foo3.com", "ex_auth_version": "v3"}, # case #5: only one version is discoverable; {"auth_url": "http://example.com", "data": [{"version": (2, 0), "url": "foo2.com"}], "ex_uri": "foo2.com", "ex_auth_version": "v2", "ex_uri_v3": "http://example.com/v3"}, # case #6: the same case, but keystone v3 is discoverable {"auth_url": "http://example.com", "data": [{"version": (3, 0), "url": "foo3.com"}], "ex_uri": "http://example.com/v2.0", "ex_auth_version": "v3", "ex_uri_v3": "foo3.com", "ex_v2_off": True} ) @ddt.unpack def test__configure_identity(self, auth_url, data, ex_uri, ex_uri_v3, ex_auth_version, ex_v2_off=False): self.tempest.conf.add_section("identity") self.tempest.conf.add_section("identity-feature-enabled") self.tempest.credential.auth_url = auth_url process_url = osclients.Keystone( self.tempest.credential, 0)._remove_url_version self.tempest.clients.keystone._remove_url_version = process_url from keystoneauth1 import discover from keystoneauth1 import session with mock.patch.object(discover, "Discover") as mock_discover: with mock.patch.object(session, "Session") as mock_session: mock_discover.return_value.version_data.return_value = data self.tempest._configure_identity() mock_discover.assert_called_once_with( mock_session.return_value, auth_url) expected = {"region": CRED["region_name"], "auth_version": ex_auth_version, "uri": ex_uri, "uri_v3": ex_uri_v3, "disable_ssl_certificate_validation": str( CRED["https_insecure"]), "ca_certificates_file": CRED["https_cacert"]} self.assertEqual(expected, dict(self.tempest.conf.items("identity"))) if ex_v2_off: self.assertEqual( "False", self.tempest.conf.get("identity-feature-enabled", "api_v2")) # Test a conf setting with a None value try: self.tempest.conf.set("identity", "region", None) except TypeError as e: self.fail("self.tempest.conf.set('identity', 'region', None) " "raised a TypeError: " + str(e)) def test__configure_network_if_neutron(self): self.tempest.available_services = ["neutron"] client = self.tempest.clients.neutron() client.list_networks.return_value = { "networks": [ { "status": "ACTIVE", "id": "test_id", "name": "test_name", "router:external": True } ] } self.tempest.conf.add_section("network") self.tempest._configure_network() self.assertEqual("test_id", self.tempest.conf.get("network", "public_network_id")) self.assertEqual("test_name", self.tempest.conf.get("network", "floating_network_name")) def test__configure_network_if_nova(self): self.tempest.available_services = ["nova"] client = self.tempest.clients.nova() client.networks.list.return_value = [ mock.MagicMock(human_id="fake-network")] self.tempest.conf.add_section("compute") self.tempest.conf.add_section("validation") self.tempest._configure_network() expected = {"compute": ("fixed_network_name", "fake-network"), "validation": ("network_for_ssh", "fake-network")} for section, option in expected.items(): result = self.tempest.conf.items(section) self.assertIn(option, result) def test__configure_network_feature_enabled(self): self.tempest.available_services = ["neutron"] client = self.tempest.clients.neutron() client.list_ext.return_value = { "extensions": [ {"alias": "dvr"}, {"alias": "extra_dhcp_opt"}, {"alias": "extraroute"} ] } self.tempest.conf.add_section("network-feature-enabled") self.tempest._configure_network_feature_enabled() client.list_ext.assert_called_once_with("extensions", "/extensions", retrieve_all=True) self.assertEqual("dvr,extra_dhcp_opt,extraroute", self.tempest.conf.get("network-feature-enabled", "api_extensions")) def test__configure_object_storage(self): self.tempest.conf.add_section("object-storage") self.tempest._configure_object_storage() expected = ( ("operator_role", CONF.openstack.swift_operator_role), ("reseller_admin_role", CONF.openstack.swift_reseller_admin_role)) result = self.tempest.conf.items("object-storage") for item in expected: self.assertIn(item, result) def test__configure_orchestration(self): self.tempest.conf.add_section("orchestration") self.tempest._configure_orchestration() expected = ( ("stack_owner_role", CONF.openstack.heat_stack_owner_role), ("stack_user_role", CONF.openstack.heat_stack_user_role)) result = self.tempest.conf.items("orchestration") for item in expected: self.assertIn(item, result) def test__configure_service_available(self): available_services = ("nova", "cinder", "glance", "sahara") self.tempest.available_services = available_services self.tempest.conf.add_section("service_available") self.tempest._configure_service_available() expected = ( ("neutron", "False"), ("heat", "False"), ("nova", "True"), ("swift", "False"), ("cinder", "True"), ("sahara", "True"), ("glance", "True")) result = self.tempest.conf.items("service_available") for item in expected: self.assertIn(item, result) @ddt.data({}, {"service": "neutron", "connect_method": "floating"}) @ddt.unpack def test__configure_validation(self, service="nova", connect_method="fixed"): self.tempest.available_services = [service] self.tempest.conf.add_section("validation") self.tempest._configure_validation() expected = (("connect_method", connect_method), ) result = self.tempest.conf.items("validation") for item in expected: self.assertIn(item, result) @mock.patch("%s.io.StringIO" % PATH) @mock.patch("%s.open" % PATH, side_effect=mock.mock_open()) @mock.patch("inspect.getmembers") def test_create(self, mock_inspect_getmembers, mock_open, mock_string_io): configure_something_method = mock.MagicMock() mock_inspect_getmembers.return_value = [("_configure_something", configure_something_method)] self.tempest.conf.read = mock.Mock() self.tempest.conf.write = mock.Mock() self.tempest.conf.read.return_value = "[section]\noption = value" fake_extra_conf = {"section2": {"option2": "value2"}} self.tempest.create("/path/to/fake/conf", fake_extra_conf) self.assertEqual(1, configure_something_method.call_count) self.assertIn(("option2", "value2"), self.tempest.conf.items("section2")) mock_open.assert_called_once_with("/path/to/fake/conf", "w") self.tempest.conf.write.assert_has_calls( [mock.call(mock_open.side_effect()), mock.call(mock_string_io.return_value)]) mock_string_io.return_value.getvalue.assert_called_once_with()