Move all the server related tests to the test_server.py file to make maintenance easier. Signed-off-by: Aline Manera <alinefm(a)linux.vnet.ibm.com&gt; --- tests/test_mockmodel.py | 35 ------- tests/test_model.py | 17 --- tests/test_rest.py | 176 ------------------------------- tests/test_server.py | 274 +++++++++++++++++++++++++++++++++++++++++++++--- 4 files changed, 260 insertions(+), 242 deletions(-) diff --git a/tests/test_mockmodel.py b/tests/test_mockmodel.py index 68a28d3..c7e733e 100644 --- a/tests/test_mockmodel.py +++ b/tests/test_mockmodel.py @@ -27,7 +27,6 @@ import unittest import kimchi.mockmodel from utils import get_free_port, patch_auth, request, run_server from utils import wait_task -from kimchi.control.base import Collection, Resource from kimchi.osinfo import get_template_default @@ -63,40 +62,6 @@ class MockModelTests(unittest.TestCase): def setUp(self): model.reset() - def test_collection(self): - c = Collection(model) - - # The base Collection is always empty - cherrypy.request.method = 'GET' - self.assertEquals('[]', c.index()) - - # POST and DELETE raise HTTP:405 by default - for method in ('POST', 'DELETE'): - cherrypy.request.method = method - try: - c.index() - except cherrypy.HTTPError, e: - self.assertEquals(405, e.code) - else: - self.fail("Expected exception not raised") - - def test_resource(self): - r = Resource(model) - - # Test the base Resource representation - cherrypy.request.method = 'GET' - self.assertEquals('{}', r.index()) - - # POST and DELETE raise HTTP:405 by default - for method in ('POST', 'DELETE'): - cherrypy.request.method = method - try: - r.index() - except cherrypy.HTTPError, e: - self.assertEquals(405, e.code) - else: - self.fail("Expected exception not raised") - def test_screenshot_refresh(self): # Create a VM req = json.dumps({'name': 'test', 'cdrom': fake_iso}) diff --git a/tests/test_model.py b/tests/test_model.py index 2a57507..c87ce50 100644 --- a/tests/test_model.py +++ b/tests/test_model.py @@ -26,7 +26,6 @@ import pwd import re import shutil import socket -import threading import time import urlparse import unittest @@ -690,22 +689,6 @@ class ModelTests(unittest.TestCase): self.assertEquals([], inst.vm_lookup(u'пeω-∨м')['users']) self.assertEquals([], inst.vm_lookup(u'пeω-∨м')['groups']) - def test_multithreaded_connection(self): - def worker(): - for i in xrange(100): - ret = inst.vms_get_list() - self.assertEquals('test', ret[0]) - - inst = model.Model('test:///default', self.tmp_store) - threads = [] - for i in xrange(100): - t = threading.Thread(target=worker) - t.setDaemon(True) - t.start() - threads.append(t) - for t in threads: - t.join() - def test_get_interfaces(self): inst = model.Model('test:///default', objstore_loc=self.tmp_store) diff --git a/tests/test_rest.py b/tests/test_rest.py index 4ecf3ce..7d9dacf 100644 --- a/tests/test_rest.py +++ b/tests/test_rest.py @@ -18,7 +18,6 @@ # License along with this library; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA -import base64 import json import os import re @@ -94,70 +93,6 @@ class RestTests(unittest.TestCase): resp = self.request(*args) self.assertEquals(code, resp.status) - def assertValidJSON(self, txt): - try: - json.loads(txt) - except ValueError: - self.fail("Invalid JSON: %s" % txt) - - def test_404(self): - """ - A non-existent path should return HTTP:404 - """ - url_list = ['/doesnotexist', '/vms/blah'] - for url in url_list: - self.assertHTTPStatus(404, url) - - # Make sure it fails for bad HTML requests - # We must be authenticated first. Otherwise all requests will return - # HTTP:401. Since HTTP Simple Auth is not allowed for text/html, we - # need to use the login API and establish a session. - user, pw = kimchi.mockmodel.fake_user.items()[0] - req = json.dumps({'username': user, 'password': pw}) - resp = self.request('/login', req, 'POST') - self.assertEquals(200, resp.status) - cookie = resp.getheader('set-cookie') - - self.assertHTTPStatus(404, url, None, 'GET', - {'Accept': 'text/html', - 'Cookie': cookie}) - - # Verify it works for DELETE too - self.assertHTTPStatus(404, '/templates/blah', '', 'DELETE') - - def test_accepts(self): - """ - Verify the following expectations regarding the client Accept header: - If omitted, default to html - If 'application/json', serve the rest api - If 'text/html', serve the UI - If both of the above (in any order), serve the rest api - If neither of the above, HTTP:406 - """ - resp = self.request("/", headers={}) - location = resp.getheader('location') - self.assertTrue(location.endswith("login.html")) - resp = self.request("/login.html", headers={}) - self.assertTrue('<!doctype html>' in resp.read().lower()) - - resp = self.request("/", headers={'Accept': 'application/json'}) - self.assertValidJSON(resp.read()) - - resp = self.request("/", headers={'Accept': 'text/html'}) - location = resp.getheader('location') - self.assertTrue(location.endswith("login.html")) - - resp = self.request("/", headers={'Accept': - 'application/json, text/html'}) - self.assertValidJSON(resp.read()) - - resp = self.request("/", headers={'Accept': - 'text/html, application/json'}) - self.assertValidJSON(resp.read()) - - h = {'Accept': 'text/plain'} - self.assertHTTPStatus(406, "/", None, 'GET', h) - def test_host_devices(self): resp = self.request('/host/devices?_cap=scsi_host') nodedevs = json.loads(resp.read()) @@ -1077,92 +1012,6 @@ class RestTests(unittest.TestCase): resp = self.request('/peers').read() self.assertEquals([], json.loads(resp)) - def test_auth_unprotected(self): - hdrs = {'AUTHORIZATION': ''} - uris = ['/js/kimchi.min.js', - '/css/theme-default.min.css', - '/libs/jquery-1.10.0.min.js', - '/images/icon-vm.png', - '/login.html', - '/logout'] - for uri in uris: - resp = self.request(uri, None, 'HEAD', hdrs) - self.assertEquals(200, resp.status) - - user, pw = kimchi.mockmodel.fake_user.items()[0] - req = json.dumps({'username': user, 'password': pw}) - resp = self.request('/login', req, 'POST', hdrs) - self.assertEquals(200, resp.status) - - def test_auth_protected(self): - hdrs = {'AUTHORIZATION': ''} - uris = ['/vms', - '/vms/doesnotexist', - '/tasks'] - for uri in uris: - resp = self.request(uri, None, 'GET', hdrs) - self.assertEquals(401, resp.status) - - def test_auth_bad_creds(self): - # Test HTTPBA - hdrs = {'AUTHORIZATION': "Basic " + base64.b64encode("nouser:badpass")} - resp = self.request('/vms', None, 'GET', hdrs) - self.assertEquals(401, resp.status) - - # Test REST API - hdrs = {'AUTHORIZATION': ''} - req = json.dumps({'username': 'nouser', 'password': 'badpass'}) - resp = self.request('/login', req, 'POST', hdrs) - self.assertEquals(401, resp.status) - - def test_auth_browser_no_httpba(self): - # Kimchi detects REST requests from the browser by looking for a - # specific header - hdrs = {"X-Requested-With": "XMLHttpRequest"} - - # Try our request (Note that request() will add a valid HTTPBA header) - resp = self.request('/vms', None, 'GET', hdrs) - self.assertEquals(401, resp.status) - self.assertEquals(None, resp.getheader('WWW-Authenticate')) - - def test_auth_session(self): - hdrs = {'AUTHORIZATION': '', - 'Content-Type': 'application/json', - 'Accept': 'application/json'} - - # Test we are logged out - resp = self.request('/tasks', None, 'GET', hdrs) - self.assertEquals(401, resp.status) - - # Execute a login call - user, pw = kimchi.mockmodel.fake_user.items()[0] - req = json.dumps({'username': user, 'password': pw}) - resp = self.request('/login', req, 'POST', hdrs) - self.assertEquals(200, resp.status) - - user_info = json.loads(resp.read()) - self.assertEquals(sorted(user_info.keys()), - ['groups', 'roles', 'username']) - roles = user_info['roles'] - for tab, role in roles.iteritems(): - self.assertEquals(role, u'admin') - - cookie = resp.getheader('set-cookie') - hdrs['Cookie'] = cookie - - # Test we are logged in with the cookie - resp = self.request('/tasks', None, 'GET', hdrs) - self.assertEquals(200, resp.status) - - # Execute a logout call - resp = self.request('/logout', '{}', 'POST', hdrs) - self.assertEquals(200, resp.status) - del hdrs['Cookie'] - - # Test we are logged out - resp = self.request('/tasks', None, 'GET', hdrs) - self.assertEquals(401, resp.status) - def test_distros(self): resp = self.request('/config/distros').read() distros = json.loads(resp) @@ -1306,31 +1155,6 @@ class RestTests(unittest.TestCase): self.assertEquals(task_info['status'], 'finished') self.assertIn(u'All packages updated', task_info['message']) - def test_get_param(self): - req = json.dumps({'name': 'test', 'cdrom': fake_iso}) - self.request('/templates', req, 'POST') - - # Create a VM - req = json.dumps({'name': 'test-vm1', 'template': '/templates/test'}) - resp = self.request('/vms', req, 'POST') - self.assertEquals(201, resp.status) - req = json.dumps({'name': 'test-vm2', 'template': '/templates/test'}) - resp = self.request('/vms', req, 'POST') - self.assertEquals(201, resp.status) - - resp = request(host, ssl_port, '/vms') - self.assertEquals(200, resp.status) - res = json.loads(resp.read()) - self.assertEquals(3, len(res)) - - # FIXME: control/base.py also allows filter by regex so it is returning - # 2 vms when querying for 'test-vm1': 'test' and 'test-vm1' - resp = request(host, ssl_port, '/vms?name=test-vm1') - self.assertEquals(200, resp.status) - res = json.loads(resp.read()) - self.assertEquals(2, len(res)) - self.assertIn('test-vm1', [r['name'] for r in res]) - def test_repositories(self): def verify_repo(t, res): for field in ('repo_id', 'enabled', 'baseurl', 'config'): diff --git a/tests/test_server.py b/tests/test_server.py index bebc383..42bebbe 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,7 +1,7 @@ # # Project Kimchi # -# Copyright IBM, Corp. 2013-2014 +# Copyright IBM, Corp. 2013-2015 # # This library is free software; you can redistribute it and/or # modify it under the terms of the GNU Lesser General Public @@ -17,29 +17,275 @@ # License along with this library; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +import base64 +import cherrypy +import json import os +import tempfile +import threading import unittest +from functools import partial import utils -import kimchi.mockmodel +from kimchi import mockmodel +from kimchi.control.base import Collection, Resource + +test_server = None +model = None +host = None +port = None +ssl_port = None +cherrypy_port = None +tmpfile = None + + +def setUpModule(): + global test_server, model, host, port, ssl_port, cherrypy_port, tmpfile + + utils.patch_auth() + tmpfile = tempfile.mktemp() + model = mockmodel.MockModel(tmpfile) + host = '127.0.0.1' + port = utils.get_free_port('http') + ssl_port = utils.get_free_port('https') + cherrypy_port = utils.get_free_port('cherrypy_port') + test_server = utils.run_server(host, port, ssl_port, test_mode=True, + cherrypy_port=cherrypy_port, model=model) + + +def tearDownModule(): + test_server.stop() + os.unlink(tmpfile) class ServerTests(unittest.TestCase): + def setUp(self): + self.request = partial(utils.request, host, ssl_port) + model.reset() + + def assertValidJSON(self, txt): + try: + json.loads(txt) + except ValueError: + self.fail("Invalid JSON: %s" % txt) + def test_server_start(self): """ Test that we can start a server and receive HTTP:200. """ - host = '127.0.0.1' - port = utils.get_free_port('http') - ssl_port = utils.get_free_port('https') - model = kimchi.mockmodel.MockModel('/tmp/obj-store-test') - s = utils.run_server(host, port, ssl_port, test_mode=True, model=model) - try: - resp = utils.request(host, ssl_port, '/') + resp = self.request('/') + self.assertEquals(200, resp.status) + + def test_multithreaded_connection(self): + def worker(): + for i in xrange(100): + ret = model.vms_get_list() + self.assertEquals('test', ret[0]) + + threads = [] + for i in xrange(100): + t = threading.Thread(target=worker) + t.setDaemon(True) + t.start() + threads.append(t) + for t in threads: + t.join() + + def test_collection(self): + c = Collection(model) + + # The base Collection is always empty + cherrypy.request.method = 'GET' + cherrypy.request.headers['Accept'] = 'application/json' + self.assertEquals('[]', c.index()) + + # POST and DELETE raise HTTP:405 by default + for method in ('POST', 'DELETE'): + cherrypy.request.method = method + try: + c.index() + except cherrypy.HTTPError, e: + self.assertEquals(405, e.code) + else: + self.fail("Expected exception not raised") + + def test_resource(self): + r = Resource(model) + + # Test the base Resource representation + cherrypy.request.method = 'GET' + cherrypy.request.headers['Accept'] = 'application/json' + self.assertEquals('{}', r.index()) + + # POST and DELETE raise HTTP:405 by default + for method in ('POST', 'DELETE'): + cherrypy.request.method = method + try: + r.index() + except cherrypy.HTTPError, e: + self.assertEquals(405, e.code) + else: + self.fail("Expected exception not raised") + + def test_404(self): + """ + A non-existent path should return HTTP:404 + """ + url_list = ['/doesnotexist', '/vms/blah'] + for url in url_list: + resp = self.request(url) + self.assertEquals(404, resp.status) + + # Verify it works for DELETE too + resp = self.request('/templates/blah', '', 'DELETE') + self.assertEquals(404, resp.status) + + def test_accepts(self): + """ + Verify the following expectations regarding the client Accept header: + If omitted, default to html + If 'application/json', serve the rest api + If 'text/html', serve the UI + If both of the above (in any order), serve the rest api + If neither of the above, HTTP:406 + """ + resp = self.request("/", headers={}) + location = resp.getheader('location') + self.assertTrue(location.endswith("login.html")) + resp = self.request("/login.html", headers={}) + self.assertTrue('<!doctype html>' in resp.read().lower()) + + resp = self.request("/", headers={'Accept': 'application/json'}) + self.assertValidJSON(resp.read()) + + resp = self.request("/", headers={'Accept': 'text/html'}) + location = resp.getheader('location') + self.assertTrue(location.endswith("login.html")) + + resp = self.request("/", headers={'Accept': + 'application/json, text/html'}) + self.assertValidJSON(resp.read()) + + resp = self.request("/", headers={'Accept': + 'text/html, application/json'}) + self.assertValidJSON(resp.read()) + + h = {'Accept': 'text/plain'} + resp = self.request('/', None, 'GET', h) + self.assertEquals(406, resp.status) + + def test_auth_unprotected(self): + hdrs = {'AUTHORIZATION': ''} + uris = ['/js/kimchi.min.js', + '/css/theme-default.min.css', + '/libs/jquery-1.10.0.min.js', + '/images/icon-vm.png', + '/login.html', + '/logout'] + + for uri in uris: + resp = self.request(uri, None, 'HEAD', hdrs) self.assertEquals(200, resp.status) - except: - raise - finally: - os.unlink('/tmp/obj-store-test') - s.stop() + + def test_auth_protected(self): + hdrs = {'AUTHORIZATION': ''} + uris = ['/vms', + '/vms/doesnotexist', + '/tasks'] + + for uri in uris: + resp = self.request(uri, None, 'GET', hdrs) + self.assertEquals(401, resp.status) + + def test_auth_bad_creds(self): + # Test HTTPBA + hdrs = {'AUTHORIZATION': "Basic " + base64.b64encode("nouser:badpass")} + resp = self.request('/vms', None, 'GET', hdrs) + self.assertEquals(401, resp.status) + + # Test REST API + hdrs = {'AUTHORIZATION': ''} + req = json.dumps({'username': 'nouser', 'password': 'badpass'}) + resp = self.request('/login', req, 'POST', hdrs) + self.assertEquals(401, resp.status) + + def test_auth_browser_no_httpba(self): + # Kimchi detects REST requests from the browser by looking for a + # specific header + hdrs = {"X-Requested-With": "XMLHttpRequest"} + + # Try our request (Note that request() will add a valid HTTPBA header) + resp = self.request('/vms', None, 'GET', hdrs) + self.assertEquals(401, resp.status) + self.assertEquals(None, resp.getheader('WWW-Authenticate')) + + def test_auth_session(self): + hdrs = {'AUTHORIZATION': '', + 'Content-Type': 'application/json', + 'Accept': 'application/json'} + + # Test we are logged out + resp = self.request('/tasks', None, 'GET', hdrs) + self.assertEquals(401, resp.status) + + # Execute a login call + user, pw = mockmodel.fake_user.items()[0] + req = json.dumps({'username': user, 'password': pw}) + resp = self.request('/login', req, 'POST', hdrs) + self.assertEquals(200, resp.status) + + user_info = json.loads(resp.read()) + self.assertEquals(sorted(user_info.keys()), + ['groups', 'roles', 'username']) + roles = user_info['roles'] + for tab, role in roles.iteritems(): + self.assertEquals(role, u'admin') + + cookie = resp.getheader('set-cookie') + hdrs['Cookie'] = cookie + + # Test we are logged in with the cookie + resp = self.request('/tasks', None, 'GET', hdrs) + self.assertEquals(200, resp.status) + + # Execute a logout call + resp = self.request('/logout', '{}', 'POST', hdrs) + self.assertEquals(200, resp.status) + del hdrs['Cookie'] + + # Test we are logged out + resp = self.request('/tasks', None, 'GET', hdrs) + self.assertEquals(401, resp.status) + + def test_get_param(self): + # Create a mock ISO file + mockiso = '/tmp/mock.iso' + open('/tmp/mock.iso', 'w').close() + + req = json.dumps({'name': 'test', 'cdrom': mockiso}) + self.request('/templates', req, 'POST') + + # Create a VM + req = json.dumps({'name': 'test-vm1', 'template': '/templates/test'}) + resp = self.request('/vms', req, 'POST') + self.assertEquals(201, resp.status) + req = json.dumps({'name': 'test-vm2', 'template': '/templates/test'}) + resp = self.request('/vms', req, 'POST') + self.assertEquals(201, resp.status) + + # Remove mock iso + os.unlink(mockiso) + + resp = self.request('/vms') + self.assertEquals(200, resp.status) + res = json.loads(resp.read()) + self.assertEquals(3, len(res)) + + # FIXME: control/base.py also allows filter by regex so it is returning + # 2 vms when querying for 'test-vm1': 'test' and 'test-vm1' + resp = self.request('/vms?name=test-vm1') + self.assertEquals(200, resp.status) + res = json.loads(resp.read()) + self.assertEquals(2, len(res)) + self.assertIn('test-vm1', [r['name'] for r in res]) -- 2.1.0