openstack/nova
Code Issues Proposed changes

Make tests start with a clean database for every test.

Browse Source
This commit is contained in:
Vishvananda Ishaya 2011-02-25 01:04:25 +00:00 committed by Tarmac
commit ed7c71f56c
23 changed files with 105 additions and 108 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
bin/nova-dhcpbridge
View File

@ -105,18 +105,7 @@ def main():
logging.setup()
interface = os.environ.get('DNSMASQ_INTERFACE', 'br0')
if int(os.environ.get('TESTING', '0')):
FLAGS.fake_rabbit = True
FLAGS.network_size = 16
FLAGS.connection_type = 'fake'
FLAGS.fake_network = True
FLAGS.auth_driver = 'nova.auth.dbdriver.DbDriver'
FLAGS.num_networks = 5
path = os.path.abspath(os.path.join(os.path.dirname(__file__),
'..',
'nova',
'tests',
'tests.sqlite'))
FLAGS.sql_connection = 'sqlite:///%s' % path
from nova.tests import fake_flags
action = argv[1]
if action in ['add', 'del', 'old']:
mac = argv[2]

3
nova/flags.py
View File

@ -324,8 +324,9 @@ DEFINE_string('state_path', os.path.join(os.path.dirname(__file__), '../'),
DEFINE_string('logdir', None, 'output to a per-service log file in named '
'directory')
DEFINE_string('sqlite_db', 'nova.sqlite', 'file name for sqlite')
DEFINE_string('sql_connection',
'sqlite:///$state_path/nova.sqlite',
'sqlite:///$state_path/$sqlite_db',
'connection string for sql database')
DEFINE_integer('sql_idle_timeout',
3600,

32
nova/test.py
View File

@ -22,26 +22,28 @@ Allows overriding of flags for use of fakes,
and some black magic for inline callbacks.
"""
import datetime
import os
import shutil
import uuid
import unittest
import mox
import shutil
import stubout
from nova import context
from nova import db
from nova import fakerabbit
from nova import flags
from nova import log as logging
from nova import rpc
from nova import service
from nova.network import manager as network_manager
FLAGS = flags.FLAGS
flags.DEFINE_bool('flush_db', True,
'Flush the database before running fake tests')
flags.DEFINE_string('sqlite_clean_db', 'clean.sqlite',
'File name of clean sqlite db')
flags.DEFINE_bool('fake_tests', True,
'should we use everything for testing')
@ -66,15 +68,8 @@ class TestCase(unittest.TestCase):
# now that we have some required db setup for the system
# to work properly.
self.start = datetime.datetime.utcnow()
ctxt = context.get_admin_context()
if db.network_count(ctxt) != 5:
network_manager.VlanManager().create_networks(ctxt,
FLAGS.fixed_range,
5, 16,
FLAGS.fixed_range_v6,
FLAGS.vlan_start,
FLAGS.vpn_start,
)
shutil.copyfile(os.path.join(FLAGS.state_path, FLAGS.sqlite_clean_db),
os.path.join(FLAGS.state_path, FLAGS.sqlite_db))
# emulate some of the mox stuff, we can't use the metaclass
# because it screws with our generators
@ -96,17 +91,6 @@ class TestCase(unittest.TestCase):
self.mox.VerifyAll()
super(TestCase, self).tearDown()
finally:
try:
# Clean up any ips associated during the test.
ctxt = context.get_admin_context()
db.fixed_ip_disassociate_all_by_timeout(ctxt, FLAGS.host,
self.start)
db.network_disassociate_all(ctxt)
db.security_group_destroy_all(ctxt)
except Exception:
pass
# Clean out fake_rabbit's queue if we used it
if FLAGS.fake_rabbit:
fakerabbit.reset_all()

25
nova/tests/__init__.py
View File

@ -37,5 +37,30 @@ setattr(__builtin__, '_', lambda x: x)
def setup():
import os
import shutil
from nova import context
from nova import flags
from nova.db import migration
from nova.network import manager as network_manager
from nova.tests import fake_flags
FLAGS = flags.FLAGS
testdb = os.path.join(FLAGS.state_path, FLAGS.sqlite_db)
if os.path.exists(testdb):
os.unlink(testdb)
migration.db_sync()
ctxt = context.get_admin_context()
network_manager.VlanManager().create_networks(ctxt,
FLAGS.fixed_range,
FLAGS.num_networks,
FLAGS.network_size,
FLAGS.fixed_range_v6,
FLAGS.vlan_start,
FLAGS.vpn_start,
)
cleandb = os.path.join(FLAGS.state_path, FLAGS.sqlite_clean_db)
shutil.copyfile(testdb, cleandb)

4
nova/tests/api/openstack/__init__.py
View File

@ -16,7 +16,7 @@
# under the License.
import webob.dec
import unittest
from nova import test
from nova import context
from nova import flags
@ -33,7 +33,7 @@ def simple_wsgi(req):
return ""
class RateLimitingMiddlewareTest(unittest.TestCase):
class RateLimitingMiddlewareTest(test.TestCase):
def test_get_action_name(self):
middleware = RateLimitingMiddleware(simple_wsgi)

11
nova/tests/api/openstack/test_adminapi.py
View File

@ -15,13 +15,13 @@
# License for the specific language governing permissions and limitations
# under the License.
import unittest
import stubout
import webob
from paste import urlmap
from nova import flags
from nova import test
from nova.api import openstack
from nova.api.openstack import ratelimiting
from nova.api.openstack import auth
@ -30,9 +30,10 @@ from nova.tests.api.openstack import fakes
FLAGS = flags.FLAGS
class AdminAPITest(unittest.TestCase):
class AdminAPITest(test.TestCase):
def setUp(self):
super(AdminAPITest, self).setUp()
self.stubs = stubout.StubOutForTesting()
fakes.FakeAuthManager.auth_data = {}
fakes.FakeAuthDatabase.data = {}
@ -44,6 +45,7 @@ class AdminAPITest(unittest.TestCase):
def tearDown(self):
self.stubs.UnsetAll()
FLAGS.allow_admin_api = self.allow_admin
super(AdminAPITest, self).tearDown()
def test_admin_enabled(self):
FLAGS.allow_admin_api = True
@ -58,8 +60,5 @@ class AdminAPITest(unittest.TestCase):
# We should still be able to access public operations.
req = webob.Request.blank('/v1.0/flavors')
res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status_int, 200)
# TODO: Confirm admin operations are unavailable.
if __name__ == '__main__':
unittest.main()
self.assertEqual(res.status_int, 200)

4
nova/tests/api/openstack/test_api.py
View File

@ -15,17 +15,17 @@
# License for the specific language governing permissions and limitations
# under the License.
import unittest
import webob.exc
import webob.dec
from webob import Request
from nova import test
from nova.api import openstack
from nova.api.openstack import faults
class APITest(unittest.TestCase):
class APITest(test.TestCase):
def _wsgi_app(self, inner_app):
# simpler version of the app than fakes.wsgi_app

14
nova/tests/api/openstack/test_auth.py
View File

@ -16,7 +16,6 @@
# under the License.
import datetime
import unittest
import stubout
import webob
@ -27,12 +26,14 @@ import nova.api.openstack.auth
import nova.auth.manager
from nova import auth
from nova import context
from nova import test
from nova.tests.api.openstack import fakes
class Test(unittest.TestCase):
class Test(test.TestCase):
def setUp(self):
super(Test, self).setUp()
self.stubs = stubout.StubOutForTesting()
self.stubs.Set(nova.api.openstack.auth.AuthMiddleware,
'__init__', fakes.fake_auth_init)
@ -45,6 +46,7 @@ class Test(unittest.TestCase):
def tearDown(self):
self.stubs.UnsetAll()
fakes.fake_data_store = {}
super(Test, self).tearDown()
def test_authorize_user(self):
f = fakes.FakeAuthManager()
@ -128,8 +130,9 @@ class Test(unittest.TestCase):
self.assertEqual(result.status, '401 Unauthorized')
class TestLimiter(unittest.TestCase):
class TestLimiter(test.TestCase):
def setUp(self):
super(TestLimiter, self).setUp()
self.stubs = stubout.StubOutForTesting()
self.stubs.Set(nova.api.openstack.auth.AuthMiddleware,
'__init__', fakes.fake_auth_init)
@ -141,6 +144,7 @@ class TestLimiter(unittest.TestCase):
def tearDown(self):
self.stubs.UnsetAll()
fakes.fake_data_store = {}
super(TestLimiter, self).tearDown()
def test_authorize_token(self):
f = fakes.FakeAuthManager()
@ -161,7 +165,3 @@ class TestLimiter(unittest.TestCase):
result = req.get_response(fakes.wsgi_app())
self.assertEqual(result.status, '200 OK')
self.assertEqual(result.headers['X-Test-Success'], 'True')
if __name__ == '__main__':
unittest.main()

5
nova/tests/api/openstack/test_common.py
View File

@ -19,14 +19,14 @@
Test suites for 'common' code used throughout the OpenStack HTTP API.
"""
import unittest
from webob import Request
from nova import test
from nova.api.openstack.common import limited
class LimiterTest(unittest.TestCase):
class LimiterTest(test.TestCase):
"""
Unit tests for the `nova.api.openstack.common.limited` method which takes
in a list of items and, depending on the 'offset' and 'limit' GET params,
@ -37,6 +37,7 @@ class LimiterTest(unittest.TestCase):
"""
Run before each test.
"""
super(LimiterTest, self).setUp()
self.tiny = range(1)
self.small = range(10)
self.medium = range(1000)

4
nova/tests/api/openstack/test_faults.py
View File

@ -15,15 +15,15 @@
# License for the specific language governing permissions and limitations
# under the License.
import unittest
import webob
import webob.dec
import webob.exc
from nova import test
from nova.api.openstack import faults
class TestFaults(unittest.TestCase):
class TestFaults(test.TestCase):
def test_fault_parts(self):
req = webob.Request.blank('/.xml')

10
nova/tests/api/openstack/test_flavors.py
View File

@ -15,18 +15,18 @@
# License for the specific language governing permissions and limitations
# under the License.
import unittest
import stubout
import webob
from nova import test
import nova.api
from nova.api.openstack import flavors
from nova.tests.api.openstack import fakes
class FlavorsTest(unittest.TestCase):
class FlavorsTest(test.TestCase):
def setUp(self):
super(FlavorsTest, self).setUp()
self.stubs = stubout.StubOutForTesting()
fakes.FakeAuthManager.auth_data = {}
fakes.FakeAuthDatabase.data = {}
@ -36,6 +36,7 @@ class FlavorsTest(unittest.TestCase):
def tearDown(self):
self.stubs.UnsetAll()
super(FlavorsTest, self).tearDown()
def test_get_flavor_list(self):
req = webob.Request.blank('/v1.0/flavors')
@ -43,6 +44,3 @@ class FlavorsTest(unittest.TestCase):
def test_get_flavor_by_id(self):
pass
if __name__ == '__main__':
unittest.main()

14
nova/tests/api/openstack/test_images.py
View File

@ -22,7 +22,6 @@ and as a WSGI layer
import json
import datetime
import unittest
import stubout
import webob
@ -30,6 +29,7 @@ import webob
from nova import context
from nova import exception
from nova import flags
from nova import test
from nova import utils
import nova.api.openstack
from nova.api.openstack import images
@ -130,12 +130,13 @@ class BaseImageServiceTests(object):
self.assertEquals(1, num_images)
class LocalImageServiceTest(unittest.TestCase,
class LocalImageServiceTest(test.TestCase,
BaseImageServiceTests):
"""Tests the local image service"""
def setUp(self):
super(LocalImageServiceTest, self).setUp()
self.stubs = stubout.StubOutForTesting()
service_class = 'nova.image.local.LocalImageService'
self.service = utils.import_object(service_class)
@ -145,14 +146,16 @@ class LocalImageServiceTest(unittest.TestCase,
self.service.delete_all()
self.service.delete_imagedir()
self.stubs.UnsetAll()
super(LocalImageServiceTest, self).tearDown()
class GlanceImageServiceTest(unittest.TestCase,
class GlanceImageServiceTest(test.TestCase,
BaseImageServiceTests):
"""Tests the local image service"""
def setUp(self):
super(GlanceImageServiceTest, self).setUp()
self.stubs = stubout.StubOutForTesting()
fakes.stub_out_glance(self.stubs)
fakes.stub_out_compute_api_snapshot(self.stubs)
@ -163,9 +166,10 @@ class GlanceImageServiceTest(unittest.TestCase,
def tearDown(self):
self.stubs.UnsetAll()
super(GlanceImageServiceTest, self).tearDown()
class ImageControllerWithGlanceServiceTest(unittest.TestCase):
class ImageControllerWithGlanceServiceTest(test.TestCase):
"""Test of the OpenStack API /images application controller"""
@ -194,6 +198,7 @@ class ImageControllerWithGlanceServiceTest(unittest.TestCase):
'image_type': 'ramdisk'}]
def setUp(self):
super(ImageControllerWithGlanceServiceTest, self).setUp()
self.orig_image_service = FLAGS.image_service
FLAGS.image_service = 'nova.image.glance.GlanceImageService'
self.stubs = stubout.StubOutForTesting()
@ -208,6 +213,7 @@ class ImageControllerWithGlanceServiceTest(unittest.TestCase):
def tearDown(self):
self.stubs.UnsetAll()
FLAGS.image_service = self.orig_image_service
super(ImageControllerWithGlanceServiceTest, self).tearDown()
def test_get_image_index(self):
req = webob.Request.blank('/v1.0/images')

15
nova/tests/api/openstack/test_ratelimiting.py
View File

@ -1,15 +1,16 @@
import httplib
import StringIO
import time
import unittest
import webob
from nova import test
import nova.api.openstack.ratelimiting as ratelimiting
class LimiterTest(unittest.TestCase):
class LimiterTest(test.TestCase):
def setUp(self):
super(LimiterTest, self).setUp()
self.limits = {
'a': (5, ratelimiting.PER_SECOND),
'b': (5, ratelimiting.PER_MINUTE),
@ -83,9 +84,10 @@ class FakeLimiter(object):
return self._delay
class WSGIAppTest(unittest.TestCase):
class WSGIAppTest(test.TestCase):
def setUp(self):
super(WSGIAppTest, self).setUp()
self.limiter = FakeLimiter(self)
self.app = ratelimiting.WSGIApp(self.limiter)
@ -206,7 +208,7 @@ def wire_HTTPConnection_to_WSGI(host, app):
httplib.HTTPConnection = HTTPConnectionDecorator(httplib.HTTPConnection)
class WSGIAppProxyTest(unittest.TestCase):
class WSGIAppProxyTest(test.TestCase):
def setUp(self):
"""Our WSGIAppProxy is going to call across an HTTPConnection to a
@ -218,6 +220,7 @@ class WSGIAppProxyTest(unittest.TestCase):
at the WSGIApp. And the limiter isn't real -- it's a fake that
behaves the way we tell it to.
"""
super(WSGIAppProxyTest, self).setUp()
self.limiter = FakeLimiter(self)
app = ratelimiting.WSGIApp(self.limiter)
wire_HTTPConnection_to_WSGI('100.100.100.100:80', app)
@ -238,7 +241,3 @@ class WSGIAppProxyTest(unittest.TestCase):
self.limiter.mock('murder', 'brutus', None)
self.proxy.perform('stab', 'brutus')
self.assertRaises(AssertionError, shouldRaise)
if __name__ == '__main__':
unittest.main()

10
nova/tests/api/openstack/test_servers.py
View File

@ -17,13 +17,13 @@
import datetime
import json
import unittest
import stubout
import webob
from nova import db
from nova import flags
from nova import test
import nova.api.openstack
from nova.api.openstack import servers
import nova.db.api
@ -113,9 +113,10 @@ def fake_compute_api(cls, req, id):
return True
class ServersTest(unittest.TestCase):
class ServersTest(test.TestCase):
def setUp(self):
super(ServersTest, self).setUp()
self.stubs = stubout.StubOutForTesting()
fakes.FakeAuthManager.auth_data = {}
fakes.FakeAuthDatabase.data = {}
@ -146,6 +147,7 @@ class ServersTest(unittest.TestCase):
def tearDown(self):
self.stubs.UnsetAll()
FLAGS.allow_admin_api = self.allow_admin
super(ServersTest, self).tearDown()
def test_get_server_by_id(self):
req = webob.Request.blank('/v1.0/servers/1')
@ -429,7 +431,3 @@ class ServersTest(unittest.TestCase):
res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status, '202 Accepted')
self.assertEqual(self.server_delete_called, True)
if __name__ == "__main__":
unittest.main()

7
nova/tests/api/openstack/test_shared_ip_groups.py
View File

@ -15,19 +15,20 @@
# License for the specific language governing permissions and limitations
# under the License.
import unittest
import stubout
from nova import test
from nova.api.openstack import shared_ip_groups
class SharedIpGroupsTest(unittest.TestCase):
class SharedIpGroupsTest(test.TestCase):
def setUp(self):
super(SharedIpGroupsTest, self).setUp()
self.stubs = stubout.StubOutForTesting()
def tearDown(self):
self.stubs.UnsetAll()
super(SharedIpGroupsTest, self).tearDown()
def test_get_shared_ip_groups(self):
pass

10
nova/tests/api/openstack/test_zones.py
View File

@ -13,7 +13,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import unittest
import stubout
import webob
@ -22,6 +21,7 @@ import json
import nova.db
from nova import context
from nova import flags
from nova import test
from nova.api.openstack import zones
from nova.tests.api.openstack import fakes
@ -60,8 +60,9 @@ def zone_get_all(context):
password='qwerty')]
class ZonesTest(unittest.TestCase):
class ZonesTest(test.TestCase):
def setUp(self):
super(ZonesTest, self).setUp()
self.stubs = stubout.StubOutForTesting()
fakes.FakeAuthManager.auth_data = {}
fakes.FakeAuthDatabase.data = {}
@ -81,6 +82,7 @@ class ZonesTest(unittest.TestCase):
def tearDown(self):
self.stubs.UnsetAll()
FLAGS.allow_admin_api = self.allow_admin
super(ZonesTest, self).tearDown()
def test_get_zone_list(self):
req = webob.Request.blank('/v1.0/zones')
@ -134,7 +136,3 @@ class ZonesTest(unittest.TestCase):
self.assertEqual(res_dict['zone']['id'], 1)
self.assertEqual(res_dict['zone']['api_url'], 'http://foo.com')
self.assertFalse('username' in res_dict['zone'])
if __name__ == '__main__':
unittest.main()

6
nova/tests/api/test_wsgi.py
View File

@ -21,7 +21,7 @@
Test WSGI basics and provide some helper functions for other WSGI tests.
"""
import unittest
from nova import test
import routes
import webob
@ -29,7 +29,7 @@ import webob
from nova import wsgi
class Test(unittest.TestCase):
class Test(test.TestCase):
def test_debug(self):
@ -92,7 +92,7 @@ class Test(unittest.TestCase):
self.assertNotEqual(result.body, "123")
class SerializerTest(unittest.TestCase):
class SerializerTest(test.TestCase):
def match(self, url, accept, expect):
input_dict = dict(servers=dict(a=(2, 3)))

6
nova/tests/fake_flags.py
View File

@ -29,8 +29,8 @@ FLAGS.auth_driver = 'nova.auth.dbdriver.DbDriver'
flags.DECLARE('network_size', 'nova.network.manager')
flags.DECLARE('num_networks', 'nova.network.manager')
flags.DECLARE('fake_network', 'nova.network.manager')
FLAGS.network_size = 16
FLAGS.num_networks = 5
FLAGS.network_size = 8
FLAGS.num_networks = 2
FLAGS.fake_network = True
flags.DECLARE('num_shelves', 'nova.volume.driver')
flags.DECLARE('blades_per_shelf', 'nova.volume.driver')
@ -39,5 +39,5 @@ FLAGS.num_shelves = 2
FLAGS.blades_per_shelf = 4
FLAGS.iscsi_num_targets = 8
FLAGS.verbose = True
FLAGS.sql_connection = 'sqlite:///tests.sqlite'
FLAGS.sqlite_db = "tests.sqlite"
FLAGS.use_ipv6 = True

1
nova/tests/objectstore_unittest.py
View File

@ -311,4 +311,5 @@ class S3APITestCase(test.TestCase):
self.auth_manager.delete_user('admin')
self.auth_manager.delete_project('admin')
stop_listening = defer.maybeDeferred(self.listening_port.stopListening)
super(S3APITestCase, self).tearDown()
return defer.DeferredList([stop_listening])

1
nova/tests/test_direct.py
View File

@ -52,6 +52,7 @@ class DirectTestCase(test.TestCase):
def tearDown(self):
direct.ROUTES = {}
super(DirectTestCase, self).tearDown()
def test_delegated_auth(self):
req = webob.Request.blank('/fake/context')

8
nova/tests/test_network.py
View File

@ -42,15 +42,13 @@ class NetworkTestCase(test.TestCase):
# flags in the corresponding section in nova-dhcpbridge
self.flags(connection_type='fake',
fake_call=True,
fake_network=True,
network_size=16,
num_networks=5)
fake_network=True)
self.manager = manager.AuthManager()
self.user = self.manager.create_user('netuser', 'netuser', 'netuser')
self.projects = []
self.network = utils.import_object(FLAGS.network_manager)
self.context = context.RequestContext(project=None, user=self.user)
for i in range(5):
for i in range(FLAGS.num_networks):
name = 'project%s' % i
project = self.manager.create_project(name, 'netuser', name)
self.projects.append(project)
@ -195,7 +193,7 @@ class NetworkTestCase(test.TestCase):
first = self._create_address(0)
lease_ip(first)
instance_ids = []
for i in range(1, 5):
for i in range(1, FLAGS.num_networks):
instance_ref = self._create_instance(i, mac=utils.generate_mac())
instance_ids.append(instance_ref['id'])
address = self._create_address(i, instance_ref['id'])

3
nova/tests/test_virt.py
View File

@ -207,9 +207,9 @@ class LibvirtConnTestCase(test.TestCase):
db.instance_destroy(user_context, instance_ref['id'])
def tearDown(self):
super(LibvirtConnTestCase, self).tearDown()
self.manager.delete_project(self.project)
self.manager.delete_user(self.user)
super(LibvirtConnTestCase, self).tearDown()
class IptablesFirewallTestCase(test.TestCase):
@ -390,6 +390,7 @@ class NWFilterTestCase(test.TestCase):
def tearDown(self):
self.manager.delete_project(self.project)
self.manager.delete_user(self.user)
super(NWFilterTestCase, self).tearDown()
def test_cidr_rule_nwfilter_xml(self):
cloud_controller = cloud.CloudController()

7
run_tests.py
View File

@ -61,8 +61,8 @@ import unittest
import sys
from nose import config
from nose import result
from nose import core
from nose import result
from nova import log as logging
from nova.tests import fake_flags
@ -280,10 +280,6 @@ class NovaTestRunner(core.TextTestRunner):
if __name__ == '__main__':
logging.setup()
testdir = os.path.abspath(os.path.join("nova", "tests"))
testdb = os.path.join(testdir, "tests.sqlite")
if os.path.exists(testdb):
os.unlink(testdb)
# If any argument looks like a test name but doesn't have "nova.tests" in
# front of it, automatically add that so we don't have to type as much
argv = []
@ -293,6 +289,7 @@ if __name__ == '__main__':
else:
argv.append(x)
testdir = os.path.abspath(os.path.join("nova", "tests"))
c = config.Config(stream=sys.stdout,
env=os.environ,
verbosity=3,