summaryrefslogtreecommitdiffstats
path: root/nova/tests
diff options
context:
space:
mode:
Diffstat (limited to 'nova/tests')
-rw-r--r--nova/tests/api/fakes.py26
-rw-r--r--nova/tests/api/openstack/__init__.py21
-rw-r--r--nova/tests/api/openstack/fakes.py75
-rw-r--r--nova/tests/api/openstack/test_adminapi.py10
-rw-r--r--nova/tests/api/openstack/test_api.py27
-rw-r--r--nova/tests/api/openstack/test_auth.py18
-rw-r--r--nova/tests/api/openstack/test_flavors.py2
-rw-r--r--nova/tests/api/openstack/test_images.py4
-rw-r--r--nova/tests/api/openstack/test_servers.py40
-rw-r--r--nova/tests/api/test.py81
-rw-r--r--nova/tests/test_access.py22
-rw-r--r--nova/tests/test_api.py13
-rw-r--r--nova/tests/test_cloud.py8
-rw-r--r--nova/tests/test_compute.py15
-rw-r--r--nova/tests/test_console.py3
-rw-r--r--nova/tests/test_direct.py103
-rw-r--r--nova/tests/test_log.py8
-rw-r--r--nova/tests/test_middleware.py2
-rw-r--r--nova/tests/test_twistd.py2
19 files changed, 261 insertions, 219 deletions
diff --git a/nova/tests/api/fakes.py b/nova/tests/api/fakes.py
deleted file mode 100644
index 0aedcaff0..000000000
--- a/nova/tests/api/fakes.py
+++ /dev/null
@@ -1,26 +0,0 @@
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-# Copyright 2010 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.
-
-import webob.dec
-from nova import wsgi
-
-
-class APIStub(object):
- """Class to verify request and mark it was called."""
- @webob.dec.wsgify
- def __call__(self, req):
- return req.path_info
diff --git a/nova/tests/api/openstack/__init__.py b/nova/tests/api/openstack/__init__.py
index 9e183bd0d..14eaaa62c 100644
--- a/nova/tests/api/openstack/__init__.py
+++ b/nova/tests/api/openstack/__init__.py
@@ -15,23 +15,28 @@
# License for the specific language governing permissions and limitations
# under the License.
+import webob.dec
import unittest
from nova import context
from nova import flags
from nova.api.openstack.ratelimiting import RateLimitingMiddleware
from nova.api.openstack.common import limited
-from nova.tests.api.fakes import APIStub
-from nova import utils
+from nova.tests.api.openstack import fakes
from webob import Request
FLAGS = flags.FLAGS
+@webob.dec.wsgify
+def simple_wsgi(req):
+ return ""
+
+
class RateLimitingMiddlewareTest(unittest.TestCase):
def test_get_action_name(self):
- middleware = RateLimitingMiddleware(APIStub())
+ middleware = RateLimitingMiddleware(simple_wsgi)
def verify(method, url, action_name):
req = Request.blank(url)
@@ -61,19 +66,19 @@ class RateLimitingMiddlewareTest(unittest.TestCase):
self.assertTrue('Retry-After' in resp.headers)
def test_single_action(self):
- middleware = RateLimitingMiddleware(APIStub())
+ middleware = RateLimitingMiddleware(simple_wsgi)
self.exhaust(middleware, 'DELETE', '/servers/4', 'usr1', 100)
self.exhaust(middleware, 'DELETE', '/servers/4', 'usr2', 100)
def test_POST_servers_action_implies_POST_action(self):
- middleware = RateLimitingMiddleware(APIStub())
+ middleware = RateLimitingMiddleware(simple_wsgi)
self.exhaust(middleware, 'POST', '/servers/4', 'usr1', 10)
self.exhaust(middleware, 'POST', '/images/4', 'usr2', 10)
self.assertTrue(set(middleware.limiter._levels) == \
set(['usr1:POST', 'usr1:POST servers', 'usr2:POST']))
def test_POST_servers_action_correctly_ratelimited(self):
- middleware = RateLimitingMiddleware(APIStub())
+ middleware = RateLimitingMiddleware(simple_wsgi)
# Use up all of our "POST" allowance for the minute, 5 times
for i in range(5):
self.exhaust(middleware, 'POST', '/servers/4', 'usr1', 10)
@@ -83,9 +88,9 @@ class RateLimitingMiddlewareTest(unittest.TestCase):
self.exhaust(middleware, 'POST', '/servers/4', 'usr1', 0)
def test_proxy_ctor_works(self):
- middleware = RateLimitingMiddleware(APIStub())
+ middleware = RateLimitingMiddleware(simple_wsgi)
self.assertEqual(middleware.limiter.__class__.__name__, "Limiter")
- middleware = RateLimitingMiddleware(APIStub(), service_host='foobar')
+ middleware = RateLimitingMiddleware(simple_wsgi, service_host='foobar')
self.assertEqual(middleware.limiter.__class__.__name__, "WSGIAppProxy")
diff --git a/nova/tests/api/openstack/fakes.py b/nova/tests/api/openstack/fakes.py
index 194304e79..fb282f1c9 100644
--- a/nova/tests/api/openstack/fakes.py
+++ b/nova/tests/api/openstack/fakes.py
@@ -22,6 +22,9 @@ import string
import webob
import webob.dec
+from paste import urlmap
+
+from glance import client as glance_client
from nova import auth
from nova import context
@@ -29,6 +32,7 @@ from nova import exception as exc
from nova import flags
from nova import utils
import nova.api.openstack.auth
+from nova.api import openstack
from nova.api.openstack import auth
from nova.api.openstack import ratelimiting
from nova.image import glance
@@ -69,6 +73,17 @@ def fake_wsgi(self, req):
return self.application
+def wsgi_app(inner_application=None):
+ if not inner_application:
+ inner_application = openstack.APIRouter()
+ mapper = urlmap.URLMap()
+ api = openstack.FaultWrapper(auth.AuthMiddleware(
+ ratelimiting.RateLimitingMiddleware(inner_application)))
+ mapper['/v1.0'] = api
+ mapper['/'] = openstack.FaultWrapper(openstack.Versions())
+ return mapper
+
+
def stub_out_key_pair_funcs(stubs):
def key_pair(context, user_id):
return [dict(name='key', public_key='public_key')]
@@ -116,64 +131,60 @@ def stub_out_compute_api_snapshot(stubs):
stubs.Set(nova.compute.API, 'snapshot', snapshot)
-def stub_out_glance(stubs, initial_fixtures=[]):
+def stub_out_glance(stubs, initial_fixtures=None):
- class FakeParallaxClient:
+ class FakeGlanceClient:
def __init__(self, initial_fixtures):
- self.fixtures = initial_fixtures
+ self.fixtures = initial_fixtures or []
- def fake_get_image_index(self):
+ def fake_get_images(self):
return [dict(id=f['id'], name=f['name'])
for f in self.fixtures]
- def fake_get_image_details(self):
+ def fake_get_images_detailed(self):
return self.fixtures
- def fake_get_image_metadata(self, image_id):
+ def fake_get_image_meta(self, image_id):
for f in self.fixtures:
if f['id'] == image_id:
return f
return None
- def fake_add_image_metadata(self, image_data):
+ def fake_add_image(self, image_meta):
id = ''.join(random.choice(string.letters) for _ in range(20))
- image_data['id'] = id
- self.fixtures.append(image_data)
+ image_meta['id'] = id
+ self.fixtures.append(image_meta)
return id
- def fake_update_image_metadata(self, image_id, image_data):
- f = self.fake_get_image_metadata(image_id)
+ def fake_update_image(self, image_id, image_meta):
+ f = self.fake_get_image_meta(image_id)
if not f:
raise exc.NotFound
- f.update(image_data)
+ f.update(image_meta)
- def fake_delete_image_metadata(self, image_id):
- f = self.fake_get_image_metadata(image_id)
+ def fake_delete_image(self, image_id):
+ f = self.fake_get_image_meta(image_id)
if not f:
raise exc.NotFound
self.fixtures.remove(f)
- def fake_delete_all(self):
- self.fixtures = []
-
- fake_parallax_client = FakeParallaxClient(initial_fixtures)
- stubs.Set(nova.image.glance.ParallaxClient, 'get_image_index',
- fake_parallax_client.fake_get_image_index)
- stubs.Set(nova.image.glance.ParallaxClient, 'get_image_details',
- fake_parallax_client.fake_get_image_details)
- stubs.Set(nova.image.glance.ParallaxClient, 'get_image_metadata',
- fake_parallax_client.fake_get_image_metadata)
- stubs.Set(nova.image.glance.ParallaxClient, 'add_image_metadata',
- fake_parallax_client.fake_add_image_metadata)
- stubs.Set(nova.image.glance.ParallaxClient, 'update_image_metadata',
- fake_parallax_client.fake_update_image_metadata)
- stubs.Set(nova.image.glance.ParallaxClient, 'delete_image_metadata',
- fake_parallax_client.fake_delete_image_metadata)
- stubs.Set(nova.image.glance.GlanceImageService, 'delete_all',
- fake_parallax_client.fake_delete_all)
+ ##def fake_delete_all(self):
+ ## self.fixtures = []
+
+ GlanceClient = glance_client.Client
+ fake = FakeGlanceClient(initial_fixtures)
+
+ stubs.Set(GlanceClient, 'get_images', fake.fake_get_images)
+ stubs.Set(GlanceClient, 'get_images_detailed',
+ fake.fake_get_images_detailed)
+ stubs.Set(GlanceClient, 'get_image_meta', fake.fake_get_image_meta)
+ stubs.Set(GlanceClient, 'add_image', fake.fake_add_image)
+ stubs.Set(GlanceClient, 'update_image', fake.fake_update_image)
+ stubs.Set(GlanceClient, 'delete_image', fake.fake_delete_image)
+ #stubs.Set(GlanceClient, 'delete_all', fake.fake_delete_all)
class FakeToken(object):
diff --git a/nova/tests/api/openstack/test_adminapi.py b/nova/tests/api/openstack/test_adminapi.py
index 1b2e1654d..73120c31d 100644
--- a/nova/tests/api/openstack/test_adminapi.py
+++ b/nova/tests/api/openstack/test_adminapi.py
@@ -19,15 +19,19 @@ import unittest
import stubout
import webob
+from paste import urlmap
-import nova.api
from nova import flags
+from nova.api import openstack
+from nova.api.openstack import ratelimiting
+from nova.api.openstack import auth
from nova.tests.api.openstack import fakes
FLAGS = flags.FLAGS
class AdminAPITest(unittest.TestCase):
+
def setUp(self):
self.stubs = stubout.StubOutForTesting()
fakes.FakeAuthManager.auth_data = {}
@@ -45,7 +49,7 @@ class AdminAPITest(unittest.TestCase):
FLAGS.allow_admin_api = True
# We should still be able to access public operations.
req = webob.Request.blank('/v1.0/flavors')
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status_int, 200)
# TODO: Confirm admin operations are available.
@@ -53,7 +57,7 @@ class AdminAPITest(unittest.TestCase):
FLAGS.allow_admin_api = False
# We should still be able to access public operations.
req = webob.Request.blank('/v1.0/flavors')
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status_int, 200)
# TODO: Confirm admin operations are unavailable.
diff --git a/nova/tests/api/openstack/test_api.py b/nova/tests/api/openstack/test_api.py
index d8b202e21..db0fe1060 100644
--- a/nova/tests/api/openstack/test_api.py
+++ b/nova/tests/api/openstack/test_api.py
@@ -19,14 +19,18 @@ import unittest
import webob.exc
import webob.dec
-import nova.api.openstack
-from nova.api.openstack import API
-from nova.api.openstack import faults
from webob import Request
+from nova.api import openstack
+from nova.api.openstack import faults
+
class APITest(unittest.TestCase):
+ def _wsgi_app(self, inner_app):
+ # simpler version of the app than fakes.wsgi_app
+ return openstack.FaultWrapper(inner_app)
+
def test_exceptions_are_converted_to_faults(self):
@webob.dec.wsgify
@@ -46,29 +50,32 @@ class APITest(unittest.TestCase):
exc = webob.exc.HTTPNotFound(explanation='Raised a webob.exc')
return faults.Fault(exc)
- api = API()
-
- api.application = succeed
+ #api.application = succeed
+ api = self._wsgi_app(succeed)
resp = Request.blank('/').get_response(api)
self.assertFalse('computeFault' in resp.body, resp.body)
self.assertEqual(resp.status_int, 200, resp.body)
- api.application = raise_webob_exc
+ #api.application = raise_webob_exc
+ api = self._wsgi_app(raise_webob_exc)
resp = Request.blank('/').get_response(api)
self.assertFalse('computeFault' in resp.body, resp.body)
self.assertEqual(resp.status_int, 404, resp.body)
- api.application = raise_api_fault
+ #api.application = raise_api_fault
+ api = self._wsgi_app(raise_api_fault)
resp = Request.blank('/').get_response(api)
self.assertTrue('itemNotFound' in resp.body, resp.body)
self.assertEqual(resp.status_int, 404, resp.body)
- api.application = fail
+ #api.application = fail
+ api = self._wsgi_app(fail)
resp = Request.blank('/').get_response(api)
self.assertTrue('{"computeFault' in resp.body, resp.body)
self.assertEqual(resp.status_int, 500, resp.body)
- api.application = fail
+ #api.application = fail
+ api = self._wsgi_app(fail)
resp = Request.blank('/.xml').get_response(api)
self.assertTrue('<computeFault' in resp.body, resp.body)
self.assertEqual(resp.status_int, 500, resp.body)
diff --git a/nova/tests/api/openstack/test_auth.py b/nova/tests/api/openstack/test_auth.py
index 489a1dfbf..0dd65d321 100644
--- a/nova/tests/api/openstack/test_auth.py
+++ b/nova/tests/api/openstack/test_auth.py
@@ -53,7 +53,7 @@ class Test(unittest.TestCase):
req = webob.Request.blank('/v1.0/')
req.headers['X-Auth-User'] = 'herp'
req.headers['X-Auth-Key'] = 'derp'
- result = req.get_response(nova.api.API('os'))
+ result = req.get_response(fakes.wsgi_app())
self.assertEqual(result.status, '204 No Content')
self.assertEqual(len(result.headers['X-Auth-Token']), 40)
self.assertEqual(result.headers['X-CDN-Management-Url'],
@@ -67,7 +67,7 @@ class Test(unittest.TestCase):
req = webob.Request.blank('/v1.0/', {'HTTP_HOST': 'foo'})
req.headers['X-Auth-User'] = 'herp'
req.headers['X-Auth-Key'] = 'derp'
- result = req.get_response(nova.api.API('os'))
+ result = req.get_response(fakes.wsgi_app())
self.assertEqual(result.status, '204 No Content')
self.assertEqual(len(result.headers['X-Auth-Token']), 40)
self.assertEqual(result.headers['X-Server-Management-Url'],
@@ -81,7 +81,7 @@ class Test(unittest.TestCase):
fakes.FakeRouter)
req = webob.Request.blank('/v1.0/fake')
req.headers['X-Auth-Token'] = token
- result = req.get_response(nova.api.API('os'))
+ result = req.get_response(fakes.wsgi_app())
self.assertEqual(result.status, '200 OK')
self.assertEqual(result.headers['X-Test-Success'], 'True')
@@ -105,7 +105,7 @@ class Test(unittest.TestCase):
req = webob.Request.blank('/v1.0/')
req.headers['X-Auth-Token'] = 'bacon'
- result = req.get_response(nova.api.API('os'))
+ result = req.get_response(fakes.wsgi_app())
self.assertEqual(result.status, '401 Unauthorized')
self.assertEqual(self.destroy_called, True)
@@ -113,18 +113,18 @@ class Test(unittest.TestCase):
req = webob.Request.blank('/v1.0/')
req.headers['X-Auth-User'] = 'herp'
req.headers['X-Auth-Key'] = 'derp'
- result = req.get_response(nova.api.API('os'))
+ result = req.get_response(fakes.wsgi_app())
self.assertEqual(result.status, '401 Unauthorized')
def test_no_user(self):
req = webob.Request.blank('/v1.0/')
- result = req.get_response(nova.api.API('os'))
+ result = req.get_response(fakes.wsgi_app())
self.assertEqual(result.status, '401 Unauthorized')
def test_bad_token(self):
req = webob.Request.blank('/v1.0/')
req.headers['X-Auth-Token'] = 'baconbaconbacon'
- result = req.get_response(nova.api.API('os'))
+ result = req.get_response(fakes.wsgi_app())
self.assertEqual(result.status, '401 Unauthorized')
@@ -149,7 +149,7 @@ class TestLimiter(unittest.TestCase):
req = webob.Request.blank('/v1.0/')
req.headers['X-Auth-User'] = 'herp'
req.headers['X-Auth-Key'] = 'derp'
- result = req.get_response(nova.api.API('os'))
+ result = req.get_response(fakes.wsgi_app())
self.assertEqual(len(result.headers['X-Auth-Token']), 40)
token = result.headers['X-Auth-Token']
@@ -158,7 +158,7 @@ class TestLimiter(unittest.TestCase):
req = webob.Request.blank('/v1.0/fake')
req.method = 'POST'
req.headers['X-Auth-Token'] = token
- result = req.get_response(nova.api.API('os'))
+ result = req.get_response(fakes.wsgi_app())
self.assertEqual(result.status, '200 OK')
self.assertEqual(result.headers['X-Test-Success'], 'True')
diff --git a/nova/tests/api/openstack/test_flavors.py b/nova/tests/api/openstack/test_flavors.py
index 41018afdf..1bdaea161 100644
--- a/nova/tests/api/openstack/test_flavors.py
+++ b/nova/tests/api/openstack/test_flavors.py
@@ -39,7 +39,7 @@ class FlavorsTest(unittest.TestCase):
def test_get_flavor_list(self):
req = webob.Request.blank('/v1.0/flavors')
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
def test_get_flavor_by_id(self):
pass
diff --git a/nova/tests/api/openstack/test_images.py b/nova/tests/api/openstack/test_images.py
index 00ca739a5..5d9ddefbe 100644
--- a/nova/tests/api/openstack/test_images.py
+++ b/nova/tests/api/openstack/test_images.py
@@ -210,7 +210,7 @@ class ImageControllerWithGlanceServiceTest(unittest.TestCase):
def test_get_image_index(self):
req = webob.Request.blank('/v1.0/images')
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
fixture_index = [dict(id=f['id'], name=f['name']) for f
@@ -222,7 +222,7 @@ class ImageControllerWithGlanceServiceTest(unittest.TestCase):
def test_get_image_details(self):
req = webob.Request.blank('/v1.0/images/detail')
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
def _is_equivalent_subset(x, y):
diff --git a/nova/tests/api/openstack/test_servers.py b/nova/tests/api/openstack/test_servers.py
index 0396daf98..29883e7c8 100644
--- a/nova/tests/api/openstack/test_servers.py
+++ b/nova/tests/api/openstack/test_servers.py
@@ -100,14 +100,14 @@ class ServersTest(unittest.TestCase):
def test_get_server_by_id(self):
req = webob.Request.blank('/v1.0/servers/1')
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(res_dict['server']['id'], '1')
self.assertEqual(res_dict['server']['name'], 'server1')
def test_get_server_list(self):
req = webob.Request.blank('/v1.0/servers')
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
i = 0
@@ -160,14 +160,14 @@ class ServersTest(unittest.TestCase):
req.method = 'POST'
req.body = json.dumps(body)
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status_int, 200)
def test_update_no_body(self):
req = webob.Request.blank('/v1.0/servers/1')
req.method = 'PUT'
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status_int, 422)
def test_update_bad_params(self):
@@ -186,7 +186,7 @@ class ServersTest(unittest.TestCase):
req = webob.Request.blank('/v1.0/servers/1')
req.method = 'PUT'
req.body = self.body
- req.get_response(nova.api.API('os'))
+ req.get_response(fakes.wsgi_app())
def test_update_server(self):
inst_dict = dict(name='server_test', adminPass='bacon')
@@ -202,28 +202,28 @@ class ServersTest(unittest.TestCase):
req = webob.Request.blank('/v1.0/servers/1')
req.method = 'PUT'
req.body = self.body
- req.get_response(nova.api.API('os'))
+ req.get_response(fakes.wsgi_app())
def test_create_backup_schedules(self):
req = webob.Request.blank('/v1.0/servers/1/backup_schedules')
req.method = 'POST'
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status, '404 Not Found')
def test_delete_backup_schedules(self):
req = webob.Request.blank('/v1.0/servers/1/backup_schedules')
req.method = 'DELETE'
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status, '404 Not Found')
def test_get_server_backup_schedules(self):
req = webob.Request.blank('/v1.0/servers/1/backup_schedules')
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status, '404 Not Found')
def test_get_all_server_details(self):
req = webob.Request.blank('/v1.0/servers/detail')
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
i = 0
@@ -242,7 +242,7 @@ class ServersTest(unittest.TestCase):
req.method = 'POST'
req.content_type = 'application/json'
req.body = json.dumps(body)
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status_int, 202)
def test_server_unpause(self):
@@ -254,7 +254,7 @@ class ServersTest(unittest.TestCase):
req.method = 'POST'
req.content_type = 'application/json'
req.body = json.dumps(body)
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status_int, 202)
def test_server_suspend(self):
@@ -266,7 +266,7 @@ class ServersTest(unittest.TestCase):
req.method = 'POST'
req.content_type = 'application/json'
req.body = json.dumps(body)
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status_int, 202)
def test_server_resume(self):
@@ -278,19 +278,19 @@ class ServersTest(unittest.TestCase):
req.method = 'POST'
req.content_type = 'application/json'
req.body = json.dumps(body)
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status_int, 202)
def test_server_diagnostics(self):
req = webob.Request.blank("/v1.0/servers/1/diagnostics")
req.method = "GET"
- res = req.get_response(nova.api.API("os"))
+ res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status_int, 404)
def test_server_actions(self):
req = webob.Request.blank("/v1.0/servers/1/actions")
req.method = "GET"
- res = req.get_response(nova.api.API("os"))
+ res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status_int, 404)
def test_server_reboot(self):
@@ -301,7 +301,7 @@ class ServersTest(unittest.TestCase):
req.method = 'POST'
req.content_type = 'application/json'
req.body = json.dumps(body)
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
def test_server_rebuild(self):
body = dict(server=dict(
@@ -311,7 +311,7 @@ class ServersTest(unittest.TestCase):
req.method = 'POST'
req.content_type = 'application/json'
req.body = json.dumps(body)
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
def test_server_resize(self):
body = dict(server=dict(
@@ -321,7 +321,7 @@ class ServersTest(unittest.TestCase):
req.method = 'POST'
req.content_type = 'application/json'
req.body = json.dumps(body)
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
def test_delete_server_instance(self):
req = webob.Request.blank('/v1.0/servers/1')
@@ -335,7 +335,7 @@ class ServersTest(unittest.TestCase):
self.stubs.Set(nova.db.api, 'instance_destroy',
instance_destroy_mock)
- res = req.get_response(nova.api.API('os'))
+ res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status, '202 Accepted')
self.assertEqual(self.server_delete_called, True)
diff --git a/nova/tests/api/test.py b/nova/tests/api/test.py
deleted file mode 100644
index 9caa8c9d0..000000000
--- a/nova/tests/api/test.py
+++ /dev/null
@@ -1,81 +0,0 @@
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-# Copyright 2010 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.
-
-"""
-Test for the root WSGI middleware for all API controllers.
-"""
-
-import unittest
-
-import stubout
-import webob
-import webob.dec
-
-import nova.exception
-from nova import api
-from nova.tests.api.fakes import APIStub
-
-
-class Test(unittest.TestCase):
-
- def setUp(self):
- self.stubs = stubout.StubOutForTesting()
-
- def tearDown(self):
- self.stubs.UnsetAll()
-
- def _request(self, url, subdomain, **kwargs):
- environ_keys = {'HTTP_HOST': '%s.example.com' % subdomain}
- environ_keys.update(kwargs)
- req = webob.Request.blank(url, environ_keys)
- return req.get_response(api.API('ec2'))
-
- def test_openstack(self):
- self.stubs.Set(api.openstack, 'API', APIStub)
- result = self._request('/v1.0/cloud', 'api')
- self.assertEqual(result.body, "/cloud")
-
- def test_ec2(self):
- self.stubs.Set(api.ec2, 'API', APIStub)
- result = self._request('/services/cloud', 'ec2')
- self.assertEqual(result.body, "/cloud")
-
- def test_not_found(self):
- self.stubs.Set(api.ec2, 'API', APIStub)
- self.stubs.Set(api.openstack, 'API', APIStub)
- result = self._request('/test/cloud', 'ec2')
- self.assertNotEqual(result.body, "/cloud")
-
- def test_query_api_versions(self):
- result = self._request('/', 'api')
- self.assertTrue('CURRENT' in result.body)
-
- def test_metadata(self):
- def go(url):
- result = self._request(url, 'ec2', REMOTE_ADDR='128.192.151.2')
- # Each should get to the ORM layer and fail to find the IP
- self.assertRaises(nova.exception.NotFound, go, '/latest/')
- self.assertRaises(nova.exception.NotFound, go, '/2009-04-04/')
- self.assertRaises(nova.exception.NotFound, go, '/1.0/')
-
- def test_ec2_root(self):
- result = self._request('/', 'ec2')
- self.assertTrue('2007-12-15\n' in result.body)
-
-
-if __name__ == '__main__':
- unittest.main()
diff --git a/nova/tests/test_access.py b/nova/tests/test_access.py
index 0929903cf..e170ccee6 100644
--- a/nova/tests/test_access.py
+++ b/nova/tests/test_access.py
@@ -20,21 +20,31 @@ import unittest
import webob
from nova import context
-from nova import exception
from nova import flags
from nova import test
from nova.api import ec2
from nova.auth import manager
-
FLAGS = flags.FLAGS
-class Context(object):
+class FakeControllerClass(object):
pass
+class FakeApiRequest(object):
+ def __init__(self, action):
+ self.controller = FakeControllerClass()
+ self.action = action
+
+
class AccessTestCase(test.TestCase):
+ def _env_for(self, ctxt, action):
+ env = {}
+ env['ec2.context'] = ctxt
+ env['ec2.request'] = FakeApiRequest(action)
+ return env
+
def setUp(self):
super(AccessTestCase, self).setUp()
um = manager.AuthManager()
@@ -64,7 +74,7 @@ class AccessTestCase(test.TestCase):
return ['']
self.mw = ec2.Authorizer(noopWSGIApp)
- self.mw.action_roles = {'str': {
+ self.mw.action_roles = {'FakeControllerClass': {
'_allow_all': ['all'],
'_allow_none': [],
'_allow_project_manager': ['projectmanager'],
@@ -84,9 +94,7 @@ class AccessTestCase(test.TestCase):
def response_status(self, user, methodName):
ctxt = context.RequestContext(user, self.project)
- environ = {'ec2.context': ctxt,
- 'ec2.controller': 'some string',
- 'ec2.action': methodName}
+ environ = self._env_for(ctxt, methodName)
req = webob.Request.blank('/', environ)
resp = req.get_response(self.mw)
return resp.status_int
diff --git a/nova/tests/test_api.py b/nova/tests/test_api.py
index d22d7beb1..66a16b0cb 100644
--- a/nova/tests/test_api.py
+++ b/nova/tests/test_api.py
@@ -26,9 +26,8 @@ import StringIO
import webob
from nova import context
-from nova import flags
from nova import test
-from nova import api
+from nova.api import ec2
from nova.api.ec2 import cloud
from nova.api.ec2 import apirequest
from nova.auth import manager
@@ -79,7 +78,7 @@ class FakeHttplibConnection(object):
pass
-class XmlConversionTestCase(test.TrialTestCase):
+class XmlConversionTestCase(test.TestCase):
"""Unit test api xml conversion"""
def test_number_conversion(self):
conv = apirequest._try_convert
@@ -96,16 +95,14 @@ class XmlConversionTestCase(test.TrialTestCase):
self.assertEqual(conv('-0'), 0)
-class ApiEc2TestCase(test.TrialTestCase):
+class ApiEc2TestCase(test.TestCase):
"""Unit test for the cloud controller on an EC2 API"""
def setUp(self):
super(ApiEc2TestCase, self).setUp()
-
self.manager = manager.AuthManager()
-
self.host = '127.0.0.1'
-
- self.app = api.API('ec2')
+ self.app = ec2.Authenticate(ec2.Requestify(ec2.Executor(),
+ 'nova.api.ec2.cloud.CloudController'))
def expect_http(self, host=None, is_secure=False):
"""Returns a new EC2 connection"""
diff --git a/nova/tests/test_cloud.py b/nova/tests/test_cloud.py
index 2e350cd5a..771b1fcc0 100644
--- a/nova/tests/test_cloud.py
+++ b/nova/tests/test_cloud.py
@@ -21,6 +21,7 @@ import json
from M2Crypto import BIO
from M2Crypto import RSA
import os
+import shutil
import tempfile
import time
@@ -50,6 +51,8 @@ IMAGES_PATH = os.path.join(OSS_TEMPDIR, 'images')
os.makedirs(IMAGES_PATH)
+# TODO(termie): these tests are rather fragile, they should at the lest be
+# wiping database state after each run
class CloudTestCase(test.TestCase):
def setUp(self):
super(CloudTestCase, self).setUp()
@@ -287,6 +290,7 @@ class CloudTestCase(test.TestCase):
db.service_destroy(self.context, comp1['id'])
def test_instance_update_state(self):
+ # TODO(termie): what is this code even testing?
def instance(num):
return {
'reservation_id': 'r-1',
@@ -305,7 +309,8 @@ class CloudTestCase(test.TestCase):
'state': 0x01,
'user_data': ''}
rv = self.cloud._format_describe_instances(self.context)
- self.assert_(len(rv['reservationSet']) == 0)
+ logging.error(str(rv))
+ self.assertEqual(len(rv['reservationSet']), 0)
# simulate launch of 5 instances
# self.cloud.instances['pending'] = {}
@@ -368,6 +373,7 @@ class CloudTestCase(test.TestCase):
self.assertEqual('Foo Img', img.metadata['description'])
self._fake_set_image_description(self.context, 'ami-testing', '')
self.assertEqual('', img.metadata['description'])
+ shutil.rmtree(pathdir)
def test_update_of_instance_display_fields(self):
inst = db.instance_create(self.context, {})
diff --git a/nova/tests/test_compute.py b/nova/tests/test_compute.py
index a7d47961c..09f6ee94a 100644
--- a/nova/tests/test_compute.py
+++ b/nova/tests/test_compute.py
@@ -32,8 +32,9 @@ from nova import utils
from nova.auth import manager
-FLAGS = flags.FLAGS
LOG = logging.getLogger('nova.tests.compute')
+FLAGS = flags.FLAGS
+flags.DECLARE('stub_network', 'nova.compute.manager')
class ComputeTestCase(test.TestCase):
@@ -75,7 +76,7 @@ class ComputeTestCase(test.TestCase):
ref = self.compute_api.create(self.context,
FLAGS.default_instance_type, None, **instance)
try:
- self.assertNotEqual(ref[0].display_name, None)
+ self.assertNotEqual(ref[0]['display_name'], None)
finally:
db.instance_destroy(self.context, ref[0]['id'])
@@ -86,10 +87,14 @@ class ComputeTestCase(test.TestCase):
'user_id': self.user.id,
'project_id': self.project.id}
group = db.security_group_create(self.context, values)
- ref = self.compute_api.create(self.context,
- FLAGS.default_instance_type, None, security_group=['default'])
+ ref = self.compute_api.create(
+ self.context,
+ instance_type=FLAGS.default_instance_type,
+ image_id=None,
+ security_group=['default'])
try:
- self.assertEqual(len(ref[0]['security_groups']), 1)
+ self.assertEqual(len(db.security_group_get_by_instance(
+ self.context, ref[0]['id'])), 1)
finally:
db.security_group_destroy(self.context, group['id'])
db.instance_destroy(self.context, ref[0]['id'])
diff --git a/nova/tests/test_console.py b/nova/tests/test_console.py
index 31b5ca79c..85bf94458 100644
--- a/nova/tests/test_console.py
+++ b/nova/tests/test_console.py
@@ -111,12 +111,14 @@ class ConsoleTestCase(test.TestCase):
console_instances = [con['instance_id'] for con in pool.consoles]
self.assert_(instance_id in console_instances)
+ db.instance_destroy(self.context, instance_id)
def test_add_console_does_not_duplicate(self):
instance_id = self._create_instance()
cons1 = self.console.add_console(self.context, instance_id)
cons2 = self.console.add_console(self.context, instance_id)
self.assertEqual(cons1, cons2)
+ db.instance_destroy(self.context, instance_id)
def test_remove_console(self):
instance_id = self._create_instance()
@@ -127,3 +129,4 @@ class ConsoleTestCase(test.TestCase):
db.console_get,
self.context,
console_id)
+ db.instance_destroy(self.context, instance_id)
diff --git a/nova/tests/test_direct.py b/nova/tests/test_direct.py
new file mode 100644
index 000000000..8a74b2296
--- /dev/null
+++ b/nova/tests/test_direct.py
@@ -0,0 +1,103 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2010 United States Government as represented by the
+# Administrator of the National Aeronautics and Space Administration.
+# 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.
+
+"""Tests for Direct API."""
+
+import json
+import logging
+
+import webob
+
+from nova import compute
+from nova import context
+from nova import exception
+from nova import test
+from nova import utils
+from nova.api import direct
+from nova.tests import test_cloud
+
+
+class FakeService(object):
+ def echo(self, context, data):
+ return {'data': data}
+
+ def context(self, context):
+ return {'user': context.user_id,
+ 'project': context.project_id}
+
+
+class DirectTestCase(test.TestCase):
+ def setUp(self):
+ super(DirectTestCase, self).setUp()
+ direct.register_service('fake', FakeService())
+ self.router = direct.PostParamsMiddleware(
+ direct.JsonParamsMiddleware(
+ direct.Router()))
+ self.auth_router = direct.DelegatedAuthMiddleware(self.router)
+ self.context = context.RequestContext('user1', 'proj1')
+
+ def tearDown(self):
+ direct.ROUTES = {}
+
+ def test_delegated_auth(self):
+ req = webob.Request.blank('/fake/context')
+ req.headers['X-OpenStack-User'] = 'user1'
+ req.headers['X-OpenStack-Project'] = 'proj1'
+ resp = req.get_response(self.auth_router)
+ data = json.loads(resp.body)
+ self.assertEqual(data['user'], 'user1')
+ self.assertEqual(data['project'], 'proj1')
+
+ def test_json_params(self):
+ req = webob.Request.blank('/fake/echo')
+ req.environ['openstack.context'] = self.context
+ req.method = 'POST'
+ req.body = 'json=%s' % json.dumps({'data': 'foo'})
+ resp = req.get_response(self.router)
+ resp_parsed = json.loads(resp.body)
+ self.assertEqual(resp_parsed['data'], 'foo')
+
+ def test_post_params(self):
+ req = webob.Request.blank('/fake/echo')
+ req.environ['openstack.context'] = self.context
+ req.method = 'POST'
+ req.body = 'data=foo'
+ resp = req.get_response(self.router)
+ resp_parsed = json.loads(resp.body)
+ self.assertEqual(resp_parsed['data'], 'foo')
+
+ def test_proxy(self):
+ proxy = direct.Proxy(self.router)
+ rv = proxy.fake.echo(self.context, data='baz')
+ self.assertEqual(rv['data'], 'baz')
+
+
+class DirectCloudTestCase(test_cloud.CloudTestCase):
+ def setUp(self):
+ super(DirectCloudTestCase, self).setUp()
+ compute_handle = compute.API(image_service=self.cloud.image_service,
+ network_api=self.cloud.network_api,
+ volume_api=self.cloud.volume_api)
+ direct.register_service('compute', compute_handle)
+ self.router = direct.JsonParamsMiddleware(direct.Router())
+ proxy = direct.Proxy(self.router)
+ self.cloud.compute_api = proxy.compute
+
+ def tearDown(self):
+ super(DirectCloudTestCase, self).tearDown()
+ direct.ROUTES = {}
diff --git a/nova/tests/test_log.py b/nova/tests/test_log.py
index beb1d97cf..868a5ead3 100644
--- a/nova/tests/test_log.py
+++ b/nova/tests/test_log.py
@@ -9,7 +9,7 @@ def _fake_context():
return context.RequestContext(1, 1)
-class RootLoggerTestCase(test.TrialTestCase):
+class RootLoggerTestCase(test.TestCase):
def setUp(self):
super(RootLoggerTestCase, self).setUp()
self.log = log.logging.root
@@ -46,7 +46,7 @@ class RootLoggerTestCase(test.TrialTestCase):
self.assert_(True) # didn't raise exception
-class NovaFormatterTestCase(test.TrialTestCase):
+class NovaFormatterTestCase(test.TestCase):
def setUp(self):
super(NovaFormatterTestCase, self).setUp()
self.flags(logging_context_format_string="HAS CONTEXT "\
@@ -78,7 +78,7 @@ class NovaFormatterTestCase(test.TrialTestCase):
self.assertEqual("NOCTXT: baz --DBG\n", self.stream.getvalue())
-class NovaLoggerTestCase(test.TrialTestCase):
+class NovaLoggerTestCase(test.TestCase):
def setUp(self):
super(NovaLoggerTestCase, self).setUp()
self.flags(default_log_levels=["nova-test=AUDIT"], verbose=False)
@@ -96,7 +96,7 @@ class NovaLoggerTestCase(test.TrialTestCase):
self.assertEqual(log.AUDIT, l.level)
-class VerboseLoggerTestCase(test.TrialTestCase):
+class VerboseLoggerTestCase(test.TestCase):
def setUp(self):
super(VerboseLoggerTestCase, self).setUp()
self.flags(default_log_levels=["nova.test=AUDIT"], verbose=True)
diff --git a/nova/tests/test_middleware.py b/nova/tests/test_middleware.py
index 0febf52d6..9d49167ba 100644
--- a/nova/tests/test_middleware.py
+++ b/nova/tests/test_middleware.py
@@ -38,7 +38,7 @@ def conditional_forbid(req):
return 'OK'
-class LockoutTestCase(test.TrialTestCase):
+class LockoutTestCase(test.TestCase):
"""Test case for the Lockout middleware."""
def setUp(self): # pylint: disable-msg=C0103
super(LockoutTestCase, self).setUp()
diff --git a/nova/tests/test_twistd.py b/nova/tests/test_twistd.py
index 75007b9c8..ff8627c3b 100644
--- a/nova/tests/test_twistd.py
+++ b/nova/tests/test_twistd.py
@@ -28,7 +28,7 @@ from nova import test
FLAGS = flags.FLAGS
-class TwistdTestCase(test.TrialTestCase):
+class TwistdTestCase(test.TestCase):
def setUp(self):
super(TwistdTestCase, self).setUp()
self.Options = twistd.WrapTwistedOptions(twistd.TwistdServerOptions)
generated by cgit (git 2.34.1) at 2026-02-14 19:36:00 +0000