From 6d97d94509d14e237c01cf6b25cdd2ad1cf2ed57 Mon Sep 17 00:00:00 2001 From: Brad Hall Date: Wed, 1 Jun 2011 11:00:15 -0700 Subject: [PATCH] Copy over miniclient from testscripts and port tests.py to use unittest --- smoketests/__init__.py | 0 smoketests/miniclient.py | 98 +++++++++++++++++++++++++++++ smoketests/tests.py | 133 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 231 insertions(+) create mode 100644 smoketests/__init__.py create mode 100644 smoketests/miniclient.py create mode 100644 smoketests/tests.py diff --git a/smoketests/__init__.py b/smoketests/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/smoketests/miniclient.py b/smoketests/miniclient.py new file mode 100644 index 00000000000..fb1ebc8fec7 --- /dev/null +++ b/smoketests/miniclient.py @@ -0,0 +1,98 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2011 Citrix 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. + +import httplib +import socket +import urllib + +class MiniClient(object): + + """A base client class - derived from Glance.BaseClient""" + + action_prefix = '/v0.1/tenants/{tenant_id}' + + def __init__(self, host, port, use_ssl): + """ + Creates a new client to some service. + + :param host: The host where service resides + :param port: The port where service resides + :param use_ssl: Should we use HTTPS? + """ + self.host = host + self.port = port + self.use_ssl = use_ssl + self.connection = None + + def get_connection_type(self): + """ + Returns the proper connection type + """ + if self.use_ssl: + return httplib.HTTPSConnection + else: + return httplib.HTTPConnection + + def do_request(self, tenant, method, action, body=None, + headers=None, params=None): + """ + Connects to the server and issues a request. + Returns the result data, or raises an appropriate exception if + HTTP status code is not 2xx + + :param method: HTTP method ("GET", "POST", "PUT", etc...) + :param body: string of data to send, or None (default) + :param headers: mapping of key/value pairs to add as headers + :param params: dictionary of key/value pairs to add to append + to action + + """ + action = MiniClient.action_prefix + action + action = action.replace('{tenant_id}',tenant) + if type(params) is dict: + action += '?' + urllib.urlencode(params) + + try: + connection_type = self.get_connection_type() + headers = headers or {} + + # Open connection and send request + c = connection_type(self.host, self.port) + c.request(method, action, body, headers) + res = c.getresponse() + status_code = self.get_status_code(res) + if status_code in (httplib.OK, + httplib.CREATED, + httplib.ACCEPTED, + httplib.NO_CONTENT): + return res + else: + raise Exception("Server returned error: %s" % res.read()) + + except (socket.error, IOError), e: + raise Exception("Unable to connect to " + "server. Got error: %s" % e) + + def get_status_code(self, response): + """ + Returns the integer status code from the response, which + can be either a Webob.Response (used in testing) or httplib.Response + """ + if hasattr(response, 'status_int'): + return response.status_int + else: + return response.status \ No newline at end of file diff --git a/smoketests/tests.py b/smoketests/tests.py new file mode 100644 index 00000000000..f9cd77418b5 --- /dev/null +++ b/smoketests/tests.py @@ -0,0 +1,133 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2011 Citrix Systems +# Copyright 2011 Nicira Networks +# 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. + +import gettext +import simplejson +import sys +import unittest + +gettext.install('quantum', unicode=1) + +from miniclient import MiniClient +from quantum.common.wsgi import Serializer + +HOST = '127.0.0.1' +PORT = 9696 +USE_SSL = False + +TENANT_ID = 'totore' +FORMAT = "json" + +test_network1_data = \ + {'network': {'network-name': 'test1' }} +test_network2_data = \ + {'network': {'network-name': 'test2' }} + +def print_response(res): + content = res.read() + print "Status: %s" %res.status + print "Content: %s" %content + return content + +class QuantumTest(unittest.TestCase): + def setUp(self): + self.client = MiniClient(HOST, PORT, USE_SSL) + + def create_network(self, data): + content_type = "application/" + FORMAT + body = Serializer().serialize(data, content_type) + res = self.client.do_request(TENANT_ID, 'POST', "/networks." + FORMAT, + body=body) + self.assertEqual(res.status, 200, "bad response: %s" % res.read()) + + def test_listNetworks(self): + self.create_network(test_network1_data) + self.create_network(test_network2_data) + res = self.client.do_request(TENANT_ID,'GET', "/networks." + FORMAT) + self.assertEqual(res.status, 200, "bad response: %s" % res.read()) + + def test_createNetwork(self): + self.create_network(test_network1_data) + + def test_createPort(self): + self.create_network(test_network1_data) + res = self.client.do_request(TENANT_ID, 'GET', "/networks." + FORMAT) + resdict = simplejson.loads(res.read()) + for n in resdict["networks"]: + net_id = n["id"] + + # Step 1 - List Ports for network (should not find any) + res = self.client.do_request(TENANT_ID, 'GET', + "/networks/%s/ports.%s" % (net_id, FORMAT)) + self.assertEqual(res.status, 200, "Bad response: %s" % res.read()) + output = res.read() + self.assertTrue(len(output) == 0, + "Found unexpected ports: %s" % output) + + # Step 2 - Create Port for network + res = self.client.do_request(TENANT_ID, 'POST', + "/networks/%s/ports.%s" % (net_id, FORMAT)) + self.assertEqual(res.status, 200, "Bad response: %s" % output) + + # Step 3 - List Ports for network (again); should find one + res = self.client.do_request(TENANT_ID, 'GET', + "/networks/%s/ports.%s" % (net_id, FORMAT)) + output = res.read() + self.assertEqual(res.status, 200, "Bad response: %s" % output) + resdict = simplejson.loads(output) + ids = [] + for p in resdict["ports"]: + ids.append(p["id"]) + self.assertTrue(len(ids) == 1, + "Didn't find expected # of ports (1): %s" % ids) + + def test_renameNetwork(self): + self.create_network(test_network1_data) + res = self.client.do_request(TENANT_ID, 'GET', "/networks." + FORMAT) + resdict = simplejson.loads(res.read()) + net_id = resdict["networks"][0]["id"] + + data = test_network1_data.copy() + data['network']['network-name'] = 'test_renamed' + content_type = "application/" + FORMAT + body = Serializer().serialize(data, content_type) + res = self.client.do_request(TENANT_ID, 'PUT', + "/networks/%s.%s" % (net_id, FORMAT), body=body) + resdict = simplejson.loads(res.read()) + self.assertTrue(resdict["networks"]["network"]["id"] == net_id, + "Network_rename: renamed network has a different uuid") + self.assertTrue(resdict["networks"]["network"]["name"] == "test_renamed", + "Network rename didn't take effect") + + def delete_networks(self): + # Remove all the networks created on the tenant + res = self.client.do_request(TENANT_ID, 'GET', "/networks." + FORMAT) + resdict = simplejson.loads(res.read()) + for n in resdict["networks"]: + net_id = n["id"] + res = self.client.do_request(TENANT_ID, 'DELETE', + "/networks/" + net_id + "." + FORMAT) + self.assertEqual(res.status, 202) + + def tearDown(self): + self.delete_networks() + +# Standard boilerplate to call the main() function. +if __name__ == '__main__': + suite = unittest.TestLoader().loadTestsFromTestCase(QuantumTest) + unittest.TextTestRunner(verbosity=2).run(suite)