diff --git a/run_tests.sh b/run_tests.sh new file mode 100644 index 0000000..a35f95b --- /dev/null +++ b/run_tests.sh @@ -0,0 +1,15 @@ +#!/bin/sh + +rm -rf build/html +mkdir build/html + +pylint -d I0011 -i y -f html api tests >> build/html/lint.html + +nosetests --with-coverage --cover-html --cover-html-dir=build/html/ --cover-erase --cover-package=api + +pep8 --repeat --statistics --count api + +pyflakes api + +# TODO: create project! +#epydoc epydoc.prj diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..6b19b7e --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1,221 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# +# Copyright (c) 2012, Intel Performance Learning Solutions 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. + +try: + from occi import core_model + from occi.extensions import infrastructure +except ImportError: + core_model = None + +from nova import image + + +def missing_pyssf(): + return core_model is None + + +def fake_get_image_service(context, image_href): + ''' + Make sure fake image service is used. + ''' + tmp = image.fake.FakeImageService(), image_href + return tmp + + +def fake_get_default_image_service(): + ''' + Fake get default image_service... + ''' + return image.fake.FakeImageService() + + +def fake_show(meh, context, id): + ''' + Returns a single image... + ''' + return {'id': id, + 'container_format': 'ami', + 'properties': { + 'kernel_id': id, + 'ramdisk_id': id} + } + + +def fake_response(arg0, arg1): + ''' + Fake WSGI response method + ''' + pass + + +def fake_get_floating_ip_pools(meh, context): + return [{'name': 'test1'}, {'name': 'test2'}, ] + + +def fake_get_instance_nw_info(meh, ctx, instance): + return [] + + +def fake_get_resource(meh, key, extras): + name = 'DEFAULT_NETWORK' + + net_attrs = { + 'occi.core.id': name, + 'occi.network.vlan': '', + 'occi.network.label': 'public', + 'occi.network.state': 'up', + 'occi.network.address': '', + 'occi.network.gateway': '', + 'occi.network.allocation': '', + } + + default_network = core_model.Resource(name, infrastructure.NETWORK, \ + [infrastructure.IPNETWORK], [], + 'This is the network all VMs are attached to.', + 'Default Network') + + default_network.attributes = net_attrs + + return default_network + + +def fake_security_group_get_by_project(ctx, proj_id): + return [{'name':'grp1'}, {'name':'grp2'}] + + +def fake_compute_get(meh, ct, uid): + instance = {} + instance['vm_state'] = 'active' + return instance + + +def fake_compute_occi_get(meh, entity, extras): + entity = core_model.Entity("123", 'A test entity', None, []) + entity.attributes['occi.core.id'] = '123-123-123' + entity.links = [] + entity.actions = [infrastructure.START, + infrastructure.STOP, + infrastructure.SUSPEND, \ + infrastructure.RESTART] + return entity + + +def fake_compute_delete(meh, ctx, vol): + pass + + +def fake_compute_unpause(meh, context, instance): + pass + + +def fake_compute_resume(meh, context, instance): + pass + + +def fake_compute_suspend(meh, context, instance): + pass + + +def fake_compute_reboot(meh, context, instance, type): + pass + + +def fake_compute_pause(meh, context, instance): + pass + + +def fake_storage_get(meh, ct, uid): + instance = {} + instance['id'] = '321321' + instance['size'] = '1.0' + instance['status'] = 'available' + return instance + + +def fake_storage_delete(meh, ctx, vol): + pass + + +def fake_compute_set_admin_password(meh, context, instance, new_password): + pass + + +def fake_compute_revert_resize(meh, context, instance): + pass + + +def fake_compute_confirm_resize(meh, context, instance): + pass + + +def fake_compute_snapshot(m, context, instance, image_name, extra_properties): + pass + + +def fake_compute_associate_floating_ip(meh, context, instance, address): + pass + + +def fake_network_allocate_floating_ip(meh, context, pool): + pass + + +def fake_network_disassociate_floating_ip(meh, context, address): + pass + + +def fake_network_release_floating_ip(meh, context, address): + pass + + +def fake_db_security_group_get_by_name(ctx, project_id, name): + security_group = {} + security_group['id'] = '12' + security_group['rules'] = [] + return security_group + + +def fake_db_security_group_in_use(context, security_group_id): + return False + + +def fake_db_security_group_rule_get(context, id): + rule = {} + rule['occi.core.id'] = '123-123-123' + rule['occi.network.security.protocol'] = 'tcp' + rule['occi.network.security.to'] = '22' + rule['occi.network.security.from'] = '22' + rule['occi.network.security.range'] = '0.0.0.0/24' + rule['parent_group_id'] = 12 + rule['id'] = 12 + return rule + + +def fake_db_security_group_get(context, id): + security_group = {} + security_group['id'] = '12' + security_group['rules'] = [] + return security_group + + +def fake_db_security_group_rule_destroy(context, id): + pass + + +def fake_compute_trigger_security_group_rules_refresh(meh, context, + security_group_id): + pass diff --git a/tests/extensions/__init__.py b/tests/extensions/__init__.py new file mode 100644 index 0000000..b19c8f4 --- /dev/null +++ b/tests/extensions/__init__.py @@ -0,0 +1,16 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# +# Copyright (c) 2012, Intel Performance Learning Solutions 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. diff --git a/tests/extensions/test_occi_future.py b/tests/extensions/test_occi_future.py new file mode 100644 index 0000000..47c0e85 --- /dev/null +++ b/tests/extensions/test_occi_future.py @@ -0,0 +1,136 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# +# Copyright (c) 2012, Intel Performance Learning Solutions 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 gettext + +try: + from occi import core_model +except ImportError: + pass + +from nova import context +from nova import compute +from nova import db +from nova import flags +from nova import test +try: + from api.extensions import occi_future + from api import registry +except ImportError: + pass +from tests import occi + + +FLAGS = flags.FLAGS + + +class TestOcciSecurityGroupBackend(test.TestCase): + + def setUp(self): + super(TestOcciSecurityGroupBackend, self).setUp() + + gettext.install('nova-api-occi') + + # create sec context + self.user_id = 'fake' + self.project_id = 'fake' + self.context = context.RequestContext(self.user_id, + self.project_id, + is_admin=True) + + # OCCI related setup + if not occi.missing_pyssf(): + self.category = occi_future.UserSecurityGroupMixin( + term='my_grp', + scheme='http://www.mystuff.com/mygroup#', + related=[occi_future.SEC_GROUP], + attributes=None, + location='/security/my_grp/') + + self.extras = {'nova_ctx': self.context, + 'registry': registry.OCCIRegistry()} + + self.class_under_test = occi_future.SecurityGroupBackend() + + @test.skip_if(occi.missing_pyssf(), "Test requires pyssf") + def test_create_for_success(self): + self.class_under_test.init_sec_group(self.category, self.extras) + + @test.skip_if(occi.missing_pyssf(), "Test requires pyssf") + def test_delete_for_success(self): + self.stubs.Set(db, 'security_group_get_by_name', + occi.fake_db_security_group_get_by_name) + self.stubs.Set(db, 'security_group_in_use', + occi.fake_db_security_group_in_use) + self.class_under_test.destroy(self.category, self.extras) + + self.stubs.UnsetAll() + + +class TestOcciSecurityRuleBackend(test.TestCase): + + def setUp(self): + super(TestOcciSecurityRuleBackend, self).setUp() + + # create sec context + self.user_id = 'fake' + self.project_id = 'fake' + self.context = context.RequestContext(self.user_id, + self.project_id, + is_admin=True) + + # OCCI related setup + if not occi.missing_pyssf(): + self.category = occi_future.UserSecurityGroupMixin( + term='my_grp', + scheme='http://www.mystuff.com/mygroup#', + related=[occi_future.SEC_GROUP], + attributes=None, + location='/security/my_grp/') + + self.entity = core_model.Entity("123", 'A test entity', None, + [self.category]) + self.entity.attributes['occi.core.id'] = '123123123' + self.entity.attributes['occi.network.security.protocol'] = 'tcp' + self.entity.attributes['occi.network.security.to'] = '22' + self.entity.attributes['occi.network.security.from'] = '22' + self.entity.attributes['occi.network.security.range'] = \ + '0.0.0.0/24' + self.entity.links = [] + self.extras = {'nova_ctx': self.context, + 'registry': registry.OCCIRegistry()} + + self.class_under_test = occi_future.SecurityRuleBackend() + + @test.skip_if(occi.missing_pyssf(), "Test requires pyssf") + def test_create_for_success(self): + self.stubs.Set(db, 'security_group_get_by_name', + occi.fake_db_security_group_get_by_name) + self.class_under_test.create(self.entity, self.extras) + + @test.skip_if(occi.missing_pyssf(), "Test requires pyssf") + def test_delete_for_success(self): + self.stubs.Set(db, 'security_group_rule_get', + occi.fake_db_security_group_rule_get) + self.stubs.Set(db, 'security_group_get', + occi.fake_db_security_group_get) + self.stubs.Set(db, 'security_group_rule_destroy', + occi.fake_db_security_group_rule_destroy) + self.stubs.Set(compute.API, 'trigger_security_group_rules_refresh', + occi.fake_compute_trigger_security_group_rules_refresh) + + self.class_under_test.delete(self.entity, self.extras) diff --git a/tests/extensions/test_openstack.py b/tests/extensions/test_openstack.py new file mode 100644 index 0000000..851837f --- /dev/null +++ b/tests/extensions/test_openstack.py @@ -0,0 +1,133 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# +# Copyright (c) 2012, Intel Performance Learning Solutions 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 gettext + +try: + from occi import core_model +except ImportError: + pass + +from nova import context +from nova import compute +from nova import flags +from nova import test +try: + from api import registry + from api.compute import templates + from api.extensions import openstack +except ImportError: + pass +from nova.network import api as net_api +import tests as occi + + +FLAGS = flags.FLAGS + + +class TestOcciOpenStackActionBackend(test.TestCase): + + def setUp(self): + super(TestOcciOpenStackActionBackend, self).setUp() + + gettext.install('nova-api-occi') + + # create sec context + self.user_id = 'fake' + self.project_id = 'fake' + self.context = context.RequestContext(self.user_id, + self.project_id, + is_admin=True) + + self.stubs.Set(compute.API, 'get', occi.fake_compute_get) + self.stubs.Set(compute.API, 'delete', occi.fake_storage_delete) + + # OCCI related setup + if not occi.missing_pyssf(): + self.stubs.Set(registry.OCCIRegistry, 'get_resource', + occi.fake_get_resource) + self.os_template = templates.OsTemplate( + 'http://schemas.openstack.org/template/os#', + 'foo', '1') + self.resource_template = templates.ResourceTemplate( + 'http://schemas.openstack.org/template/resource#', + 'm1.small') + + self.entity = core_model.Entity("123", 'A test entity', None, + [self.os_template, self.resource_template]) + self.entity.attributes['occi.core.id'] = '123-123-123' + self.entity.links = [] + self.extras = {'nova_ctx': self.context, + 'registry': registry.OCCIRegistry()} + + self.class_under_test = openstack.OsComputeActionBackend() + + @test.skip_if(occi.missing_pyssf(), "Test requires pyssf") + def test_action_for_success(self): + self.stubs.Set(compute.API, 'set_admin_password', + occi.fake_compute_set_admin_password) + self.stubs.Set(compute.API, 'revert_resize', + occi.fake_compute_revert_resize) + self.stubs.Set(compute.API, 'confirm_resize', + occi.fake_compute_confirm_resize) + self.stubs.Set(compute.API, 'snapshot', + occi.fake_compute_snapshot) + self.stubs.Set(net_api.API, 'allocate_floating_ip', + occi.fake_network_allocate_floating_ip) + self.stubs.Set(compute.API, 'associate_floating_ip', + occi.fake_compute_associate_floating_ip) + self.stubs.Set(net_api.API, 'disassociate_floating_ip', + occi.fake_network_disassociate_floating_ip) + self.stubs.Set(net_api.API, 'release_floating_ip', + occi.fake_network_release_floating_ip) + + self.entity.attributes['occi.compute.state'] = 'active' + self.entity.actions = [openstack.OS_CHG_PWD] + self.class_under_test.action(self.entity, openstack.OS_CHG_PWD, + {'org.openstack.credentials.admin_pwd': 'passwd_'}, + self.extras) + + self.entity.attributes['occi.compute.state'] = 'active' + self.entity.actions = [openstack.OS_REVERT_RESIZE] + self.class_under_test.action(self.entity, openstack.OS_REVERT_RESIZE, + {}, self.extras) + + self.entity.attributes['occi.compute.state'] = 'active' + self.entity.actions = [openstack.OS_CONFIRM_RESIZE] + self.class_under_test.action(self.entity, openstack.OS_CONFIRM_RESIZE, + {}, self.extras) + + self.entity.attributes['occi.compute.state'] = 'active' + self.entity.actions = [openstack.OS_CREATE_IMAGE] + self.class_under_test.action(self.entity, openstack.OS_CREATE_IMAGE, + {'org.openstack.snapshot.image_name': 'img'}, + self.extras) + + self.entity.attributes['occi.compute.state'] = 'active' + self.entity.actions = [openstack.OS_ALLOC_FLOATING_IP] + self.class_under_test.action(self.entity, + openstack.OS_ALLOC_FLOATING_IP, + {'org.openstack.network.floating.pool': 'test'}, + self.extras) + + self.entity.attributes['occi.compute.state'] = 'active' + self.entity.actions = [openstack.OS_DEALLOC_FLOATING_IP] + self.class_under_test.action(self.entity, + openstack.OS_DEALLOC_FLOATING_IP, {}, + self.extras) + + self.stubs.UnsetAll() diff --git a/tests/test_computeresource.py b/tests/test_computeresource.py new file mode 100644 index 0000000..a9aeac3 --- /dev/null +++ b/tests/test_computeresource.py @@ -0,0 +1,175 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# +# Copyright (c) 2012, Intel Performance Learning Solutions 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 gettext + +try: + from occi import core_model + from occi.extensions import infrastructure +except ImportError: + pass +from webob import exc + +from nova import compute +from nova import context +from nova import flags +from nova import image +from nova import rpc +from nova import test +try: + from api.compute import computeresource + from api.compute import templates + from api import registry +except ImportError: + pass +from nova.image import fake +from nova.network import api as net_api +from nova.scheduler import driver as scheduler_driver +import tests as occi + +FLAGS = flags.FLAGS + + +def fake_rpc_cast(ctx, topic, msg, do_cast=True): + """ + The RPC cast wrapper so scheduler returns instances... + """ + if topic == FLAGS.scheduler_topic and \ + msg['method'] == 'run_instance': + request_spec = msg['args']['request_spec'] + scheduler = scheduler_driver.Scheduler + num_instances = request_spec.get('num_instances', 1) + instances = [] + for x in xrange(num_instances): + instance = scheduler().create_instance_db_entry( + ctx, request_spec) + encoded = scheduler_driver.encode_instance(instance) + instances.append(encoded) + return instances + else: + pass + + +class TestOcciComputeResource(test.TestCase): + + def setUp(self): + super(TestOcciComputeResource, self).setUp() + + gettext.install('nova-api-occi') + + # create sec context + self.user_id = 'fake' + self.project_id = 'fake' + self.context = context.RequestContext(self.user_id, + self.project_id, + is_admin=True) + + # setup image service... + self.stubs.Set(image, 'get_image_service', occi.fake_get_image_service) + self.stubs.Set(fake._FakeImageService, 'show', occi.fake_show) + self.stubs.Set(rpc, 'cast', fake_rpc_cast) + self.stubs.Set(net_api.API, 'get_instance_nw_info', + occi.fake_get_instance_nw_info) + self.stubs.Set(compute.API, 'get', occi.fake_compute_get) + self.stubs.Set(compute.API, 'delete', occi.fake_storage_delete) + + # OCCI related setup + if not occi.missing_pyssf(): + self.stubs.Set(registry.OCCIRegistry, 'get_resource', + occi.fake_get_resource) + self.os_template = templates.OsTemplate( + 'http://schemas.openstack.org/template/os#', + 'foo', '1') + self.resource_template = templates.ResourceTemplate( + 'http://schemas.openstack.org/template/resource#', + 'm1.small') + + self.entity = core_model.Entity("123", 'A test entity', None, + [self.os_template, self.resource_template]) + self.entity.attributes['occi.core.id'] = '123-123-123' + self.entity.attributes['occi.compute.state'] = 'inactive' + self.entity.links = [] + self.extras = {'nova_ctx': self.context, + 'registry': registry.OCCIRegistry(), + } + + self.class_under_test = computeresource.ComputeBackend() + + @test.skip_if(occi.missing_pyssf(), "Test requires pyssf") + def test_create_for_success(self): + """ + Try to create an OCCI entity. + """ + self.assertTrue((self.entity.attributes['occi.compute.state'] == + 'inactive')) + self.class_under_test.create(self.entity, self.extras) + self.assertTrue((self.entity.attributes['occi.compute.state'] == + 'active')) + + @test.skip_if(occi.missing_pyssf(), "Test requires pyssf") + def test_retrieve_for_success(self): + self.assertTrue(len(self.entity.actions) == 0) + self.class_under_test.retrieve(self.entity, self.extras) + self.assertTrue(len(self.entity.actions) > 0) + + @test.skip_if(occi.missing_pyssf(), "Test requires pyssf") + def test_delete_for_success(self): + try: + self.class_under_test.delete(self.entity, self.extras) + except exc.HTTPNotFound: + self.fail('Could not find entity to delete') + + @test.skip_if(occi.missing_pyssf(), "Test requires pyssf") + def test_action_for_success(self): + self.stubs.Set(computeresource.ComputeBackend, 'retrieve', + occi.fake_compute_occi_get) + self.stubs.Set(compute.API, 'unpause', occi.fake_compute_unpause) + self.stubs.Set(compute.API, 'resume', occi.fake_compute_resume) + self.stubs.Set(compute.API, 'suspend', occi.fake_compute_suspend) + self.stubs.Set(compute.API, 'reboot', occi.fake_compute_reboot) + self.stubs.Set(compute.API, 'pause', occi.fake_compute_pause) + + self.entity.attributes['occi.compute.state'] = 'inactive' + self.entity.actions = [infrastructure.START] + self.class_under_test.action(self.entity, infrastructure.START, {}, + self.extras) + self.assertTrue((self.entity.attributes['occi.compute.state'] == + 'active')) + + self.entity.attributes['occi.compute.state'] = 'active' + self.entity.actions = [infrastructure.STOP] + self.class_under_test.action(self.entity, infrastructure.STOP, {}, + self.extras) + self.assertTrue((self.entity.attributes['occi.compute.state'] == + 'inactive')) + + self.entity.attributes['occi.compute.state'] = 'active' + self.entity.actions = [infrastructure.RESTART] + self.class_under_test.action(self.entity, infrastructure.RESTART, + {'method': 'graceful'}, + self.extras) + self.assertTrue((self.entity.attributes['occi.compute.state'] == + 'inactive')) + + self.entity.attributes['occi.compute.state'] = 'active' + self.entity.actions = [infrastructure.SUSPEND] + self.class_under_test.action(self.entity, infrastructure.SUSPEND, {}, + self.extras) + self.assertTrue((self.entity.attributes['occi.compute.state'] == + 'suspended')) + + self.stubs.UnsetAll() diff --git a/tests/test_storageresource.py b/tests/test_storageresource.py new file mode 100644 index 0000000..faf2b94 --- /dev/null +++ b/tests/test_storageresource.py @@ -0,0 +1,89 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# +# Copyright (c) 2012, Intel Performance Learning Solutions 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 gettext + +try: + from occi import core_model +except ImportError: + pass +from webob import exc + +from nova import context +from nova import test +try: + from api import registry + from api.storage import storageresource +except ImportError: + pass +import tests as occi + + +class TestOcciStorageResource(test.TestCase): + + def setUp(self): + super(TestOcciStorageResource, self).setUp() + + gettext.install('nova-api-occi') + + # create sec context + self.user_id = 'fake' + self.project_id = 'fake' + self.context = context.RequestContext(self.user_id, + self.project_id, + is_admin=True) + if not occi.missing_pyssf(): + self.stubs.Set(registry.OCCIRegistry, 'get_resource', + occi.fake_get_resource) + from nova import volume + self.stubs.Set(volume.API, 'get', occi.fake_storage_get) + self.stubs.Set(volume.API, 'delete', occi.fake_storage_delete) + + # OCCI related setup + self.entity = core_model.Entity("123", 'A test entity', None, []) + self.entity.attributes['occi.storage.size'] = '1.0' + self.entity.attributes['occi.storage.state'] = 'offline' + self.entity.attributes['occi.core.id'] = '321' + self.reg = registry.OCCIRegistry() + self.extras = {'nova_ctx': self.context, + 'registry': self.reg} + + self.class_under_test = storageresource.StorageBackend() + + @test.skip_if(occi.missing_pyssf(), "Test requires pyssf") + def test_create_for_success(self): + ''' + Try to create an OCCI entity. + ''' + self.assertTrue((self.entity.attributes['occi.storage.state'] == + 'offline')) + self.class_under_test.create(self.entity, self.extras) + self.assertTrue((self.entity.attributes['occi.storage.state'] == + 'online')) + + @test.skip_if(occi.missing_pyssf(), "Test requires pyssf") + def test_retrieve_for_success(self): + self.assertTrue(len(self.entity.actions) == 0) + self.class_under_test.retrieve(self.entity, self.extras) + self.assertTrue(len(self.entity.actions) > 0) + + @test.skip_if(occi.missing_pyssf(), "Test requires pyssf") + def test_delete_for_success(self): + try: + self.class_under_test.delete(self.entity, self.extras) + except exc.HTTPNotFound: + self.fail('Could not find entity to delete') diff --git a/tests/test_wsgi.py b/tests/test_wsgi.py new file mode 100644 index 0000000..b76e790 --- /dev/null +++ b/tests/test_wsgi.py @@ -0,0 +1,174 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# +# Copyright (c) 2012, Intel Performance Learning Solutions 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 gettext + +try: + from occi.extensions import infrastructure +except ImportError: + pass + +from nova import context +from nova import db +from nova import image +from nova import test +from nova import wsgi +try: + from api.extensions import occi_future + from api.extensions import openstack + from api import wsgi as occi_wsgi +except ImportError: + pass +from nova.network import api as net_api +import tests as occi + + +class TestOcciWsgiApp(test.TestCase): + + def setUp(self): + super(TestOcciWsgiApp, self).setUp() + + gettext.install('nova-api-occi') + + # setup img service... + self.stubs.Set(image, 'get_default_image_service', + occi.fake_get_default_image_service) + self.stubs.Set(net_api.API, 'get_floating_ip_pools', + occi.fake_get_floating_ip_pools) + self.stubs.Set(db, 'security_group_get_by_project', + occi.fake_security_group_get_by_project) + + self.user_id = 'fake' + self.project_id = 'fake' + self.context = context.RequestContext(self.user_id, + self.project_id, + is_admin=True) + + @test.skip_if(occi.missing_pyssf(), "Test requires pyssf") + def test_occi_app_for_success(self): + ''' + test constructor... + ''' + occi_wsgi.OCCIApplication() + + @test.skip_if(occi.missing_pyssf(), "Test requires pyssf") + def test_occi_app_for_sanity(self): + ''' + test for sanity... + ''' + app = occi_wsgi.OCCIApplication() + occi_wsgi.FLAGS.set_default('show_default_net_config', True, None) + + self.assertTrue(isinstance(app, wsgi.Application), + 'OCCI WSGI app needs to be derived frrom wsgi.App') + + # check if all core occi infrastructure kinds and mixins are present + cats = app.registry.get_categories(None) + + self.assertTrue(infrastructure.COMPUTE in cats) + self.assertTrue(infrastructure.STORAGE in cats) + self.assertTrue(infrastructure.STORAGELINK in cats) + self.assertTrue(infrastructure.NETWORK in cats) + self.assertTrue(infrastructure.NETWORKINTERFACE in cats) + self.assertTrue(infrastructure.IPNETWORK in cats) + self.assertTrue(infrastructure.IPNETWORKINTERFACE in cats) + self.assertTrue(infrastructure.OS_TEMPLATE in cats) + self.assertTrue(infrastructure.RESOURCE_TEMPLATE in cats) + + # check if all core occi actions are present + self.assertTrue(infrastructure.START in cats) + self.assertTrue(infrastructure.STOP in cats) + self.assertTrue(infrastructure.RESTART in cats) + self.assertTrue(infrastructure.SUSPEND in cats) + + self.assertTrue(infrastructure.UP in cats) + self.assertTrue(infrastructure.DOWN in cats) + + self.assertTrue(infrastructure.ONLINE in cats) + self.assertTrue(infrastructure.OFFLINE in cats) + self.assertTrue(infrastructure.BACKUP in cats) + self.assertTrue(infrastructure.RESIZE in cats) + + # check if all necessary OpenStack Extensions are present + self.assertTrue(openstack.OS_KEY_PAIR_EXT in cats) + self.assertTrue(openstack.OS_ADMIN_PWD_EXT in cats) + self.assertTrue(openstack.OS_ACCESS_IP_EXT in cats) + + # check if all necessary OpenStack action Extensions are present + self.assertTrue(openstack.OS_CHG_PWD in cats) + self.assertTrue(openstack.OS_REVERT_RESIZE in cats) + self.assertTrue(openstack.OS_CONFIRM_RESIZE in cats) + self.assertTrue(openstack.OS_CREATE_IMAGE in cats) + self.assertTrue(openstack.OS_ALLOC_FLOATING_IP in cats) + self.assertTrue(openstack.OS_DEALLOC_FLOATING_IP in cats) + + # check if all necessary occi_future extensions are present + self.assertTrue(occi_future.CONSOLE_LINK in cats) + self.assertTrue(occi_future.SSH_CONSOLE in cats) + self.assertTrue(occi_future.VNC_CONSOLE in cats) + self.assertTrue(occi_future.SEC_RULE in cats) + self.assertTrue(occi_future.SEC_GROUP in cats) + + # make a call so OS templates get filled + environ = {'SERVER_NAME': 'localhost', + 'SERVER_PORT': '8080', + 'PATH_INFO': '/', + 'REQUEST_METHOD': 'GET', + 'nova.context': self.context} + + app(environ, occi.fake_response) + + # now test for os, resource templates and sec groups + i = 0 + types = ['m1.xlarge', 'm1.medium', 'm1.tiny', 'm1.small', 'm1.large'] + for cat in app.registry.get_categories(None): + if hasattr(cat, 'related') and \ + infrastructure.RESOURCE_TEMPLATE in cat.related: + self.assertTrue(cat.term in types) + scheme = 'http://schemas.openstack.org/template/resource#' + self.assertEquals(scheme, cat.scheme) + i += 1 + self.assertTrue(i, len(types)) + + i = 0 + images = ['fakeimage7', 'fakeimage6', 'fakeimage123456'] + for cat in app.registry.get_categories(None): + if hasattr(cat, 'related') and \ + infrastructure.OS_TEMPLATE in cat.related: + self.assertTrue(cat.term in images) + scheme = 'http://schemas.openstack.org/template/os#' + self.assertEquals(scheme, cat.scheme) + i += 1 + self.assertTrue(i, len(images)) + + i = 0 + pools = ['test1', 'test2'] + scheme = 'http://schemas.openstack.org/instance/network/pool/floating#' + for cat in app.registry.get_categories(None): + if cat.scheme == scheme: + self.assertTrue(cat.term in pools) + i += 1 + self.assertTrue(i, len(pools)) + + i = 0 + grps = ['grp1', 'grp2'] + scheme = 'http://schemas.openstack.org/infrastructure/security/group#' + for cat in app.registry.get_categories(None): + if cat.scheme == scheme: + self.assertTrue(cat.term in grps) + i += 1 + self.assertTrue(i, len(grps))