From 11f17fc16228ff667b2ad01d05d29711719ce8b5 Mon Sep 17 00:00:00 2001 From: leilei9547 Date: Sun, 7 Sep 2014 21:35:53 -0700 Subject: [PATCH] Add unit test for metadata and adapter Change-Id: I2d88c07a3c078b25d38e962eb3d0e5baf0b87182 --- compass/tests/db/api/test_adapter_holder.py | 192 +++++++++++++++++ compass/tests/db/api/test_metadata_holder.py | 209 ++++++++++++++++++- 2 files changed, 397 insertions(+), 4 deletions(-) create mode 100644 compass/tests/db/api/test_adapter_holder.py diff --git a/compass/tests/db/api/test_adapter_holder.py b/compass/tests/db/api/test_adapter_holder.py new file mode 100644 index 00000000..9689d45f --- /dev/null +++ b/compass/tests/db/api/test_adapter_holder.py @@ -0,0 +1,192 @@ +# Copyright 2014 Huawei Technologies Co. 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 datetime +import logging +import mock +import os +import unittest2 + + +os.environ['COMPASS_IGNORE_SETTING'] = 'true' + + +from compass.utils import setting_wrapper as setting +reload(setting) + + +from base import BaseTest +from compass.db.api import adapter as adapter_api +from compass.db.api import adapter_holder as adapter +from compass.db.api import database +from compass.db.api import metadata as metadata_api +from compass.db.api import metadata_holder as metadata +from compass.db.api import user as user_api +from compass.db import exception +from compass.utils import flags +from compass.utils import logsetting +from compass.utils import util + + +class AdapterTestCase(unittest2.TestCase): + """Adapter base test case.""" + + def setUp(self): + super(AdapterTestCase, self).setUp() + reload(setting) + setting.CONFIG_DIR = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + 'data' + ) + database.init('sqlite://') + database.create_db() + self.user_object = ( + user_api.get_user_object( + setting.COMPASS_ADMIN_EMAIL + ) + ) + + mock_config = mock.Mock() + self.backup_adapter_configs = util.load_configs + util.load_configs = mock_config + configs = [{ + 'NAME': 'openstack_test', + 'DISLAY_NAME': 'Test OpenStack Icehouse', + 'PACKAGE_INSTALLER': 'chef_installer', + 'OS_INSTALLER': 'cobbler', + 'SUPPORTED_OS_PATTERNS': ['(?i)centos.*', '(?i)ubuntu.*'], + 'DEPLOYABLE': True + }] + util.load_configs.return_value = configs + with database.session() as session: + adapter_api.add_adapters_internal(session) + adapter.load_adapters() + self.adapter_object = adapter.list_adapters(self.user_object) + for adapter_obj in self.adapter_object: + if adapter_obj['name'] == 'openstack_icehouse': + self.adapter_id = adapter_obj['id'] + break + + def tearDown(self): + super(AdapterTestCase, self).tearDown() + util.load_configs = self.backup_adapter_configs + reload(setting) + database.drop_db() + + +class TestListAdapters(AdapterTestCase): + """Test list adapters.""" + + def setUp(self): + super(TestListAdapters, self).setUp() + + def tearDown(self): + super(TestListAdapters, self).tearDown() + + def test_list_adapters(self): + adapters = adapter.list_adapters( + self.user_object + ) + result = [] + for item in adapters: + result.append(item['name']) + expects = [ + 'openstack_icehouse', + 'os_only', + 'ceph(chef)', + 'openstack_test' + ] + self.assertIsNotNone(adapters) + for expect in expects: + self.assertIn(expect, result) + + +class TestGetAdapter(AdapterTestCase): + """Test get adapter.""" + + def setUp(self): + super(TestGetAdapter, self).setUp() + + def tearDown(self): + super(TestGetAdapter, self).tearDown() + + def test_get_adapter(self): + get_adapter = adapter.get_adapter( + self.user_object, + self.adapter_id + ) + name = None + for k, v in get_adapter.items(): + if k == 'name': + name = v + self.assertIsNotNone(get_adapter) + self.assertEqual(name, 'openstack_icehouse') + + def test_adapter_not_exist(self): + self.assertRaises( + exception.RecordNotExists, + adapter.get_adapter, + self.user_object, + 99 + ) + + +class TestGetAdapterRoles(AdapterTestCase): + """Test get adapter roles.""" + + def setUp(self): + super(TestGetAdapterRoles, self).setUp() + + def tearDown(self): + super(TestGetAdapterRoles, self).tearDown() + + def test_get_adapter_roles(self): + adapter_roles = adapter.get_adapter_roles( + self.user_object, + self.adapter_id + ) + result = [] + for adapter_role in adapter_roles: + for k, v in adapter_role.items(): + if k == 'display_name': + result.append(v) + expected = [ + 'compute node', + 'network node', + 'storage node', + 'image node', + 'vnc proxy node', + 'controller node', + 'message queue node', + 'database node', + 'ha proxy node', + 'all in one compute' + ] + self.assertIsNotNone(adapter_roles) + self.assertListEqual(result, expected) + + def test_adapter_non_exist(self): + self.assertRaises( + exception.RecordNotExists, + adapter.get_adapter_roles, + self.user_object, + 99 + ) + + +if __name__ == '__main__': + flags.init() + logsetting.init() + unittest2.main() diff --git a/compass/tests/db/api/test_metadata_holder.py b/compass/tests/db/api/test_metadata_holder.py index 1a3bfcfa..ce470ece 100644 --- a/compass/tests/db/api/test_metadata_holder.py +++ b/compass/tests/db/api/test_metadata_holder.py @@ -15,6 +15,7 @@ import datetime import logging +import mock import os import unittest2 @@ -27,31 +28,231 @@ reload(setting) from base import BaseTest +from compass.db.api import adapter as adapter_api from compass.db.api import adapter_holder as adapter from compass.db.api import database +from compass.db.api import metadata as metadata_api from compass.db.api import metadata_holder as metadata from compass.db.api import user as user_api from compass.db import exception from compass.utils import flags from compass.utils import logsetting +from compass.utils import util -class TestGetPackageMetadata(BaseTest): - """test get package metadata.""" +class MetadataTestCase(unittest2.TestCase): + """Metadata base test case.""" + + def setUp(self): + super(MetadataTestCase, self).setUp() + reload(setting) + setting.CONFIG_DIR = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + 'data' + ) + database.init('sqlite://') + database.create_db() + adapter.load_adapters() + + #Get a os_id and adapter_id + self.user_object = ( + user_api.get_user_object( + setting.COMPASS_ADMIN_EMAIL + ) + ) + self.adapter_object = adapter.list_adapters(self.user_object) + test_adapter = None + for adapter_obj in self.adapter_object: + if adapter_obj['name'] == 'openstack_icehouse': + self.adapter_id = adapter_obj['id'] + test_adapter = adapter_obj + break + self.os_id = None + if test_adapter['flavors']: + for supported_os in test_adapter['supported_oses']: + self.os_id = supported_os['os_id'] + break + + def tearDown(self): + super(MetadataTestCase, self).setUp() + reload(setting) + database.drop_db() + + +class TestGetPackageMetadata(MetadataTestCase): def setUp(self): super(TestGetPackageMetadata, self).setUp() + mock_config = mock.Mock() + self.backup_package_configs = util.load_configs + util.load_configs = mock_config + configs = [{ + 'ADAPTER': 'openstack', + 'METADATA': { + 'security': { + '_self': { + 'required_in_whole_config': True + }, + 'service_credentials': { + '_self': { + 'mapping_to': 'service_credentials' + }, + '$service': { + 'username': { + '_self': { + 'is_required': True, + 'field': 'username', + 'mapping_to': 'username' + } + }, + 'password': { + '_self': { + 'is_required': True, + 'field': 'password', + 'mapping_to': 'password' + } + } + } + } + }, + 'test_package_metadata': { + '_self': { + 'dummy': 'fake' + } + } + } + }] + util.load_configs.return_value = configs + with database.session() as session: + metadata_api.add_package_metadata_internal(session) + metadata.load_metadatas() def tearDown(self): + util.load_configs = self.backup_package_configs super(TestGetPackageMetadata, self).tearDown() def test_get_package_metadata(self): - adapter_object = adapter.list_adapters(self.user_object) + """Test get package metadata.""" package_metadata = metadata.get_package_metadata( self.user_object, - adapter_object[0]['id'] + self.adapter_id ) + expected = [] + for k, v in package_metadata['package_config'].iteritems(): + expected.append(k) self.assertIsNotNone(package_metadata) + self.assertIn('test_package_metadata', expected) + + def test_adapter_not_exist(self): + """Test give a non-exited package_id.""" + self.assertRaises( + exception.RecordNotExists, + metadata.get_package_metadata, + self.user_object, + 99 + ) + + +class TestGetOsMetadata(MetadataTestCase): + def setUp(self): + super(TestGetOsMetadata, self).setUp() + mock_config = mock.Mock() + self.backup_os_configs = util.load_configs + util.load_configs = mock_config + configs = [{ + 'OS': 'general', + 'METADATA': { + 'general': { + '_self': { + 'required_in_whole_config': True + }, + 'language': { + '_self': { + 'field': 'general', + 'default_value': 'EN', + 'options': ['EN', 'CN'], + 'mapping_to': 'language' + } + }, + 'timezone': { + '_self': { + 'field': 'general', + 'default_value': 'UTC', + 'options': [ + 'America/New_York', 'America/Chicago', + 'America/Los_Angeles', 'Asia/Shanghai', + 'Asia/Tokyo', 'Europe/Paris', + 'Europe/London', 'Europe/Moscow', + 'Europe/Rome', 'Europe/Madrid', + 'Europe/Berlin', 'UTC' + ], + 'mapping_to': 'timezone' + } + } + }, + 'test_os_metadata': { + '_self': { + 'test': 'dummy' + } + } + } + }] + util.load_configs.return_value = configs + with database.session() as session: + metadata_api.add_os_metadata_internal(session) + metadata.load_metadatas() + + def tearDown(self): + super(TestGetOsMetadata, self).tearDown() + util.load_configs = self.backup_os_configs + + def test_get_os_metadata(self): + """Test get os metadata.""" + os_metadata = metadata.get_os_metadata( + self.user_object, + self.os_id + ) + expected = [] + for k, v in os_metadata['os_config'].iteritems(): + expected.append(k) + self.assertIsNotNone(os_metadata) + self.assertIn('test_os_metadata', expected) + + def test_os_non_exist(self): + """Test give a non-existed os_id.""" + self.assertRaises( + exception.RecordNotExists, + metadata.get_os_metadata, + self.user_object, + 99 + ) + + +class TestGetPackageOsMetadata(MetadataTestCase): + def setUp(self): + super(TestGetPackageOsMetadata, self).setUp() + + def tearDown(self): + super(TestGetPackageOsMetadata, self).tearDown() + + def test_get_package_os_metadata(self): + """Test get package and os metadata.""" + package_os_metadata = metadata.get_package_os_metadata( + self.user_object, + self.adapter_id, + self.os_id + ) + self.assertIsNotNone(package_os_metadata) + + def test_invalid_parameter(self): + """Test give a non-existed os_id.""" + self.assertRaises( + exception.InvalidParameter, + metadata.get_package_os_metadata, + self.user_object, + self.adapter_id, + 99 + ) if __name__ == '__main__':