[Kimchi-devel] [PATCH 2/2] Server tests
Aline Manera
alinefm at linux.vnet.ibm.com
Fri Apr 10 18:02:37 UTC 2015
Move all the server related tests to the test_server.py file to make
maintenance easier.
Signed-off-by: Aline Manera <alinefm at linux.vnet.ibm.com>
---
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
More information about the Kimchi-devel
mailing list