diff --git a/tests/unit/api.py b/tests/unit/api.py new file mode 100644 index 00000000000..13962322c13 --- /dev/null +++ b/tests/unit/api.py @@ -0,0 +1,649 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2011 Cisco Systems +# 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. +# @author: Tyler Smith, Cisco Systems + +import logging +import unittest +import re + +from quantum.common.wsgi import Serializer +from quantum.client import Client + +LOG = logging.getLogger('quantum.tests.test_api') + +# Set a couple tenants to use for testing +TENANT_1 = 'totore' +TENANT_2 = 'totore2' + +class ServerStub(): + """This class stubs a basic server for the API client to talk to""" + + class Response(object): + """This class stubs a basic response to send the API client""" + def __init__(self, content = None, status = None): + self.content = content + self.status = status + + def read(self): + return self.content + + def status(self): + return status + + # To test error codes, set the host to 10.0.0.1, and the port to the code + def __init__(self, host, port, key_file, cert_file): + self.host = host + self.port = port + self.key_file = key_file + self.cert_file = cert_file + + def request(self, method, action, body, headers): + self.method = method + self.action = action + self.body = body + + def status(self, status=None): + return status or 200 + + def getresponse(self): + res = self.Response(status = self.status()) + + # If the host is 10.0.0.1, return the port as an error code + if self.host == "10.0.0.1": + res.status = self.port + return res + + # Extract important information from the action string to assure sanity + match = re.search('tenants/(.+?)/(.+)\.(json|xml)$', self.action) + + (tenant,path,format) = (match.group(1),match.group(2),match.group(3)) + + data = {'data': {'method':self.method, 'action':self.action, + 'body':self.body,'tenant':tenant, 'path':path, + 'format':format, 'key_file':self.key_file, + 'cert_file':self.cert_file}} + + # Serialize it to the proper format so the API client can deserialize it + if data['data']['format'] == 'json': + res.content = Serializer().serialize(data, "application/json") + else: + res.content = Serializer().serialize(data, "application/xml") + return res + +class APITest(unittest.TestCase): + + def setUp(self): + """ Setups a test environment for the API client """ + HOST = '127.0.0.1' + PORT = 9696 + USE_SSL = True + + self.client = Client(HOST,PORT,USE_SSL,TENANT_1,'json',ServerStub, key_file="/ssl.cert", cert_file="/ssl.cert") + + def _assert_sanity(self, call, status, method, path, data=[], params={}): + """ Perform common assertions to test the sanity of client requests """ + + # Handle an error case first + if status != 200: + (self.client.host,self.client.port) = ("10.0.0.1", status) + self.assertRaises(Exception, call, *data, **params) + return + + # Make the call, then get the data from the root node and strip its data + data = call(*data, **params)['data'] + + self.assertEqual(data['method'], method) + self.assertEqual(data['format'], params['format']) + self.assertEqual(data['tenant'], params['tenant']) + self.assertEqual(data['path'], path) + + return data + + + def _test_list_networks(self, tenant=TENANT_1, format='json', status=200): + LOG.debug("_test_list_networks - tenant:%s "\ + "- format:%s - START", format, tenant) + + self._assert_sanity(self.client.list_networks, + status, + "GET", + "networks", + data = [], + params = {'tenant':tenant, 'format':format},) + + LOG.debug("_test_list_networks - tenant:%s "\ + "- format:%s - END", format, tenant) + + def _test_list_network_details(self, + tenant=TENANT_1,format='json',status=200): + LOG.debug("_test_list_network_details - tenant:%s "\ + "- format:%s - START", format, tenant) + + self._assert_sanity(self.client.list_network_details, + status, + "GET", + "networks/001", + data = ["001"], + params = {'tenant':tenant, 'format':format}) + + LOG.debug("_test_list_network_details - tenant:%s "\ + "- format:%s - END", format, tenant) + + def _test_create_network(self, tenant=TENANT_1, format='json', status=200): + LOG.debug("_test_create_network - tenant:%s "\ + "- format:%s - START", format, tenant) + + self._assert_sanity(self.client.create_network, + status, + "POST", + "networks", + data = [{'network': {'net-name': 'testNetwork'}}], + params = {'tenant':tenant, 'format':format}) + + LOG.debug("_test_create_network - tenant:%s "\ + "- format:%s - END", format, tenant) + + def _test_update_network(self, tenant=TENANT_1, format='json', status=200): + LOG.debug("_test_update_network - tenant:%s "\ + "- format:%s - START", format, tenant) + + self._assert_sanity(self.client.update_network, + status, + "PUT", + "networks/001", + data = [ + "001", + {'network': {'net-name': 'newName'}} + ], + params = {'tenant':tenant, 'format':format}) + + LOG.debug("_test_update_network - tenant:%s "\ + "- format:%s - END", format, tenant) + + def _test_delete_network(self, tenant=TENANT_1, format='json', status=200): + LOG.debug("_test_delete_network - tenant:%s "\ + "- format:%s - START", format, tenant) + + self._assert_sanity(self.client.delete_network, + status, + "DELETE", + "networks/001", + data = ["001"], + params = {'tenant':tenant, 'format':format}) + + LOG.debug("_test_delete_network - tenant:%s "\ + "- format:%s - END", format, tenant) + + def _test_list_ports(self, tenant = TENANT_1, format = 'json', status=200): + LOG.debug("_test_list_ports - tenant:%s "\ + "- format:%s - START", format, tenant) + + self._assert_sanity(self.client.list_ports, + status, + "GET", + "networks/001/ports", + data = ["001"], + params = {'tenant':tenant, 'format':format}) + + LOG.debug("_test_list_ports - tenant:%s "\ + "- format:%s - END", format, tenant) + + def _test_list_port_details(self, + tenant=TENANT_1, format='json', status=200): + LOG.debug("_test_list_port_details - tenant:%s "\ + "- format:%s - START", format, tenant) + + self._assert_sanity(self.client.list_port_details, + status, + "GET", + "networks/001/ports/001", + data = ["001","001"], + params = {'tenant':tenant, 'format':format}) + + LOG.debug("_test_list_port_details - tenant:%s "\ + "- format:%s - END", format, tenant) + + def _test_create_port(self, tenant = TENANT_1, format = 'json', status=200): + LOG.debug("_test_create_port - tenant:%s "\ + "- format:%s - START", format, tenant) + + self._assert_sanity(self.client.create_port, + status, + "POST", + "networks/001/ports", + data = ["001"], + params = {'tenant':tenant, 'format':format}) + + + LOG.debug("_test_create_port - tenant:%s "\ + "- format:%s - END", format, tenant) + + def _test_delete_port(self, tenant = TENANT_1, format = 'json', status=200): + LOG.debug("_test_delete_port - tenant:%s "\ + "- format:%s - START", format, tenant) + + self._assert_sanity(self.client.delete_port, + status, + "DELETE", + "networks/001/ports/001", + data = ["001","001"], + params = {'tenant':tenant, 'format':format}) + + LOG.debug("_test_delete_port - tenant:%s "\ + "- format:%s - END", format, tenant) + + + def _test_set_port_state(self, tenant=TENANT_1, format='json', status=200): + LOG.debug("_test_set_port_state - tenant:%s "\ + "- format:%s - START", format, tenant) + + self._assert_sanity(self.client.set_port_state, + status, + "PUT", + "networks/001/ports/001", + data = ["001","001",{'port':{'state':'ACTIVE'}}], + params = {'tenant':tenant, 'format':format}) + + LOG.debug("_test_set_port_state - tenant:%s "\ + "- format:%s - END", format, tenant) + + def _test_list_port_attachments(self, + tenant=TENANT_1, format='json', status=200): + LOG.debug("_test_list_port_attachments - tenant:%s "\ + "- format:%s - START", format, tenant) + + self._assert_sanity(self.client.list_port_attachments, + status, + "GET", + "networks/001/ports/001/attachment", + data = ["001","001"], + params = {'tenant':tenant, 'format':format}) + + LOG.debug("_test_list_port_attachments - tenant:%s "\ + "- format:%s - END", format, tenant) + + def _test_attach_resource(self, tenant=TENANT_1, format='json', status=200): + LOG.debug("_test_attach_resource - tenant:%s "\ + "- format:%s - START", format, tenant) + + self._assert_sanity(self.client.attach_resource, + status, + "PUT", + "networks/001/ports/001/attachment", + data = ["001","001",{'resource':{'id':'1234'}}], + params = {'tenant':tenant, 'format':format}) + + LOG.debug("_test_attach_resource - tenant:%s "\ + "- format:%s - END", format, tenant) + + def _test_detach_resource(self, tenant=TENANT_1, format='json', status=200): + LOG.debug("_test_detach_resource - tenant:%s "\ + "- format:%s - START", format, tenant) + + self._assert_sanity(self.client.detach_resource, + status, + "DELETE", + "networks/001/ports/001/attachment", + data = ["001","001"], + params = {'tenant':tenant, 'format':format}) + + LOG.debug("_test_detach_resource - tenant:%s "\ + "- format:%s - END", format, tenant) + + def _test_ssl_certificates(self, tenant=TENANT_1, format='json', status=200): + LOG.debug("_test_ssl_certificates - tenant:%s "\ + "- format:%s - START", format, tenant) + + # Set SSL, and our cert file + self.client.use_ssl = True + cert_file = "/fake.cert" + self.client.key_file = self.client.cert_file = cert_file + + data = self._assert_sanity(self.client.list_networks, + status, + "GET", + "networks", + data = [], + params = {'tenant':tenant, 'format':format}) + + self.assertEquals(data["key_file"], cert_file) + self.assertEquals(data["cert_file"], cert_file) + + LOG.debug("_test_ssl_certificates - tenant:%s "\ + "- format:%s - END", format, tenant) + + + def test_list_networks_json(self): + self._test_list_networks(format='json') + + def test_list_networks_xml(self): + self._test_list_networks(format='xml') + + def test_list_networks_alt_tenant(self): + self._test_list_networks(tenant=TENANT_2) + + def test_list_networks_error_470(self): + self._test_list_networks(status=470) + + def test_list_networks_error_401(self): + self._test_list_networks(status=401) + + + + def test_list_network_details_json(self): + self._test_list_network_details(format='json') + + def test_list_network_details_xml(self): + self._test_list_network_details(format='xml') + + def test_list_network_details_alt_tenant(self): + self._test_list_network_details(tenant=TENANT_2) + + def test_list_network_details_error_470(self): + self._test_list_network_details(status=470) + + def test_list_network_details_error_401(self): + self._test_list_network_details(status=401) + + def test_list_network_details_error_420(self): + self._test_list_network_details(status=420) + + + + def test_create_network_json(self): + self._test_create_network(format='json') + + def test_create_network_xml(self): + self._test_create_network(format='xml') + + def test_create_network_alt_tenant(self): + self._test_create_network(tenant=TENANT_2) + + def test_create_network_error_470(self): + self._test_create_network(status=470) + + def test_create_network_error_401(self): + self._test_create_network(status=401) + + def test_create_network_error_400(self): + self._test_create_network(status=400) + + def test_create_network_error_422(self): + self._test_create_network(status=422) + + + + + def test_update_network_json(self): + self._test_update_network(format='json') + + def test_update_network_xml(self): + self._test_update_network(format='xml') + + def test_update_network_alt_tenant(self): + self._test_update_network(tenant=TENANT_2) + + def test_update_network_error_470(self): + self._test_update_network(status=470) + + def test_update_network_error_401(self): + self._test_update_network(status=401) + + def test_update_network_error_400(self): + self._test_update_network(status=400) + + def test_update_network_error_420(self): + self._test_update_network(status=420) + + def test_update_network_error_422(self): + self._test_update_network(status=422) + + + + def test_delete_network_json(self): + self._test_delete_network(format='json') + + def test_delete_network_xml(self): + self._test_delete_network(format='xml') + + def test_delete_network_alt_tenant(self): + self._test_delete_network(tenant=TENANT_2) + + def test_delete_network_error_470(self): + self._test_delete_network(status=470) + + def test_delete_network_error_401(self): + self._test_delete_network(status=401) + + def test_delete_network_error_420(self): + self._test_delete_network(status=420) + + def test_delete_network_error_421(self): + self._test_delete_network(status=421) + + + + def test_list_ports_json(self): + self._test_list_ports(format='json') + + def test_list_ports_xml(self): + self._test_list_ports(format='xml') + + def test_list_ports_alt_tenant(self): + self._test_list_ports(tenant=TENANT_2) + + def test_list_ports_error_470(self): + self._test_list_ports(status=470) + + def test_list_ports_error_401(self): + self._test_list_ports(status=401) + + def test_list_ports_error_420(self): + self._test_list_ports(status=420) + + + + def test_list_port_details_json(self): + self._test_list_ports(format='json') + + def test_list_port_details_xml(self): + self._test_list_ports(format='xml') + + def test_list_port_details_alt_tenant(self): + self._test_list_ports(tenant=TENANT_2) + + def test_list_port_details_error_470(self): + self._test_list_port_details(status=470) + + def test_list_port_details_error_401(self): + self._test_list_ports(status=401) + + def test_list_port_details_error_420(self): + self._test_list_ports(status=420) + + def test_list_port_details_error_430(self): + self._test_list_ports(status=430) + + + + def test_create_port_json(self): + self._test_create_port(format='json') + + def test_create_port_xml(self): + self._test_create_port(format='xml') + + def test_create_port_alt_tenant(self): + self._test_create_port(tenant=TENANT_2) + + def test_create_port_error_470(self): + self._test_create_port(status=470) + + def test_create_port_error_401(self): + self._test_create_port(status=401) + + def test_create_port_error_400(self): + self._test_create_port(status=400) + + def test_create_port_error_420(self): + self._test_create_port(status=420) + + def test_create_port_error_430(self): + self._test_create_port(status=430) + + def test_create_port_error_431(self): + self._test_create_port(status=431) + + + + def test_delete_port_json(self): + self._test_delete_port(format='json') + + def test_delete_port_xml(self): + self._test_delete_port(format='xml') + + def test_delete_port_alt_tenant(self): + self._test_delete_port(tenant=TENANT_2) + + def test_delete_port_error_470(self): + self._test_delete_port(status=470) + + def test_delete_port_error_401(self): + self._test_delete_port(status=401) + + def test_delete_port_error_420(self): + self._test_delete_port(status=420) + + def test_delete_port_error_430(self): + self._test_delete_port(status=430) + + def test_delete_port_error_432(self): + self._test_delete_port(status=432) + + + + def test_set_port_state_json(self): + self._test_set_port_state(format='json') + + def test_set_port_state_xml(self): + self._test_set_port_state(format='xml') + + def test_set_port_state_alt_tenant(self): + self._test_set_port_state(tenant=TENANT_2) + + def test_set_port_state_error_470(self): + self._test_set_port_state(status=470) + + def test_set_port_state_error_401(self): + self._test_set_port_state(status=401) + + def test_set_port_state_error_400(self): + self._test_set_port_state(status=400) + + def test_set_port_state_error_420(self): + self._test_set_port_state(status=420) + + def test_set_port_state_error_430(self): + self._test_set_port_state(status=430) + + def test_set_port_state_error_431(self): + self._test_set_port_state(status=431) + + + + def test_list_port_attachments_json(self): + self._test_list_port_attachments(format='json') + + def test_list_port_attachments_xml(self): + self._test_list_port_attachments(format='xml') + + def test_list_port_attachments_alt_tenant(self): + self._test_list_port_attachments(tenant=TENANT_2) + + def test_list_port_attachments_error_470(self): + self._test_list_port_attachments(status=470) + + def test_list_port_attachments_error_401(self): + self._test_list_port_attachments(status=401) + + def test_list_port_attachments_error_400(self): + self._test_list_port_attachments(status=400) + + def test_list_port_attachments_error_420(self): + self._test_list_port_attachments(status=420) + + def test_list_port_attachments_error_430(self): + self._test_list_port_attachments(status=430) + + + + def test_attach_resource_json(self): + self._test_attach_resource(format='json') + + def test_attach_resource_xml(self): + self._test_attach_resource(format='xml') + + def test_attach_resource_alt_tenant(self): + self._test_attach_resource(tenant=TENANT_2) + + def test_attach_resource_error_470(self): + self._test_attach_resource(status=470) + + def test_attach_resource_error_401(self): + self._test_attach_resource(status=401) + + def test_attach_resource_error_400(self): + self._test_attach_resource(status=400) + + def test_attach_resource_error_420(self): + self._test_attach_resource(status=420) + + def test_attach_resource_error_430(self): + self._test_attach_resource(status=430) + + def test_attach_resource_error_432(self): + self._test_attach_resource(status=432) + + def test_attach_resource_error_440(self): + self._test_attach_resource(status=440) + + + + def test_detach_resource_json(self): + self._test_detach_resource(format='json') + + def test_detach_resource_xml(self): + self._test_detach_resource(format='xml') + + def test_detach_resource_alt_tenant(self): + self._test_detach_resource(tenant=TENANT_2) + + def test_detach_resource_error_470(self): + self._test_detach_resource(status=470) + + def test_detach_resource_error_401(self): + self._test_detach_resource(status=401) + + def test_detach_resource_error_420(self): + self._test_detach_resource(status=420) + + def test_detach_resource_error_430(self): + self._test_detach_resource(status=430) + + + def test_ssl_certificates(self): + self._test_ssl_certificates() +