From dca478bd025ea3af28b3eae9362a0baf868b4576 Mon Sep 17 00:00:00 2001 From: Shweta P Date: Fri, 19 Aug 2011 19:42:04 -0700 Subject: [PATCH] Adding Entension API unt tests --- etc/quantum.conf.mocktest | 24 + quantum/plugins/cisco/README | 7 +- .../cisco/tests/unit/test_cisco_extension.py | 849 ++++++++++++++++++ 3 files changed, 878 insertions(+), 2 deletions(-) create mode 100644 etc/quantum.conf.mocktest create mode 100644 quantum/plugins/cisco/tests/unit/test_cisco_extension.py diff --git a/etc/quantum.conf.mocktest b/etc/quantum.conf.mocktest new file mode 100644 index 000000000..d094b89d0 --- /dev/null +++ b/etc/quantum.conf.mocktest @@ -0,0 +1,24 @@ +[DEFAULT] +# Show more verbose log output (sets INFO log level output) +verbose = True + +# Show debugging output in logs (sets DEBUG log level output) +debug = False + +# Address to bind the API server +bind_host = 0.0.0.0 + +# Port the bind the API server to +bind_port = 9696 + +# Path to the extensions +api_extensions_path = ../extensions + +[pipeline:extensions_app_with_filter] +pipeline = extensions extensions_test_app + +[filter:extensions] +paste.filter_factory = quantum.common.extensions:plugin_aware_extension_middleware_factory + +[app:extensions_test_app] +paste.app_factory = tests.unit.test_mockext:app_factory diff --git a/quantum/plugins/cisco/README b/quantum/plugins/cisco/README index f7674401a..e15c9ac38 100755 --- a/quantum/plugins/cisco/README +++ b/quantum/plugins/cisco/README @@ -206,7 +206,7 @@ result the quantum/plugins/cisco/run_tests.py script. E.g.: python quantum/plugins/cisco/run_tests.py - quantum.plugins.cisco.tests.unit.test_ucs_plugin.py + quantum.plugins.cisco.tests.unit.test_ucs_plugin 3. All unit tests (needs environment setup as indicated in the pre-requisites): @@ -217,7 +217,10 @@ result the quantum/plugins/cisco/run_tests.py script. python quantum/plugins/cisco/run_tests.py quantum.plugins.cisco.tests.unit - +4. Testing the Extension API + The script is placed alongwith the other cisco unit tests. The location may + change later. + Location quantum/plugins/cisco/tests/unit/test_cisco_extension.py Additional installation required on Nova Compute ------------------------------------------------ diff --git a/quantum/plugins/cisco/tests/unit/test_cisco_extension.py b/quantum/plugins/cisco/tests/unit/test_cisco_extension.py new file mode 100644 index 000000000..14f614c6e --- /dev/null +++ b/quantum/plugins/cisco/tests/unit/test_cisco_extension.py @@ -0,0 +1,849 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 +# Copyright 2011 OpenStack LLC. +# 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. +# +# @authors: Shweta Padubidri, Cisco Systems, Inc. +# Peter Strunk , Cisco Systems, Inc. +# Shubhangi Satras , Cisco Systems, Inc. +import json +import os.path +import routes +import unittest +import logging +import webob +from webtest import TestApp +from extensions import credential +from extensions import portprofile +from extensions import novatenant +from extensions import qos +from quantum.plugins.cisco.db import api as db +from quantum.common import wsgi +from quantum.common import config +from quantum.common import extensions +from quantum import api as server +from quantum.plugins.cisco.l2network_plugin import L2Network +from tests.unit.extension_stubs import StubBaseAppController +from quantum.common.extensions import (PluginAwareExtensionManager, + ExtensionMiddleware) +from quantum.manager import QuantumManager +from quantum.plugins.cisco import l2network_plugin + +TEST_CONF_FILE = os.path.join(os.path.dirname(__file__), os.pardir, + os.pardir, + 'etc', 'quantum.conf.mocktest') +EXTENSIONS_PATH = os.path.join(os.path.dirname(__file__), os.pardir, + os.pardir, + "extensions") + +LOG = logging.getLogger('quantum.plugins.cisco.tests.test_cisco_extensions') + + +class ExtensionsTestApp(wsgi.Router): + + def __init__(self, options={}): + mapper = routes.Mapper() + controller = StubBaseAppController() + mapper.resource("dummy_resource", "/dummy_resources", + controller=controller) + super(ExtensionsTestApp, self).__init__(mapper) + + +class PortprofileExtensionTest(unittest.TestCase): + + def setUp(self): + parent_resource = dict(member_name="tenant", + collection_name="extensions/csco/tenants") + member_actions = {'associate_portprofile': "PUT", + 'disassociate_portprofile': "POST"} + controller = portprofile.PortprofilesController( + QuantumManager.get_plugin()) + res_ext = extensions.ResourceExtension('portprofiles', controller, + parent=parent_resource, + member_actions=member_actions) + self.test_app = setup_extensions_test_app( + SimpleExtensionManager(res_ext)) + self.contenttype = 'application/json' + self.profile_path = '/extensions/csco/tenants/tt/portprofiles' + self.portprofile_path = '/extensions/csco/tenants/tt/portprofiles/' + self.test_port_profile = {'portprofile': + {'portprofile_name': 'cisco_test_portprofile', + 'qos_name': 'test-qos1'}} + self.tenant_id = "test_tenant" + self.network_name = "test_network" + options = {} + options['plugin_provider'] = 'quantum.plugins.cisco.l2network_plugin'\ + '.L2Network' + self.api = server.APIRouterV01(options) + self._l2network_plugin = l2network_plugin.L2Network() + + def test_list_portprofile(self): + + LOG.debug("test_list_portprofile - START") + req_body1 = json.dumps(self.test_port_profile) + create_response1 = self.test_app.post( + self.profile_path, req_body1, + content_type=self.contenttype) + req_body2 = json.dumps({'portprofile': + {'portprofile_name': 'cisco_test_portprofile2', + 'qos_name': 'test-qos2'}}) + create_response2 = self.test_app.post( + self.profile_path, req_body2, + content_type=self.contenttype) + + index_response = self.test_app.get(self.profile_path) + self.assertEqual(200, index_response.status_int) + + # Clean Up - Delete the Port Profiles + resp_body1 = wsgi.Serializer().deserialize(create_response1.body, + self.contenttype) + portprofile_path1_temp = self.portprofile_path +\ + resp_body1['portprofiles']['portprofile']['id'] + portprofile_path1 = str(portprofile_path1_temp) + resp_body2 = wsgi.Serializer().deserialize(create_response2.body, + self.contenttype) + portprofile_path2_temp = self.portprofile_path +\ + resp_body2['portprofiles']['portprofile']['id'] + portprofile_path2 = str(portprofile_path2_temp) + self.tear_down_profile(portprofile_path1) + self.tear_down_profile(portprofile_path2) + LOG.debug("test_list_portprofile - END") + + def test_create_portprofile(self): + + LOG.debug("test_create_portprofile - START") + req_body = json.dumps(self.test_port_profile) + index_response = self.test_app.post(self.profile_path, req_body, + content_type=self.contenttype) + self.assertEqual(200, index_response.status_int) + + # Clean Up - Delete the Port Profile + resp_body = wsgi.Serializer().deserialize(index_response.body, + self.contenttype) + portprofile_path_temp = self.portprofile_path +\ + resp_body['portprofiles']['portprofile']['id'] + portprofile_path = str(portprofile_path_temp) + self.tear_down_profile(portprofile_path) + LOG.debug("test_create_portprofile - END") + + def test_create_portprofileBADRequest(self): + + LOG.debug("test_create_portprofileBADRequest - START") + index_response = self.test_app.post(self.profile_path, 'BAD_REQUEST', + content_type=self.contenttype, + status='*') + self.assertEqual(400, index_response.status_int) + LOG.debug("test_create_portprofileBADRequest - END") + + def test_show_portprofile(self): + + LOG.debug("test_show_portprofile - START") + req_body = json.dumps(self.test_port_profile) + index_response = self.test_app.post(self.profile_path, req_body, + content_type=self.contenttype) + resp_body = wsgi.Serializer().deserialize(index_response.body, + self.contenttype) + show_path_temp = self.portprofile_path +\ + resp_body['portprofiles']['portprofile']['id'] + show_port_path = str(show_path_temp) + show_response = self.test_app.get(show_port_path) + self.assertEqual(200, show_response.status_int) + + # Clean Up - Delete the Port Profile + self.tear_down_profile(show_port_path) + LOG.debug("test_show_portprofile - END") + + def test_show_portprofileDNE(self, portprofile_id='100'): + + LOG.debug("test_show_portprofileDNE - START") + show_path_temp = self.portprofile_path + portprofile_id + show_port_path = str(show_path_temp) + show_response = self.test_app.get(show_port_path, status='*') + self.assertEqual(450, show_response.status_int) + LOG.debug("test_show_portprofileDNE - END") + + def test_update_portprofile(self): + + LOG.debug("test_update_portprofile - START") + req_body = json.dumps(self.test_port_profile) + index_response = self.test_app.post( + self.profile_path, req_body, + content_type=self.contenttype) + resp_body = wsgi.Serializer().deserialize(index_response.body, + self.contenttype) + rename_port_profile = {'portprofile': + {'portprofile_name': 'cisco_rename_portprofile', + 'qos_name': 'test-qos1'}} + rename_req_body = json.dumps(rename_port_profile) + rename_path_temp = self.portprofile_path +\ + resp_body['portprofiles']['portprofile']['id'] + rename_path = str(rename_path_temp) + rename_response = self.test_app.put(rename_path, rename_req_body) + self.assertEqual(200, rename_response.status_int) + + # Clean Up - Delete the Port Profile + self.tear_down_profile(rename_path) + LOG.debug("test_update_portprofile - END") + + def test_update_portprofileBADRequest(self): + + LOG.debug("test_update_portprofileBADRequest - START") + req_body = json.dumps(self.test_port_profile) + index_response = self.test_app.post( + self.profile_path, req_body, + content_type=self.contenttype) + resp_body = wsgi.Serializer().deserialize(index_response.body, + self.contenttype) + rename_path_temp = self.portprofile_path +\ + resp_body['portprofiles']['portprofile']['id'] + rename_path = str(rename_path_temp) + rename_response = self.test_app.put(rename_path, 'BAD_REQUEST', + status='*') + self.assertEqual(400, rename_response.status_int) + + # Clean Up - Delete the Port Profile + self.tear_down_profile(rename_path) + LOG.debug("test_update_portprofileBADRequest - END") + + def test_update_portprofileDNE(self, portprofile_id='100'): + + LOG.debug("test_update_portprofileiDNE - START") + rename_port_profile = {'portprofile': + {'portprofile_name': 'cisco_rename_portprofile', + 'qos_name': 'test-qos1'}} + rename_req_body = json.dumps(rename_port_profile) + update_path_temp = self.portprofile_path + portprofile_id + update_path = str(update_path_temp) + update_response = self.test_app.put(update_path, rename_req_body, + status='*') + self.assertEqual(450, update_response.status_int) + LOG.debug("test_update_portprofileDNE - START") + + def test_delete_portprofile(self): + + LOG.debug("test_delete_portprofile - START") + req_body = json.dumps(self.test_port_profile) + index_response = self.test_app.post( + self.profile_path, req_body, + content_type=self.contenttype) + resp_body = wsgi.Serializer().deserialize(index_response.body, + self.contenttype) + delete_path_temp = self.portprofile_path +\ + resp_body['portprofiles']['portprofile']['id'] + delete_path = str(delete_path_temp) + delete_response = self.test_app.delete(delete_path) + + self.assertEqual(202, delete_response.status_int) + LOG.debug("test_delete_portprofile - END") + + def test_delete_portprofileDNE(self, portprofile_id='100'): + + LOG.debug("test_delete_portprofileDNE - START") + delete_path_temp = self.portprofile_path + portprofile_id + delete_path = str(delete_path_temp) + delete_response = self.test_app.delete(delete_path, status='*') + self.assertEqual(450, delete_response.status_int) + LOG.debug("test_delete_portprofileDNE - END") + + def create_request(self, path, body, content_type, method='GET'): + + LOG.debug("test_create_request - START") + req = webob.Request.blank(path) + req.method = method + req.headers = {} + req.headers['Accept'] = content_type + req.body = body + return req + LOG.debug("test_create_request - END") + + def _create_network(self, name=None): + + LOG.debug("Creating network - START") + if name: + net_name = name + else: + net_name = self.network_name + net_path = "/tenants/tt/networks" + net_data = {'network': {'net-name': '%s' % net_name}} + req_body = wsgi.Serializer().serialize(net_data, self.contenttype) + network_req = self.create_request(net_path, req_body, + self.contenttype, 'POST') + network_res = network_req.get_response(self.api) + network_data = wsgi.Serializer().deserialize(network_res.body, + self.contenttype) + return network_data['networks']['network']['id'] + LOG.debug("Creating network - END") + + def _create_port(self, network_id, port_state): + + LOG.debug("Creating port for network %s - START", network_id) + port_path = "/tenants/tt/networks/%s/ports" % network_id + port_req_data = {'port': {'port-state': '%s' % port_state}} + req_body = wsgi.Serializer().serialize(port_req_data, + self.contenttype) + port_req = self.create_request(port_path, req_body, + self.contenttype, 'POST') + port_res = port_req.get_response(self.api) + port_data = wsgi.Serializer().deserialize(port_res.body, + self.contenttype) + return port_data['ports']['port']['id'] + LOG.debug("Creating port for network - END") + + def test_associate_portprofile(self): + + LOG.debug("test_associate_portprofile - START") + net_id = self._create_network() + port_id = self._create_port(net_id, "ACTIVE") + req_body = json.dumps(self.test_port_profile) + index_response = self.test_app.post( + self.profile_path, req_body, + content_type=self.contenttype) + resp_body = wsgi.Serializer().deserialize(index_response.body, + self.contenttype) + test_port_assign_data = {'portprofile': {'network-id': net_id, + 'port-id': port_id}} + req_assign_body = json.dumps(test_port_assign_data) + associate_path_temp = self.portprofile_path +\ + resp_body['portprofiles']['portprofile']['id'] +\ + "/associate_portprofile" + associate_path = str(associate_path_temp) + associate_response = self.test_app.put( + associate_path, req_assign_body, + content_type=self.contenttype) + self.assertEqual(202, associate_response.status_int) + + # Clean Up - Disassociate and Delete the Port Profile + disassociate_path_temp = self.portprofile_path +\ + resp_body['portprofiles']['portprofile']['id'] +\ + "/disassociate_portprofile" + disassociate_path = str(disassociate_path_temp) + delete_path_temp = self.portprofile_path +\ + resp_body['portprofiles']['portprofile']['id'] + delete_path = str(delete_path_temp) + self.tear_down_associate_profile(delete_path, disassociate_path, + req_assign_body) + LOG.debug("test_associate_portprofile - END") + + def test_associate_portprofileDNE(self, portprofile_id='100'): + + LOG.debug("test_associate_portprofileDNE - START") + test_port_assign_data = {'portprofile': {'network-id': '001', + 'port-id': '1'}} + req_assign_body = json.dumps(test_port_assign_data) + associate_path = self.portprofile_path + portprofile_id +\ + "/associate_portprofile" + associate_response = self.test_app.put( + associate_path, req_assign_body, + content_type=self.contenttype, status='*') + self.assertEqual(450, associate_response.status_int) + LOG.debug("test_associate_portprofileDNE - END") + + def test_disassociate_portprofile(self): + + LOG.debug("test_disassociate_portprofile - START") + net_id = self._create_network() + port_id = self._create_port(net_id, "ACTIVE") + + req_body = json.dumps(self.test_port_profile) + index_response = self.test_app.post( + self.profile_path, req_body, + content_type=self.contenttype) + resp_body = wsgi.Serializer().deserialize(index_response.body, + self.contenttype) + + test_port_assign_data = {'portprofile': {'network-id': net_id, + 'port-id': port_id}} + req_assign_body = json.dumps(test_port_assign_data) + associate_path_temp = self.portprofile_path +\ + resp_body['portprofiles']['portprofile']['id'] +\ + "/associate_portprofile" + associate_path = str(associate_path_temp) + self.test_app.put(associate_path, req_assign_body, + content_type=self.contenttype) + disassociate_path_temp = self.portprofile_path +\ + resp_body['portprofiles']['portprofile']['id'] +\ + "/disassociate_portprofile" + + disassociate_path = str(disassociate_path_temp) + disassociate_response = self.test_app.post( + disassociate_path, req_assign_body, + content_type=self.contenttype) + self.assertEqual(202, disassociate_response.status_int) + resp_body = wsgi.Serializer().deserialize(index_response.body, + self.contenttype) + delete_path_temp = self.portprofile_path +\ + resp_body['portprofiles']['portprofile']['id'] + delete_path = str(delete_path_temp) + self.tear_down_profile(delete_path) + LOG.debug("test_disassociate_portprofile - END") + + def tear_down_profile(self, delete_profile_path): + self.test_app.delete(delete_profile_path) + + def tear_down_associate_profile(self, delete_profile_path, + dissociate_profile_path, req_body): + self.test_app.post(dissociate_profile_path, req_body, + content_type=self.contenttype) + self.tear_down_profile(delete_profile_path) + + def tearDown(self): + db.clear_db() + + +class NovatenantExtensionTest(unittest.TestCase): + + def setUp(self): + parent_resource = dict(member_name="tenant", + collection_name="extensions/csco/tenants") + member_actions = {'get_host': "PUT", + 'get_instance_port': "PUT"} + controller = novatenant.NovatenantsController( + QuantumManager.get_plugin()) + res_ext = extensions.ResourceExtension('novatenants', controller, + parent=parent_resource, + member_actions=member_actions) + self.test_app = setup_extensions_test_app( + SimpleExtensionManager(res_ext)) + self.contenttype = 'application/json' + self.novatenants_path = '/extensions/csco/tenants/tt/novatenants/' + self.test_instance_data = {'novatenant': {'instance_id': 1, + 'instance_desc': {'key1': '1', + 'key2': '2'}}} + + def test_get_host(self): + LOG.debug("test_get_host - START") + req_body = json.dumps(self.test_instance_data) + host_path = self.novatenants_path + "001/get_host" + host_response = self.test_app.put( + host_path, req_body, + content_type=self.contenttype) + self.assertEqual(200, host_response.status_int) + LOG.debug("test_get_host - END") + + def test_get_hostBADRequest(self): + LOG.debug("test_get_hostBADRequest - START") + host_path = self.novatenants_path + "001/get_host" + host_response = self.test_app.put( + host_path, 'BAD_REQUEST', + content_type=self.contenttype, status='*') + self.assertEqual(400, host_response.status_int) + LOG.debug("test_get_hostBADRequest - END") + + def test_instance_port(self): + LOG.debug("test_instance_port - START") + req_body = json.dumps(self.test_instance_data) + instance_port_path = self.novatenants_path + "001/get_instance_port" + instance_port_response = self.test_app.put( + instance_port_path, req_body, + content_type=self.contenttype) + self.assertEqual(200, instance_port_response.status_int) + LOG.debug("test_instance_port - END") + + +class QosExtensionTest(unittest.TestCase): + + def setUp(self): + + parent_resource = dict(member_name="tenant", + collection_name="extensions/csco/tenants") + controller = qos.QosController(QuantumManager.get_plugin()) + res_ext = extensions.ResourceExtension('qos', controller, + parent=parent_resource) + + self.test_app = setup_extensions_test_app( + SimpleExtensionManager(res_ext)) + self.contenttype = 'application/json' + self.qos_path = '/extensions/csco/tenants/tt/qos' + self.qos_second_path = '/extensions/csco/tenants/tt/qos/' + self.test_qos_data = {'qos': {'qos_name': 'cisco_test_qos', + 'qos_desc': {'PPS': 50, 'TTL': 5}}} + + def test_create_qos(self): + + LOG.debug("test_create_qos - START") + req_body = json.dumps(self.test_qos_data) + index_response = self.test_app.post(self.qos_path, + req_body, + content_type=self.contenttype) + self.assertEqual(200, index_response.status_int) + + # Clean Up - Delete the qos + resp_body = wsgi.Serializer().deserialize(index_response.body, + self.contenttype) + qos_path_temp = self.qos_second_path +\ + resp_body['qoss']['qos']['id'] + qos_path = str(qos_path_temp) + self.tearDownQos(qos_path) + LOG.debug("test_create_qos - END") + + def test_create_qosBADRequest(self): + + LOG.debug("test_create_qosBADRequest - START") + index_response = self.test_app.post(self.qos_path, + 'BAD_REQUEST', + content_type=self.contenttype, + status='*') + self.assertEqual(400, index_response.status_int) + LOG.debug("test_create_qosBADRequest - END") + + def test_list_qoss(self): + LOG.debug("test_list_qoss - START") + req_body1 = json.dumps(self.test_qos_data) + create_resp1 = self.test_app.post(self.qos_path, req_body1, + content_type=self.contenttype) + req_body2 = json.dumps({'qos': {'qos_name': 'cisco_test_qos2', + 'qos_desc': {'PPS': 50, 'TTL': 5}}}) + create_resp2 = self.test_app.post(self.qos_path, req_body2, + content_type=self.contenttype) + index_response = self.test_app.get(self.qos_path) + self.assertEqual(200, index_response.status_int) + + # Clean Up - Delete the qos's + resp_body1 = wsgi.Serializer().deserialize(create_resp1.body, + self.contenttype) + qos_path1_temp = self.qos_second_path +\ + resp_body1['qoss']['qos']['id'] + qos_path1 = str(qos_path1_temp) + resp_body2 = wsgi.Serializer().deserialize(create_resp2.body, + self.contenttype) + qos_path2_temp = self.qos_second_path +\ + resp_body2['qoss']['qos']['id'] + qos_path2 = str(qos_path2_temp) + self.tearDownQos(qos_path1) + self.tearDownQos(qos_path2) + LOG.debug("test_list_qoss - END") + + def test_show_qos(self): + LOG.debug("test_show_qos - START") + req_body = json.dumps(self.test_qos_data) + index_response = self.test_app.post(self.qos_path, req_body, + content_type=self.contenttype) + resp_body = wsgi.Serializer().deserialize(index_response.body, + self.contenttype) + show_path_temp = self.qos_second_path +\ + resp_body['qoss']['qos']['id'] + show_qos_path = str(show_path_temp) + show_response = self.test_app.get(show_qos_path) + self.assertEqual(200, show_response.status_int) + + # Clean Up - Delete the qos + self.tearDownQos(show_qos_path) + LOG.debug("test_show_qos - END") + + def test_show_qosDNE(self, qos_id='100'): + LOG.debug("test_show_qosDNE - START") + show_path_temp = self.qos_second_path + qos_id + show_qos_path = str(show_path_temp) + show_response = self.test_app.get(show_qos_path, status='*') + self.assertEqual(452, show_response.status_int) + LOG.debug("test_show_qosDNE - END") + + def test_update_qos(self): + + LOG.debug("test_update_qos - START") + req_body = json.dumps(self.test_qos_data) + index_response = self.test_app.post(self.qos_path, req_body, + content_type=self.contenttype) + resp_body = wsgi.Serializer().deserialize(index_response.body, + self.contenttype) + rename_req_body = json.dumps({'qos': {'qos_name': 'cisco_rename_qos', + 'qos_desc': {'PPS': 50, 'TTL': 5}}}) + rename_path_temp = self.qos_second_path +\ + resp_body['qoss']['qos']['id'] + rename_path = str(rename_path_temp) + rename_response = self.test_app.put(rename_path, rename_req_body) + self.assertEqual(200, rename_response.status_int) + self.tearDownQos(rename_path) + LOG.debug("test_update_qos - END") + + def test_update_qosDNE(self, qos_id='100'): + LOG.debug("test_update_qosDNE - START") + rename_req_body = json.dumps({'qos': {'qos_name': 'cisco_rename_qos', + 'qos_desc': {'PPS': 50, 'TTL': 5}}}) + rename_path_temp = self.qos_second_path + qos_id + rename_path = str(rename_path_temp) + rename_response = self.test_app.put(rename_path, rename_req_body, + status='*') + self.assertEqual(452, rename_response.status_int) + LOG.debug("test_update_qosDNE - END") + + def test_update_qosBADRequest(self): + LOG.debug("test_update_qosBADRequest - START") + req_body = json.dumps(self.test_qos_data) + index_response = self.test_app.post(self.qos_path, req_body, + content_type=self.contenttype) + resp_body = wsgi.Serializer().deserialize(index_response.body, + self.contenttype) + rename_path_temp = self.qos_second_path +\ + resp_body['qoss']['qos']['id'] + rename_path = str(rename_path_temp) + rename_response = self.test_app.put(rename_path, 'BAD_REQUEST', + status="*") + self.assertEqual(400, rename_response.status_int) + + # Clean Up - Delete the Port Profile + self.tearDownQos(rename_path) + LOG.debug("test_update_qosBADRequest - END") + + def test_delete_qos(self): + + LOG.debug("test_delete_qos - START") + req_body = json.dumps({'qos': {'qos_name': 'cisco_test_qos', + 'qos_desc': {'PPS': 50, 'TTL': 5}}}) + index_response = self.test_app.post(self.qos_path, req_body, + content_type=self.contenttype) + resp_body = wsgi.Serializer().deserialize(index_response.body, + self.contenttype) + delete_path_temp = self.qos_second_path +\ + resp_body['qoss']['qos']['id'] + delete_path = str(delete_path_temp) + delete_response = self.test_app.delete(delete_path) + self.assertEqual(202, delete_response.status_int) + LOG.debug("test_delete_qos - END") + + def test_delete_qosDNE(self, qos_id='100'): + LOG.debug("test_delete_qosDNE - START") + delete_path_temp = self.qos_second_path + qos_id + delete_path = str(delete_path_temp) + delete_response = self.test_app.delete(delete_path, status='*') + self.assertEqual(452, delete_response.status_int) + LOG.debug("test_delete_qosDNE - END") + + def tearDownQos(self, delete_profile_path): + self.test_app.delete(delete_profile_path) + + +class CredentialExtensionTest(unittest.TestCase): + + def setUp(self): + parent_resource = dict(member_name="tenant", + collection_name="extensions/csco/tenants") + controller = credential.CredentialController( + QuantumManager.get_plugin()) + res_ext = extensions.ResourceExtension('credentials', controller, + parent=parent_resource) + self.test_app = setup_extensions_test_app( + SimpleExtensionManager(res_ext)) + self.contenttype = 'application/json' + self.credential_path = '/extensions/csco/tenants/tt/credentials' + self.cred_second_path = '/extensions/csco/tenants/tt/credentials/' + self.test_credential_data = {'credential': + {'credential_name': 'cred8', + 'user_name': 'newUser2', + 'password': 'newPasswd1' + }} + + def test_list_credentials(self): + #Create Credential before listing + LOG.debug("test_list_credentials - START") + req_body1 = json.dumps(self.test_credential_data) + create_response1 = self.test_app.post( + self.credential_path, req_body1, + content_type=self.contenttype) + req_body2 = json.dumps({'credential': + {'credential_name': 'cred9', + 'user_name': 'newUser2', + 'password': 'newPasswd2' + }}) + create_response2 = self.test_app.post( + self.credential_path, req_body2, + content_type=self.contenttype) + index_response = self.test_app.get( + self.credential_path) + self.assertEqual(200, index_response.status_int) + #CLean Up - Deletion of the Credentials + resp_body1 = wsgi.Serializer().deserialize( + create_response1.body, self.contenttype) + delete_path1_temp = self.cred_second_path +\ + resp_body1['credentials']['credential']['id'] + delete_path1 = str(delete_path1_temp) + resp_body2 = wsgi.Serializer().deserialize( + create_response2.body, self.contenttype) + delete_path2_temp = self.cred_second_path +\ + resp_body2['credentials']['credential']['id'] + delete_path2 = str(delete_path2_temp) + self.tearDownCredential(delete_path1) + self.tearDownCredential(delete_path2) + LOG.debug("test_list_credentials - END") + + def test_create_credential(self): + LOG.debug("test_create_credential - START") + req_body = json.dumps(self.test_credential_data) + index_response = self.test_app.post( + self.credential_path, req_body, + content_type=self.contenttype) + self.assertEqual(200, index_response.status_int) + #CLean Up - Deletion of the Credentials + resp_body = wsgi.Serializer().deserialize( + index_response.body, self.contenttype) + delete_path_temp = self.cred_second_path +\ + resp_body['credentials']['credential']['id'] + delete_path = str(delete_path_temp) + self.tearDownCredential(delete_path) + LOG.debug("test_create_credential - END") + + def test_create_credentialBADRequest(self): + LOG.debug("test_create_credentialBADRequest - START") + index_response = self.test_app.post( + self.credential_path, 'BAD_REQUEST', + content_type=self.contenttype, status='*') + self.assertEqual(400, index_response.status_int) + LOG.debug("test_create_credentialBADRequest - END") + + def test_show_credential(self): + LOG.debug("test_show_credential - START") + req_body = json.dumps(self.test_credential_data) + index_response = self.test_app.post( + self.credential_path, req_body, + content_type=self.contenttype) + resp_body = wsgi.Serializer().deserialize( + index_response.body, self.contenttype) + show_path_temp = self.cred_second_path +\ + resp_body['credentials']['credential']['id'] + show_cred_path = str(show_path_temp) + show_response = self.test_app.get(show_cred_path) + self.assertEqual(200, show_response.status_int) + LOG.debug("test_show_credential - END") + + def test_show_credentialDNE(self, credential_id='100'): + LOG.debug("test_show_credentialDNE - START") + show_path_temp = self.cred_second_path + credential_id + show_cred_path = str(show_path_temp) + show_response = self.test_app.get(show_cred_path, status='*') + self.assertEqual(451, show_response.status_int) + LOG.debug("test_show_credentialDNE - END") + + def test_update_credential(self): + LOG.debug("test_update_credential - START") + req_body = json.dumps(self.test_credential_data) + + index_response = self.test_app.post( + self.credential_path, req_body, + content_type=self.contenttype) + resp_body = wsgi.Serializer().deserialize( + index_response.body, self.contenttype) + rename_req_body = json.dumps({'credential': + {'credential_name': 'cred3', + 'user_name': 'RenamedUser', + 'password': 'Renamedpassword' + }}) + rename_path_temp = self.cred_second_path +\ + resp_body['credentials']['credential']['id'] + rename_path = str(rename_path_temp) + rename_response = self.test_app.put(rename_path, rename_req_body) + self.assertEqual(200, rename_response.status_int) + # Clean Up - Delete the Credentials + self.tearDownCredential(rename_path) + LOG.debug("test_update_credential - END") + + def test_update_credBADReq(self): + LOG.debug("test_update_credBADReq - START") + req_body = json.dumps(self.test_credential_data) + index_response = self.test_app.post( + self.credential_path, req_body, + content_type=self.contenttype) + resp_body = wsgi.Serializer().deserialize( + index_response.body, self.contenttype) + rename_path_temp = self.cred_second_path +\ + resp_body['credentials']['credential']['id'] + rename_path = str(rename_path_temp) + rename_response = self.test_app.put(rename_path, 'BAD_REQUEST', + status='*') + self.assertEqual(400, rename_response.status_int) + LOG.debug("test_update_credBADReq - END") + + def test_update_credentialDNE(self, credential_id='100'): + LOG.debug("test_update_credentialDNE - START") + rename_req_body = json.dumps({'credential': + {'credential_name': 'cred3', + 'user_name': 'RenamedUser', + 'password': 'Renamedpassword' + }}) + rename_path_temp = self.cred_second_path + credential_id + rename_path = str(rename_path_temp) + rename_response = self.test_app.put(rename_path, rename_req_body, + status='*') + self.assertEqual(451, rename_response.status_int) + LOG.debug("test_update_credentialDNE - END") + + def test_delete_credential(self): + LOG.debug("test_delete_credential - START") + req_body = json.dumps(self.test_credential_data) + index_response = self.test_app.post( + self.credential_path, req_body, + content_type=self.contenttype) + resp_body = wsgi.Serializer().deserialize( + index_response.body, self.contenttype) + delete_path_temp = self.cred_second_path +\ + resp_body['credentials']['credential']['id'] + delete_path = str(delete_path_temp) + delete_response = self.test_app.delete(delete_path) + self.assertEqual(202, delete_response.status_int) + LOG.debug("test_delete_credential - END") + + def test_delete_credentialDNE(self, credential_id='100'): + LOG.debug("test_delete_credentialDNE - START") + delete_path_temp = self.cred_second_path + credential_id + delete_path = str(delete_path_temp) + delete_response = self.test_app.delete(delete_path, status='*') + self.assertEqual(451, delete_response.status_int) + LOG.debug("test_delete_credentialDNE - END") + + def tearDownCredential(self, delete_path): + self.test_app.delete(delete_path) + + +def app_factory(global_conf, **local_conf): + conf = global_conf.copy() + conf.update(local_conf) + return ExtensionsTestApp(conf) + + +def setup_extensions_middleware(extension_manager=None): + extension_manager = (extension_manager or + PluginAwareExtensionManager(EXTENSIONS_PATH, + L2Network())) + options = {'config_file': TEST_CONF_FILE} + conf, app = config.load_paste_app('extensions_test_app', options, None) + return ExtensionMiddleware(app, conf, ext_mgr=extension_manager) + + +def setup_extensions_test_app(extension_manager=None): + return TestApp(setup_extensions_middleware(extension_manager)) + + +class SimpleExtensionManager(object): + + def __init__(self, resource_ext=None, action_ext=None, request_ext=None): + self.resource_ext = resource_ext + self.action_ext = action_ext + self.request_ext = request_ext + + def get_resources(self): + resource_exts = [] + if self.resource_ext: + resource_exts.append(self.resource_ext) + return resource_exts + + def get_actions(self): + action_exts = [] + if self.action_ext: + action_exts.append(self.action_ext) + return action_exts + + def get_request_extensions(self): + request_extensions = [] + if self.request_ext: + request_extensions.append(self.request_ext) + return request_extensions -- 2.45.2