x/python-ganttclient
Code Issues Proposed changes

merged trunk rev1348

Browse Source
This commit is contained in:
Kei Masumoto
2011-08-02 18:43:30 +09:00
parent cdb97ae9a2 e68ab42daf
commit 084ad1d9fa
29 changed files with 665 additions and 643 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
Authors
View File

@@ -57,6 +57,7 @@ Joshua McKenty <jmckenty@gmail.com>
Justin Santa Barbara <justin@fathomdb.com>
Justin Shepherd <jshepher@rackspace.com>
Kei Masumoto <masumotok@nttdata.co.jp>
masumoto<masumotok@nttdata.co.jp>
Ken Pepple <ken.pepple@gmail.com>
Kevin Bringard <kbringard@attinteractive.com>
Kevin L. Mitchell <kevin.mitchell@rackspace.com>
@@ -67,6 +68,7 @@ Lvov Maxim <usrleon@gmail.com>
Mark Washenberger <mark.washenberger@rackspace.com>
Masanori Itoh <itoumsn@nttdata.co.jp>
Matt Dietz <matt.dietz@rackspace.com>
Matthew Hooker <matt@cloudscaling.com>
Michael Gundlach <michael.gundlach@rackspace.com>
Mike Scherbakov <mihgen@gmail.com>
Mohammed Naser <mnaser@vexxhost.com>
@@ -105,3 +107,4 @@ Yoshiaki Tamura <yoshi@midokura.jp>
Youcef Laribi <Youcef.Laribi@eu.citrix.com>
Yuriy Taraday <yorik.sar@gmail.com>
Zhixue Wu <Zhixue.Wu@citrix.com>
Zed Shaw <zedshaw@zedshaw.com>

118
bin/nova-manage
View File

@@ -631,40 +631,75 @@ class NetworkCommands(object):
"""Class for managing networks."""
@args('--label', dest="label", metavar='<label>',
help='Label(ex: public)')
@args('--network', dest="fixed_range", metavar='<x.x.x.x/yy>',
help='Network')
help='Label for network (ex: public)')
@args('--fixed_range_v4', dest="fixed_range_v4", metavar='<x.x.x.x/yy>',
help='IPv4 subnet (ex: 10.0.0.0/8)')
@args('--num_networks', dest="num_networks", metavar='<number>',
help='How many networks create')
help='Number of networks to create')
@args('--network_size', dest="network_size", metavar='<number>',
help='How many hosts in network')
help='Number of IPs per network')
@args('--vlan', dest="vlan_start", metavar='<vlan id>', help='vlan id')
@args('--vpn', dest="vpn_start", help='vpn start')
@args('--fixed_range_v6', dest="fixed_range_v6", help='fixed ipv6 range')
@args('--fixed_range_v6', dest="fixed_range_v6",
help='IPv6 subnet (ex: fe80::/64')
@args('--gateway_v6', dest="gateway_v6", help='ipv6 gateway')
@args('--flat_network_bridge', dest="flat_network_bridge",
metavar='<flat network bridge>', help='Flat_network_bridge')
@args('--bridge', dest="bridge",
metavar='<bridge>',
help='VIFs on this network are connected to this bridge')
@args('--bridge_interface', dest="bridge_interface",
metavar='<bridge interface>', help='Bridge_interface')
metavar='<bridge interface>',
help='the bridge is connected to this interface')
@args('--multi_host', dest="multi_host", metavar="<'T'|'F'>",
help='Multi host')
@args('--dns1', dest="dns1", metavar="<DNS Address>", help='First DNS')
@args('--dns2', dest="dns2", metavar="<DNS Address>", help='Second DNS')
def create(self, label=None, fixed_range=None, num_networks=None,
def create(self, label=None, fixed_range_v4=None, num_networks=None,
network_size=None, multi_host=None, vlan_start=None,
vpn_start=None, fixed_range_v6=None, gateway_v6=None,
flat_network_bridge=None, bridge_interface=None,
dns1=None, dns2=None):
bridge=None, bridge_interface=None, dns1=None, dns2=None):
"""Creates fixed ips for host by range"""
# check for certain required inputs
if not label:
msg = _('a label (ex: public) is required to create networks.')
print msg
raise TypeError(msg)
if not fixed_range:
msg = _('Fixed range in the form of 10.0.0.0/8 is '
'required to create networks.')
print msg
raise TypeError(msg)
raise exception.NetworkNotCreated(req='--label')
if not fixed_range_v4:
raise exception.NetworkNotCreated(req='--fixed_range_v4')
bridge = bridge or FLAGS.flat_network_bridge
if not bridge:
bridge_required = ['nova.network.manager.FlatManager',
'nova.network.manager.FlatDHCPManager']
if FLAGS.network_manager in bridge_required:
# TODO(tr3buchet) - swap print statement and following line for
# raise statement in diablo 4
print _('--bridge parameter required or FLAG '
'flat_network_bridge must be set to create networks\n'
'WARNING! ACHTUNG! Setting the bridge to br100 '
'automatically is deprecated and will be removed in '
'Diablo milestone 4. Prepare yourself accordingly.')
time.sleep(10)
bridge = 'br100'
#raise exception.NetworkNotCreated(req='--bridge')
bridge_interface = bridge_interface or FLAGS.flat_interface or \
FLAGS.vlan_interface
if not bridge_interface:
interface_required = ['nova.network.manager.FlatDHCPManager',
'nova.network.manager.VlanManager']
if FLAGS.network_manager in interface_required:
raise exception.NetworkNotCreated(req='--bridge_interface')
if FLAGS.use_ipv6:
fixed_range_v6 = fixed_range_v6 or FLAGS.fixed_range_v6
if not fixed_range_v6:
raise exception.NetworkNotCreated(req='with use_ipv6, '
'--fixed_range_v6')
gateway_v6 = gateway_v6 or FLAGS.gateway_v6
if not gateway_v6:
raise exception.NetworkNotCreated(req='with use_ipv6, '
'--gateway_v6')
# sanitize other input using FLAGS if necessary
if not num_networks:
num_networks = FLAGS.num_networks
if not network_size:
@@ -677,36 +712,25 @@ class NetworkCommands(object):
vlan_start = FLAGS.vlan_start
if not vpn_start:
vpn_start = FLAGS.vpn_start
if not fixed_range_v6:
fixed_range_v6 = FLAGS.fixed_range_v6
if not flat_network_bridge:
flat_network_bridge = FLAGS.flat_network_bridge
if not bridge_interface:
bridge_interface = FLAGS.flat_interface or FLAGS.vlan_interface
if not gateway_v6:
gateway_v6 = FLAGS.gateway_v6
if not dns1 and FLAGS.flat_network_dns:
dns1 = FLAGS.flat_network_dns
net_manager = utils.import_object(FLAGS.network_manager)
try:
net_manager.create_networks(context.get_admin_context(),
label=label,
cidr=fixed_range,
multi_host=multi_host,
num_networks=int(num_networks),
network_size=int(network_size),
vlan_start=int(vlan_start),
vpn_start=int(vpn_start),
cidr_v6=fixed_range_v6,
gateway_v6=gateway_v6,
bridge=flat_network_bridge,
bridge_interface=bridge_interface,
dns1=dns1,
dns2=dns2)
except ValueError, e:
print e
raise e
# create the network
net_manager = utils.import_object(FLAGS.network_manager)
net_manager.create_networks(context.get_admin_context(),
label=label,
cidr=fixed_range_v4,
multi_host=multi_host,
num_networks=int(num_networks),
network_size=int(network_size),
vlan_start=int(vlan_start),
vpn_start=int(vpn_start),
cidr_v6=fixed_range_v6,
gateway_v6=gateway_v6,
bridge=bridge,
bridge_interface=bridge_interface,
dns1=dns1,
dns2=dns2)
def list(self):
"""List all created networks"""

2
nova/auth/dbdriver.py
View File

@@ -127,7 +127,7 @@ class DbDriver(object):
try:
project = db.project_create(context.get_admin_context(), values)
except exception.Duplicate:
except exception.DBError:
raise exception.ProjectExists(project=name)
for member in members:

14
nova/auth/manager.py
View File

@@ -518,6 +518,14 @@ class AuthManager(object):
return drv.get_user_roles(User.safe_id(user),
Project.safe_id(project))
def get_active_roles(self, user, project=None):
"""Get all active roles for context"""
if project:
roles = FLAGS.allowed_roles + ['projectmanager']
else:
roles = FLAGS.global_roles
return [role for role in roles if self.has_role(user, role, project)]
def get_project(self, pid):
"""Get project object by id"""
with self.driver() as drv:
@@ -730,10 +738,6 @@ class AuthManager(object):
with self.driver() as drv:
drv.modify_user(uid, access_key, secret_key, admin)
@staticmethod
def get_key_pairs(context):
return db.key_pair_get_all_by_user(context.elevated(), context.user_id)
def get_credentials(self, user, project=None, use_dmz=True):
"""Get credential zip for user in project"""
if not isinstance(user, User):
@@ -785,7 +789,7 @@ class AuthManager(object):
return read_buffer
def get_environment_rc(self, user, project=None, use_dmz=True):
"""Get credential zip for user in project"""
"""Get environment rc for user in project"""
if not isinstance(user, User):
user = self.get_user(user)
if project is None:

2
nova/flags.py
View File

@@ -343,7 +343,7 @@ DEFINE_string('lock_path', os.path.join(os.path.dirname(__file__), '../'),
'Directory for lock files')
DEFINE_string('logdir', None, 'output to a per-service log file in named '
'directory')
DEFINE_integer('logfile_mode', 0644, 'Default file mode of the logs.')
DEFINE_string('sqlite_db', 'nova.sqlite', 'file name for sqlite')
DEFINE_string('sql_connection',
'sqlite:///$state_path/$sqlite_db',

5
nova/log.py
View File

@@ -43,8 +43,8 @@ from nova import version
FLAGS = flags.FLAGS
flags.DEFINE_string('logging_context_format_string',
'%(asctime)s %(levelname)s %(name)s '
'[%(request_id)s %(user)s '
'%(project)s] %(message)s',
'[%(request_id)s %(user_id)s '
'%(project_id)s] %(message)s',
'format string to use for log messages with context')
flags.DEFINE_string('logging_default_format_string',
'%(asctime)s %(levelname)s %(name)s [-] '
@@ -257,6 +257,7 @@ class NovaRootLogger(NovaLogger):
self.filelog = WatchedFileHandler(logpath)
self.addHandler(self.filelog)
self.logpath = logpath
os.chmod(self.logpath, FLAGS.logfile_mode)
else:
self.removeHandler(self.filelog)
self.addHandler(self.streamlog)

66
nova/rpc/__init__.py Normal file
View File

@@ -0,0 +1,66 @@
# 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.
from nova.utils import import_object
from nova.rpc.common import RemoteError, LOG
from nova import flags
FLAGS = flags.FLAGS
flags.DEFINE_string('rpc_backend',
'nova.rpc.amqp',
"The messaging module to use, defaults to AMQP.")
RPCIMPL = import_object(FLAGS.rpc_backend)
def create_connection(new=True):
return RPCIMPL.Connection.instance(new=True)
def create_consumer(conn, topic, proxy, fanout=False):
if fanout:
return RPCIMPL.FanoutAdapterConsumer(
connection=conn,
topic=topic,
proxy=proxy)
else:
return RPCIMPL.TopicAdapterConsumer(
connection=conn,
topic=topic,
proxy=proxy)
def create_consumer_set(conn, consumers):
return RPCIMPL.ConsumerSet(connection=conn, consumer_list=consumers)
def call(context, topic, msg):
return RPCIMPL.call(context, topic, msg)
def cast(context, topic, msg):
return RPCIMPL.cast(context, topic, msg)
def fanout_cast(context, topic, msg):
return RPCIMPL.fanout_cast(context, topic, msg)
def multicall(context, topic, msg):
return RPCIMPL.multicall(context, topic, msg)

23
nova/rpc.py → nova/rpc/amqp.py
View File

@@ -44,9 +44,7 @@ from nova import fakerabbit
from nova import flags
from nova import log as logging
from nova import utils
LOG = logging.getLogger('nova.rpc')
from nova.rpc.common import RemoteError, LOG
FLAGS = flags.FLAGS
@@ -418,25 +416,6 @@ def msg_reply(msg_id, reply=None, failure=None):
publisher.close()
class RemoteError(exception.Error):
"""Signifies that a remote class has raised an exception.
Containes a string representation of the type of the original exception,
the value of the original exception, and the traceback. These are
sent to the parent as a joined string so printing the exception
contains all of the relevent info.
"""
def __init__(self, exc_type, value, traceback):
self.exc_type = exc_type
self.value = value
self.traceback = traceback
super(RemoteError, self).__init__('%s %s\n%s' % (exc_type,
value,
traceback))
def _unpack_context(msg):
"""Unpack context from msg."""
context_dict = {}

23
nova/rpc/common.py Normal file
View File

@@ -0,0 +1,23 @@
from nova import exception
from nova import log as logging
LOG = logging.getLogger('nova.rpc')
class RemoteError(exception.Error):
"""Signifies that a remote class has raised an exception.
Containes a string representation of the type of the original exception,
the value of the original exception, and the traceback. These are
sent to the parent as a joined string so printing the exception
contains all of the relevent info.
"""
def __init__(self, exc_type, value, traceback):
self.exc_type = exc_type
self.value = value
self.traceback = traceback
super(RemoteError, self).__init__('%s %s\n%s' % (exc_type,
value,
traceback))

1
nova/scheduler/least_cost.py
View File

@@ -28,6 +28,7 @@ from nova import flags
from nova import log as logging
from nova.scheduler import zone_aware_scheduler
from nova import utils
from nova import exception
LOG = logging.getLogger('nova.scheduler.least_cost')

2
nova/scheduler/zone_aware_scheduler.py
View File

@@ -81,7 +81,7 @@ class ZoneAwareScheduler(driver.Scheduler):
decryptor = crypto.decryptor(FLAGS.build_plan_encryption_key)
try:
json_entry = decryptor(blob)
return json.dumps(entry)
return json.dumps(json_entry)
except M2Crypto.EVP.EVPError:
pass
return None

1
nova/tests/fake_flags.py
View File

@@ -42,3 +42,4 @@ FLAGS['iscsi_num_targets'].SetDefault(8)
FLAGS['verbose'].SetDefault(True)
FLAGS['sqlite_db'].SetDefault("tests.sqlite")
FLAGS['use_ipv6'].SetDefault(True)
FLAGS['flat_network_bridge'].SetDefault('br100')

9
nova/tests/hyperv_unittest.py
View File

@@ -23,7 +23,6 @@ from nova import context
from nova import db
from nova import flags
from nova import test
from nova.auth import manager
from nova.virt import hyperv
FLAGS = flags.FLAGS
@@ -34,11 +33,9 @@ class HyperVTestCase(test.TestCase):
"""Test cases for the Hyper-V driver"""
def setUp(self):
super(HyperVTestCase, self).setUp()
self.manager = manager.AuthManager()
self.user = self.manager.create_user('fake', 'fake', 'fake',
admin=True)
self.project = self.manager.create_project('fake', 'fake', 'fake')
self.context = context.RequestContext(self.user, self.project)
self.user_id = 'fake'
self.project_id = 'fake'
self.context = context.RequestContext(self.user_id, self.project_id)
def test_create_destroy(self):
"""Create a VM and destroy it"""

18
nova/tests/scheduler/test_zone_aware_scheduler.py
View File

@@ -16,6 +16,8 @@
Tests For Zone Aware Scheduler.
"""
import json
import nova.db
from nova import exception
@@ -327,3 +329,19 @@ class ZoneAwareSchedulerTestCase(test.TestCase):
sched._provision_resource_from_blob(None, request_spec, 1,
request_spec, {})
self.assertTrue(was_called)
def test_decrypt_blob(self):
"""Test that the decrypt method works."""
fixture = FakeZoneAwareScheduler()
test_data = {"foo": "bar"}
class StubDecryptor(object):
def decryptor(self, key):
return lambda blob: blob
self.stubs.Set(zone_aware_scheduler, 'crypto',
StubDecryptor())
self.assertEqual(fixture._decrypt_blob(test_data),
json.dumps(test_data))

21
nova/tests/test_access.py
View File

@@ -16,7 +16,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import unittest
import webob
from nova import context
@@ -41,7 +40,7 @@ class FakeApiRequest(object):
class AccessTestCase(test.TestCase):
def _env_for(self, ctxt, action):
env = {}
env['ec2.context'] = ctxt
env['nova.context'] = ctxt
env['ec2.request'] = FakeApiRequest(action)
return env
@@ -93,7 +92,11 @@ class AccessTestCase(test.TestCase):
super(AccessTestCase, self).tearDown()
def response_status(self, user, methodName):
ctxt = context.RequestContext(user, self.project)
roles = manager.AuthManager().get_active_roles(user, self.project)
ctxt = context.RequestContext(user.id,
self.project.id,
is_admin=user.is_admin(),
roles=roles)
environ = self._env_for(ctxt, methodName)
req = webob.Request.blank('/', environ)
resp = req.get_response(self.mw)
@@ -105,30 +108,26 @@ class AccessTestCase(test.TestCase):
def shouldDeny(self, user, methodName):
self.assertEqual(401, self.response_status(user, methodName))
def test_001_allow_all(self):
def test_allow_all(self):
users = [self.testadmin, self.testpmsys, self.testnet, self.testsys]
for user in users:
self.shouldAllow(user, '_allow_all')
def test_002_allow_none(self):
def test_allow_none(self):
self.shouldAllow(self.testadmin, '_allow_none')
users = [self.testpmsys, self.testnet, self.testsys]
for user in users:
self.shouldDeny(user, '_allow_none')
def test_003_allow_project_manager(self):
def test_allow_project_manager(self):
for user in [self.testadmin, self.testpmsys]:
self.shouldAllow(user, '_allow_project_manager')
for user in [self.testnet, self.testsys]:
self.shouldDeny(user, '_allow_project_manager')
def test_004_allow_sys_and_net(self):
def test_allow_sys_and_net(self):
for user in [self.testadmin, self.testnet, self.testsys]:
self.shouldAllow(user, '_allow_sys_and_net')
# denied because it doesn't have the per project sysadmin
for user in [self.testpmsys]:
self.shouldDeny(user, '_allow_sys_and_net')
if __name__ == "__main__":
# TODO: Implement use_fake as an option
unittest.main()

18
nova/tests/test_adminapi.py
View File

@@ -25,7 +25,6 @@ from nova import log as logging
from nova import rpc
from nova import test
from nova import utils
from nova.auth import manager
from nova.api.ec2 import admin
from nova.image import fake
@@ -39,7 +38,7 @@ class AdminApiTestCase(test.TestCase):
super(AdminApiTestCase, self).setUp()
self.flags(connection_type='fake')
self.conn = rpc.Connection.instance()
self.conn = rpc.create_connection()
# set up our cloud
self.api = admin.AdminController()
@@ -51,11 +50,11 @@ class AdminApiTestCase(test.TestCase):
self.volume = self.start_service('volume')
self.image_service = utils.import_object(FLAGS.image_service)
self.manager = manager.AuthManager()
self.user = self.manager.create_user('admin', 'admin', 'admin', True)
self.project = self.manager.create_project('proj', 'admin', 'proj')
self.context = context.RequestContext(user=self.user,
project=self.project)
self.user_id = 'admin'
self.project_id = 'admin'
self.context = context.RequestContext(self.user_id,
self.project_id,
True)
def fake_show(meh, context, id):
return {'id': 1, 'properties': {'kernel_id': 1, 'ramdisk_id': 1,
@@ -73,11 +72,6 @@ class AdminApiTestCase(test.TestCase):
self.stubs.Set(rpc, 'cast', finish_cast)
def tearDown(self):
self.manager.delete_project(self.project)
self.manager.delete_user(self.user)
super(AdminApiTestCase, self).tearDown()
def test_block_external_ips(self):
"""Make sure provider firewall rules are created."""
result = self.api.block_external_addresses(self.context, '1.1.1.1/32')

82
nova/tests/test_api.py
View File

@@ -30,11 +30,11 @@ import webob
from nova import context
from nova import exception
from nova import test
from nova import wsgi
from nova.api import ec2
from nova.api.ec2 import apirequest
from nova.api.ec2 import cloud
from nova.api.ec2 import ec2utils
from nova.auth import manager
class FakeHttplibSocket(object):
@@ -192,10 +192,13 @@ 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 = ec2.Authenticate(ec2.Requestify(ec2.Executor(),
'nova.api.ec2.cloud.CloudController'))
# NOTE(vish): skipping the Authorizer
roles = ['sysadmin', 'netadmin']
ctxt = context.RequestContext('fake', 'fake', roles=roles)
self.app = wsgi.InjectContext(ctxt,
ec2.Requestify(ec2.Authorizer(ec2.Executor()),
'nova.api.ec2.cloud.CloudController'))
def expect_http(self, host=None, is_secure=False, api_version=None):
"""Returns a new EC2 connection"""
@@ -213,7 +216,11 @@ class ApiEc2TestCase(test.TestCase):
self.http = FakeHttplibConnection(
self.app, '%s:8773' % (self.host), False)
# pylint: disable=E1103
self.ec2.new_http_connection(host, is_secure).AndReturn(self.http)
if boto.Version >= '2':
self.ec2.new_http_connection(host or '%s:8773' % (self.host),
is_secure).AndReturn(self.http)
else:
self.ec2.new_http_connection(host, is_secure).AndReturn(self.http)
return self.http
def test_return_valid_isoformat(self):
@@ -242,39 +249,25 @@ class ApiEc2TestCase(test.TestCase):
self.expect_http(api_version='2010-10-30')
self.mox.ReplayAll()
user = self.manager.create_user('fake', 'fake', 'fake')
project = self.manager.create_project('fake', 'fake', 'fake')
# Any request should be fine
self.ec2.get_all_instances()
self.assertTrue(self.ec2.APIVersion in self.http.getresponsebody(),
'The version in the xmlns of the response does '
'not match the API version given in the request.')
self.manager.delete_project(project)
self.manager.delete_user(user)
def test_describe_instances(self):
"""Test that, after creating a user and a project, the describe
instances call to the API works properly"""
self.expect_http()
self.mox.ReplayAll()
user = self.manager.create_user('fake', 'fake', 'fake')
project = self.manager.create_project('fake', 'fake', 'fake')
self.assertEqual(self.ec2.get_all_instances(), [])
self.manager.delete_project(project)
self.manager.delete_user(user)
def test_terminate_invalid_instance(self):
"""Attempt to terminate an invalid instance"""
self.expect_http()
self.mox.ReplayAll()
user = self.manager.create_user('fake', 'fake', 'fake')
project = self.manager.create_project('fake', 'fake', 'fake')
self.assertRaises(EC2ResponseError, self.ec2.terminate_instances,
"i-00000005")
self.manager.delete_project(project)
self.manager.delete_user(user)
def test_get_all_key_pairs(self):
"""Test that, after creating a user and project and generating
@@ -283,16 +276,12 @@ class ApiEc2TestCase(test.TestCase):
self.mox.ReplayAll()
keyname = "".join(random.choice("sdiuisudfsdcnpaqwertasd") \
for x in range(random.randint(4, 8)))
user = self.manager.create_user('fake', 'fake', 'fake')
project = self.manager.create_project('fake', 'fake', 'fake')
# NOTE(vish): create depends on pool, so call helper directly
cloud._gen_key(context.get_admin_context(), user.id, keyname)
cloud._gen_key(context.get_admin_context(), 'fake', keyname)
rv = self.ec2.get_all_key_pairs()
results = [k for k in rv if k.name == keyname]
self.assertEquals(len(results), 1)
self.manager.delete_project(project)
self.manager.delete_user(user)
def test_create_duplicate_key_pair(self):
"""Test that, after successfully generating a keypair,
@@ -301,8 +290,6 @@ class ApiEc2TestCase(test.TestCase):
self.mox.ReplayAll()
keyname = "".join(random.choice("sdiuisudfsdcnpaqwertasd") \
for x in range(random.randint(4, 8)))
user = self.manager.create_user('fake', 'fake', 'fake')
project = self.manager.create_project('fake', 'fake', 'fake')
# NOTE(vish): create depends on pool, so call helper directly
self.ec2.create_key_pair('test')
@@ -321,27 +308,16 @@ class ApiEc2TestCase(test.TestCase):
"""Test that we can retrieve security groups"""
self.expect_http()
self.mox.ReplayAll()
user = self.manager.create_user('fake', 'fake', 'fake', admin=True)
project = self.manager.create_project('fake', 'fake', 'fake')
rv = self.ec2.get_all_security_groups()
self.assertEquals(len(rv), 1)
self.assertEquals(rv[0].name, 'default')
self.manager.delete_project(project)
self.manager.delete_user(user)
def test_create_delete_security_group(self):
"""Test that we can create a security group"""
self.expect_http()
self.mox.ReplayAll()
user = self.manager.create_user('fake', 'fake', 'fake', admin=True)
project = self.manager.create_project('fake', 'fake', 'fake')
# At the moment, you need both of these to actually be netadmin
self.manager.add_role('fake', 'netadmin')
project.add_role('fake', 'netadmin')
security_group_name = "".join(random.choice("sdiuisudfsdcnpaqwertasd")
for x in range(random.randint(4, 8)))
@@ -360,9 +336,6 @@ class ApiEc2TestCase(test.TestCase):
self.ec2.delete_security_group(security_group_name)
self.manager.delete_project(project)
self.manager.delete_user(user)
def test_authorize_revoke_security_group_cidr(self):
"""
Test that we can add and remove CIDR based rules
@@ -370,12 +343,6 @@ class ApiEc2TestCase(test.TestCase):
"""
self.expect_http()
self.mox.ReplayAll()
user = self.manager.create_user('fake', 'fake', 'fake')
project = self.manager.create_project('fake', 'fake', 'fake')
# At the moment, you need both of these to actually be netadmin
self.manager.add_role('fake', 'netadmin')
project.add_role('fake', 'netadmin')
security_group_name = "".join(random.choice("sdiuisudfsdcnpaqwertasd")
for x in range(random.randint(4, 8)))
@@ -422,9 +389,6 @@ class ApiEc2TestCase(test.TestCase):
self.assertEqual(len(rv), 1)
self.assertEqual(rv[0].name, 'default')
self.manager.delete_project(project)
self.manager.delete_user(user)
return
def test_authorize_revoke_security_group_cidr_v6(self):
@@ -434,12 +398,6 @@ class ApiEc2TestCase(test.TestCase):
"""
self.expect_http()
self.mox.ReplayAll()
user = self.manager.create_user('fake', 'fake', 'fake')
project = self.manager.create_project('fake', 'fake', 'fake')
# At the moment, you need both of these to actually be netadmin
self.manager.add_role('fake', 'netadmin')
project.add_role('fake', 'netadmin')
security_group_name = "".join(random.choice("sdiuisudfsdcnpaqwertasd")
for x in range(random.randint(4, 8)))
@@ -485,9 +443,6 @@ class ApiEc2TestCase(test.TestCase):
self.assertEqual(len(rv), 1)
self.assertEqual(rv[0].name, 'default')
self.manager.delete_project(project)
self.manager.delete_user(user)
return
def test_authorize_revoke_security_group_foreign_group(self):
@@ -497,12 +452,6 @@ class ApiEc2TestCase(test.TestCase):
"""
self.expect_http()
self.mox.ReplayAll()
user = self.manager.create_user('fake', 'fake', 'fake', admin=True)
project = self.manager.create_project('fake', 'fake', 'fake')
# At the moment, you need both of these to actually be netadmin
self.manager.add_role('fake', 'netadmin')
project.add_role('fake', 'netadmin')
rand_string = 'sdiuisudfsdcnpaqwertasd'
security_group_name = "".join(random.choice(rand_string)
@@ -556,8 +505,3 @@ class ApiEc2TestCase(test.TestCase):
self.mox.ReplayAll()
self.ec2.delete_security_group(security_group_name)
self.manager.delete_project(project)
self.manager.delete_user(user)
return

19
nova/tests/test_auth.py
View File

@@ -102,7 +102,7 @@ class _AuthManagerBaseTestCase(test.TestCase):
self.assertEqual('classified', u.secret)
self.assertEqual('private-party', u.access)
def test_004_signature_is_valid(self):
def test_signature_is_valid(self):
with user_generator(self.manager, name='admin', secret='admin',
access='admin'):
with project_generator(self.manager, name="admin",
@@ -141,15 +141,14 @@ class _AuthManagerBaseTestCase(test.TestCase):
'127.0.0.1',
'/services/Cloud'))
def test_005_can_get_credentials(self):
return
credentials = self.manager.get_user('test1').get_credentials()
self.assertEqual(credentials,
'export EC2_ACCESS_KEY="access"\n' +
'export EC2_SECRET_KEY="secret"\n' +
'export EC2_URL="http://127.0.0.1:8773/services/Cloud"\n' +
'export S3_URL="http://127.0.0.1:3333/"\n' +
'export EC2_USER_ID="test1"\n')
def test_can_get_credentials(self):
st = {'access': 'access', 'secret': 'secret'}
with user_and_project_generator(self.manager, user_state=st) as (u, p):
credentials = self.manager.get_environment_rc(u, p)
LOG.debug(credentials)
self.assertTrue('export EC2_ACCESS_KEY="access:testproj"\n'
in credentials)
self.assertTrue('export EC2_SECRET_KEY="secret"\n' in credentials)
def test_can_list_users(self):
with user_generator(self.manager):

107
nova/tests/test_cloud.py
View File

@@ -34,7 +34,6 @@ from nova import network
from nova import rpc
from nova import test
from nova import utils
from nova.auth import manager
from nova.api.ec2 import cloud
from nova.api.ec2 import ec2utils
from nova.image import fake
@@ -50,7 +49,7 @@ class CloudTestCase(test.TestCase):
self.flags(connection_type='fake',
stub_network=True)
self.conn = rpc.Connection.instance()
self.conn = rpc.create_connection()
# set up our cloud
self.cloud = cloud.CloudController()
@@ -62,12 +61,11 @@ class CloudTestCase(test.TestCase):
self.volume = self.start_service('volume')
self.image_service = utils.import_object(FLAGS.image_service)
self.manager = manager.AuthManager()
self.user = self.manager.create_user('admin', 'admin', 'admin', True)
self.project = self.manager.create_project('proj', 'admin', 'proj')
self.context = context.RequestContext(user=self.user,
project=self.project)
host = self.network.host
self.user_id = 'fake'
self.project_id = 'fake'
self.context = context.RequestContext(self.user_id,
self.project_id,
True)
def fake_show(meh, context, id):
return {'id': 1, 'container_format': 'ami',
@@ -87,17 +85,15 @@ class CloudTestCase(test.TestCase):
self.stubs.Set(rpc, 'cast', finish_cast)
def tearDown(self):
networks = db.project_get_networks(self.context, self.project.id,
networks = db.project_get_networks(self.context, self.project_id,
associate=False)
for network in networks:
db.network_disassociate(self.context, network['id'])
self.manager.delete_project(self.project)
self.manager.delete_user(self.user)
super(CloudTestCase, self).tearDown()
def _create_key(self, name):
# NOTE(vish): create depends on pool, so just call helper directly
return cloud._gen_key(self.context, self.context.user.id, name)
return cloud._gen_key(self.context, self.context.user_id, name)
def test_describe_regions(self):
"""Makes sure describe regions runs without raising an exception"""
@@ -121,7 +117,6 @@ class CloudTestCase(test.TestCase):
public_ip=address)
db.floating_ip_destroy(self.context, address)
@test.skip_test("Skipping this pending future merge")
def test_allocate_address(self):
address = "10.10.10.10"
allocate = self.cloud.allocate_address
@@ -161,13 +156,10 @@ class CloudTestCase(test.TestCase):
# ApiError: Floating ip is in use. Disassociate it before releasing.
self.assertRaises(exception.ApiError, release, self.context, address)
@test.skip_test("Skipping this pending future merge")
def test_associate_disassociate_address(self):
"""Verifies associate runs cleanly without raising an exception"""
address = "10.10.10.10"
db.floating_ip_create(self.context,
{'address': address,
'host': self.network.host})
db.floating_ip_create(self.context, {'address': address})
self.cloud.allocate_address(self.context)
# TODO(jkoelker) Probably need to query for instance_type_id and
# make sure we get a valid one
@@ -175,11 +167,14 @@ class CloudTestCase(test.TestCase):
'instance_type_id': 1})
networks = db.network_get_all(self.context)
for network in networks:
self.network.set_network_host(self.context, network['id'])
db.network_update(self.context, network['id'],
{'host': self.network.host})
project_id = self.context.project_id
type_id = inst['instance_type_id']
ips = self.network.allocate_for_instance(self.context,
instance_id=inst['id'],
host=inst['host'],
vpn=None,
instance_type_id=type_id,
project_id=project_id)
# TODO(jkoelker) Make this mas bueno
@@ -287,13 +282,30 @@ class CloudTestCase(test.TestCase):
'ip_protocol': u'tcp'}]}
self.assertTrue(authz(self.context, group_name=sec['name'], **kwargs))
def test_authorize_security_group_ingress_ip_permissions_groups(self):
def test_authorize_security_group_fail_missing_source_group(self):
kwargs = {'project_id': self.context.project_id, 'name': 'test'}
sec = db.security_group_create(self.context, kwargs)
authz = self.cloud.authorize_security_group_ingress
kwargs = {'ip_permissions': [{'to_port': 81, 'from_port': 81,
'ip_ranges':{'1': {'cidr_ip': u'0.0.0.0/0'},
'2': {'cidr_ip': u'10.10.10.10/32'}},
'groups': {'1': {'user_id': u'someuser',
'group_name': u'somegroup1'}},
'ip_protocol': u'tcp'}]}
self.assertRaises(exception.SecurityGroupNotFound, authz,
self.context, group_name=sec['name'], **kwargs)
def test_authorize_security_group_ingress_ip_permissions_groups(self):
kwargs = {'project_id': self.context.project_id, 'name': 'test'}
sec = db.security_group_create(self.context,
{'project_id': 'someuser',
'name': 'somegroup1'})
sec = db.security_group_create(self.context,
{'project_id': 'someuser',
'name': 'othergroup2'})
sec = db.security_group_create(self.context, kwargs)
authz = self.cloud.authorize_security_group_ingress
kwargs = {'ip_permissions': [{'to_port': 81, 'from_port': 81,
'groups': {'1': {'user_id': u'someuser',
'group_name': u'somegroup1'},
'2': {'user_id': u'someuser',
@@ -310,22 +322,15 @@ class CloudTestCase(test.TestCase):
revoke = self.cloud.revoke_security_group_ingress
self.assertTrue(revoke(self.context, group_name=sec['name'], **kwargs))
def test_revoke_security_group_ingress_by_id(self):
kwargs = {'project_id': self.context.project_id, 'name': 'test'}
sec = db.security_group_create(self.context, kwargs)
authz = self.cloud.authorize_security_group_ingress
kwargs = {'to_port': '999', 'from_port': '999', 'ip_protocol': 'tcp'}
authz(self.context, group_id=sec['id'], **kwargs)
revoke = self.cloud.revoke_security_group_ingress
self.assertTrue(revoke(self.context, group_id=sec['id'], **kwargs))
def test_authorize_security_group_ingress_by_id(self):
def test_authorize_revoke_security_group_ingress_by_id(self):
sec = db.security_group_create(self.context,
{'project_id': self.context.project_id,
'name': 'test'})
authz = self.cloud.authorize_security_group_ingress
kwargs = {'to_port': '999', 'from_port': '999', 'ip_protocol': 'tcp'}
self.assertTrue(authz(self.context, group_id=sec['id'], **kwargs))
authz(self.context, group_id=sec['id'], **kwargs)
revoke = self.cloud.revoke_security_group_ingress
self.assertTrue(revoke(self.context, group_id=sec['id'], **kwargs))
def test_authorize_security_group_ingress_missing_protocol_params(self):
sec = db.security_group_create(self.context,
@@ -405,8 +410,6 @@ class CloudTestCase(test.TestCase):
db.service_destroy(self.context, service1['id'])
db.service_destroy(self.context, service2['id'])
# NOTE(jkoelker): this test relies on fixed_ip being in instances
@test.skip_test("EC2 stuff needs fixed_ip in instance_ref")
def test_describe_snapshots(self):
"""Makes sure describe_snapshots works and filters results."""
vol = db.volume_create(self.context, {})
@@ -947,21 +950,6 @@ class CloudTestCase(test.TestCase):
self._wait_for_running(ec2_instance_id)
return ec2_instance_id
def test_rescue_unrescue_instance(self):
instance_id = self._run_instance(
image_id='ami-1',
instance_type=FLAGS.default_instance_type,
max_count=1)
self.cloud.rescue_instance(context=self.context,
instance_id=instance_id)
# NOTE(vish): This currently does no validation, it simply makes sure
# that the code path doesn't throw an exception.
self.cloud.unrescue_instance(context=self.context,
instance_id=instance_id)
# TODO(soren): We need this until we can stop polling in the rpc code
# for unit tests.
self.cloud.terminate_instances(self.context, [instance_id])
def test_console_output(self):
instance_id = self._run_instance(
image_id='ami-1',
@@ -990,7 +978,7 @@ class CloudTestCase(test.TestCase):
key = RSA.load_key_string(private_key, callback=lambda: None)
bio = BIO.MemoryBuffer()
public_key = db.key_pair_get(self.context,
self.context.user.id,
self.context.user_id,
'test')['public_key']
key.save_pub_key_bio(bio)
converted = crypto.ssl_pub_to_ssh_pub(bio.read())
@@ -1014,7 +1002,7 @@ class CloudTestCase(test.TestCase):
'mytestfprint')
self.assertTrue(result1)
keydata = db.key_pair_get(self.context,
self.context.user.id,
self.context.user_id,
'testimportkey1')
self.assertEqual('mytestpubkey', keydata['public_key'])
self.assertEqual('mytestfprint', keydata['fingerprint'])
@@ -1031,7 +1019,7 @@ class CloudTestCase(test.TestCase):
dummypub)
self.assertTrue(result2)
keydata = db.key_pair_get(self.context,
self.context.user.id,
self.context.user_id,
'testimportkey2')
self.assertEqual(dummypub, keydata['public_key'])
self.assertEqual(dummyfprint, keydata['fingerprint'])
@@ -1041,12 +1029,6 @@ class CloudTestCase(test.TestCase):
self.cloud.delete_key_pair(self.context, 'test')
def test_run_instances(self):
# stub out the rpc call
def stub_cast(*args, **kwargs):
pass
self.stubs.Set(rpc, 'cast', stub_cast)
kwargs = {'image_id': FLAGS.default_image,
'instance_type': FLAGS.default_instance_type,
'max_count': 1}
@@ -1056,7 +1038,7 @@ class CloudTestCase(test.TestCase):
self.assertEqual(instance['imageId'], 'ami-00000001')
self.assertEqual(instance['displayName'], 'Server 1')
self.assertEqual(instance['instanceId'], 'i-00000001')
self.assertEqual(instance['instanceState']['name'], 'scheduling')
self.assertEqual(instance['instanceState']['name'], 'running')
self.assertEqual(instance['instanceType'], 'm1.small')
def test_run_instances_image_state_none(self):
@@ -1128,16 +1110,15 @@ class CloudTestCase(test.TestCase):
self.assertEqual('c00l 1m4g3', inst['display_name'])
db.instance_destroy(self.context, inst['id'])
# NOTE(jkoelker): This test relies on mac_address in instance
@test.skip_test("EC2 stuff needs mac_address in instance_ref")
def test_update_of_instance_wont_update_private_fields(self):
inst = db.instance_create(self.context, {})
host = inst['host']
ec2_id = ec2utils.id_to_ec2_id(inst['id'])
self.cloud.update_instance(self.context, ec2_id,
display_name='c00l 1m4g3',
mac_address='DE:AD:BE:EF')
host='otherhost')
inst = db.instance_get(self.context, inst['id'])
self.assertEqual(None, inst['mac_address'])
self.assertEqual(host, inst['host'])
db.instance_destroy(self.context, inst['id'])
def test_update_of_volume_display_fields(self):
@@ -1193,7 +1174,6 @@ class CloudTestCase(test.TestCase):
elevated = self.context.elevated(read_deleted=True)
self._wait_for_state(elevated, instance_id, is_deleted)
@test.skip_test("skipping, test is hanging with multinic for rpc reasons")
def test_stop_start_instance(self):
"""Makes sure stop/start instance works"""
# enforce periodic tasks run in short time to avoid wait for 60s.
@@ -1251,7 +1231,6 @@ class CloudTestCase(test.TestCase):
self.assertEqual(vol['status'], "available")
self.assertEqual(vol['attach_status'], "detached")
@test.skip_test("skipping, test is hanging with multinic for rpc reasons")
def test_stop_start_with_volume(self):
"""Make sure run instance with block device mapping works"""
@@ -1320,7 +1299,6 @@ class CloudTestCase(test.TestCase):
self._restart_compute_service()
@test.skip_test("skipping, test is hanging with multinic for rpc reasons")
def test_stop_with_attached_volume(self):
"""Make sure attach info is reflected to block device mapping"""
# enforce periodic tasks run in short time to avoid wait for 60s.
@@ -1396,7 +1374,6 @@ class CloudTestCase(test.TestCase):
greenthread.sleep(0.3)
return result['snapshotId']
@test.skip_test("skipping, test is hanging with multinic for rpc reasons")
def test_run_with_snapshot(self):
"""Makes sure run/stop/start instance with snapshot works."""
vol = self._volume_create()

46
nova/tests/test_compute.py
View File

@@ -19,10 +19,6 @@
Tests For Compute
"""
import mox
import stubout
from nova.auth import manager
from nova import compute
from nova.compute import instance_types
from nova.compute import manager as compute_manager
@@ -67,10 +63,9 @@ class ComputeTestCase(test.TestCase):
network_manager='nova.network.manager.FlatManager')
self.compute = utils.import_object(FLAGS.compute_manager)
self.compute_api = compute.API()
self.manager = manager.AuthManager()
self.user = self.manager.create_user('fake', 'fake', 'fake')
self.project = self.manager.create_project('fake', 'fake', 'fake')
self.context = context.RequestContext('fake', 'fake', False)
self.user_id = 'fake'
self.project_id = 'fake'
self.context = context.RequestContext(self.user_id, self.project_id)
test_notifier.NOTIFICATIONS = []
def fake_show(meh, context, id):
@@ -78,19 +73,14 @@ class ComputeTestCase(test.TestCase):
self.stubs.Set(nova.image.fake._FakeImageService, 'show', fake_show)
def tearDown(self):
self.manager.delete_user(self.user)
self.manager.delete_project(self.project)
super(ComputeTestCase, self).tearDown()
def _create_instance(self, params={}):
"""Create a test instance"""
inst = {}
inst['image_ref'] = 1
inst['reservation_id'] = 'r-fakeres'
inst['launch_time'] = '10'
inst['user_id'] = self.user.id
inst['project_id'] = self.project.id
inst['user_id'] = self.user_id
inst['project_id'] = self.project_id
type_id = instance_types.get_instance_type_by_name('m1.tiny')['id']
inst['instance_type_id'] = type_id
inst['ami_launch_index'] = 0
@@ -115,8 +105,8 @@ class ComputeTestCase(test.TestCase):
def _create_group(self):
values = {'name': 'testgroup',
'description': 'testgroup',
'user_id': self.user.id,
'project_id': self.project.id}
'user_id': self.user_id,
'project_id': self.project_id}
return db.security_group_create(self.context, values)
def _get_dummy_instance(self):
@@ -350,8 +340,8 @@ class ComputeTestCase(test.TestCase):
self.assertEquals(msg['priority'], 'INFO')
self.assertEquals(msg['event_type'], 'compute.instance.create')
payload = msg['payload']
self.assertEquals(payload['tenant_id'], self.project.id)
self.assertEquals(payload['user_id'], self.user.id)
self.assertEquals(payload['tenant_id'], self.project_id)
self.assertEquals(payload['user_id'], self.user_id)
self.assertEquals(payload['instance_id'], instance_id)
self.assertEquals(payload['instance_type'], 'm1.tiny')
type_id = instance_types.get_instance_type_by_name('m1.tiny')['id']
@@ -374,8 +364,8 @@ class ComputeTestCase(test.TestCase):
self.assertEquals(msg['priority'], 'INFO')
self.assertEquals(msg['event_type'], 'compute.instance.delete')
payload = msg['payload']
self.assertEquals(payload['tenant_id'], self.project.id)
self.assertEquals(payload['user_id'], self.user.id)
self.assertEquals(payload['tenant_id'], self.project_id)
self.assertEquals(payload['user_id'], self.user_id)
self.assertEquals(payload['instance_id'], instance_id)
self.assertEquals(payload['instance_type'], 'm1.tiny')
type_id = instance_types.get_instance_type_by_name('m1.tiny')['id']
@@ -420,7 +410,7 @@ class ComputeTestCase(test.TestCase):
def fake(*args, **kwargs):
pass
self.stubs.Set(self.compute.driver, 'finish_resize', fake)
self.stubs.Set(self.compute.driver, 'finish_migration', fake)
self.stubs.Set(self.compute.network_api, 'get_instance_nw_info', fake)
context = self.context.elevated()
instance_id = self._create_instance()
@@ -457,8 +447,8 @@ class ComputeTestCase(test.TestCase):
self.assertEquals(msg['priority'], 'INFO')
self.assertEquals(msg['event_type'], 'compute.instance.resize.prep')
payload = msg['payload']
self.assertEquals(payload['tenant_id'], self.project.id)
self.assertEquals(payload['user_id'], self.user.id)
self.assertEquals(payload['tenant_id'], self.project_id)
self.assertEquals(payload['user_id'], self.user_id)
self.assertEquals(payload['instance_id'], instance_id)
self.assertEquals(payload['instance_type'], 'm1.tiny')
type_id = instance_types.get_instance_type_by_name('m1.tiny')['id']
@@ -531,8 +521,8 @@ class ComputeTestCase(test.TestCase):
def fake(*args, **kwargs):
pass
self.stubs.Set(self.compute.driver, 'finish_resize', fake)
self.stubs.Set(self.compute.driver, 'revert_resize', fake)
self.stubs.Set(self.compute.driver, 'finish_migration', fake)
self.stubs.Set(self.compute.driver, 'revert_migration', fake)
self.stubs.Set(self.compute.network_api, 'get_instance_nw_info', fake)
self.compute.run_instance(self.context, instance_id)
@@ -583,8 +573,9 @@ class ComputeTestCase(test.TestCase):
the same host"""
instance_id = self._create_instance()
self.compute.run_instance(self.context, instance_id)
inst_ref = db.instance_get(self.context, instance_id)
self.assertRaises(exception.Error, self.compute.prep_resize,
self.context, instance_id, 1)
self.context, inst_ref['uuid'], 1)
self.compute.terminate_instance(self.context, instance_id)
def test_migrate(self):
@@ -866,7 +857,6 @@ class ComputeTestCase(test.TestCase):
def test_run_kill_vm(self):
"""Detect when a vm is terminated behind the scenes"""
self.stubs = stubout.StubOutForTesting()
self.stubs.Set(compute_manager.ComputeManager,
'_report_driver_status', nop_report_driver_status)

19
nova/tests/test_console.py
View File

@@ -26,10 +26,9 @@ from nova import exception
from nova import flags
from nova import test
from nova import utils
from nova.auth import manager
from nova.console import manager as console_manager
FLAGS = flags.FLAGS
flags.DECLARE('console_driver', 'nova.console.manager')
class ConsoleTestCase(test.TestCase):
@@ -39,17 +38,11 @@ class ConsoleTestCase(test.TestCase):
self.flags(console_driver='nova.console.fake.FakeConsoleProxy',
stub_compute=True)
self.console = utils.import_object(FLAGS.console_manager)
self.manager = manager.AuthManager()
self.user = self.manager.create_user('fake', 'fake', 'fake')
self.project = self.manager.create_project('fake', 'fake', 'fake')
self.context = context.get_admin_context()
self.user_id = 'fake'
self.project_id = 'fake'
self.context = context.RequestContext(self.user_id, self.project_id)
self.host = 'test_compute_host'
def tearDown(self):
self.manager.delete_user(self.user)
self.manager.delete_project(self.project)
super(ConsoleTestCase, self).tearDown()
def _create_instance(self):
"""Create a test instance"""
inst = {}
@@ -58,8 +51,8 @@ class ConsoleTestCase(test.TestCase):
inst['image_id'] = 1
inst['reservation_id'] = 'r-fakeres'
inst['launch_time'] = '10'
inst['user_id'] = self.user.id
inst['project_id'] = self.project.id
inst['user_id'] = self.user_id
inst['project_id'] = self.project_id
inst['instance_type_id'] = 1
inst['ami_launch_index'] = 0
return db.instance_create(self.context, inst)['id']

34
nova/tests/test_db_api.py
View File

@@ -22,7 +22,6 @@ from nova import test
from nova import context
from nova import db
from nova import flags
from nova.auth import manager
FLAGS = flags.FLAGS
@@ -45,40 +44,33 @@ def _setup_networking(instance_id, ip='1.2.3.4', flo_addr='1.2.1.2'):
db.fixed_ip_create(ctxt, fixed_ip)
fix_ref = db.fixed_ip_get_by_address(ctxt, ip)
db.floating_ip_create(ctxt, {'address': flo_addr,
'fixed_ip_id': fix_ref.id})
'fixed_ip_id': fix_ref['id']})
class DbApiTestCase(test.TestCase):
def setUp(self):
super(DbApiTestCase, self).setUp()
self.manager = manager.AuthManager()
self.user = self.manager.create_user('admin', 'admin', 'admin', True)
self.project = self.manager.create_project('proj', 'admin', 'proj')
self.context = context.RequestContext(user=self.user,
project=self.project)
def tearDown(self):
self.manager.delete_project(self.project)
self.manager.delete_user(self.user)
super(DbApiTestCase, self).tearDown()
self.user_id = 'fake'
self.project_id = 'fake'
self.context = context.RequestContext(self.user_id, self.project_id)
def test_instance_get_project_vpn(self):
result = db.fixed_ip_get_all(self.context)
values = {'instance_type_id': FLAGS.default_instance_type,
'image_ref': FLAGS.vpn_image_id,
'project_id': self.project.id}
'project_id': self.project_id}
instance = db.instance_create(self.context, values)
result = db.instance_get_project_vpn(self.context, self.project.id)
self.assertEqual(instance.id, result.id)
result = db.instance_get_project_vpn(self.context.elevated(),
self.project_id)
self.assertEqual(instance['id'], result['id'])
def test_instance_get_project_vpn_joins(self):
result = db.fixed_ip_get_all(self.context)
values = {'instance_type_id': FLAGS.default_instance_type,
'image_ref': FLAGS.vpn_image_id,
'project_id': self.project.id}
'project_id': self.project_id}
instance = db.instance_create(self.context, values)
_setup_networking(instance.id)
result = db.instance_get_project_vpn(self.context, self.project.id)
self.assertEqual(instance.id, result.id)
_setup_networking(instance['id'])
result = db.instance_get_project_vpn(self.context.elevated(),
self.project_id)
self.assertEqual(instance['id'], result['id'])
self.assertEqual(result['fixed_ips'][0]['floating_ips'][0].address,
'1.2.1.2')

201
nova/tests/test_libvirt.py
View File

@@ -33,7 +33,6 @@ from nova import flags
from nova import test
from nova import utils
from nova.api.ec2 import cloud
from nova.auth import manager
from nova.compute import power_state
from nova.virt.libvirt import connection
from nova.virt.libvirt import firewall
@@ -89,7 +88,7 @@ def _setup_networking(instance_id, ip='1.2.3.4'):
'virtual_interface_id': vif_ref['id']}
db.fixed_ip_create(ctxt, fixed_ip)
db.fixed_ip_update(ctxt, ip, {'allocated': True,
'instance_id': instance_id})
'instance_id': instance_id})
class CacheConcurrencyTestCase(test.TestCase):
@@ -155,36 +154,15 @@ class LibvirtConnTestCase(test.TestCase):
super(LibvirtConnTestCase, self).setUp()
connection._late_load_cheetah()
self.flags(fake_call=True)
self.manager = manager.AuthManager()
try:
pjs = self.manager.get_projects()
pjs = [p for p in pjs if p.name == 'fake']
if 0 != len(pjs):
self.manager.delete_project(pjs[0])
users = self.manager.get_users()
users = [u for u in users if u.name == 'fake']
if 0 != len(users):
self.manager.delete_user(users[0])
except Exception, e:
pass
users = self.manager.get_users()
self.user = self.manager.create_user('fake', 'fake', 'fake',
admin=True)
self.project = self.manager.create_project('fake', 'fake', 'fake')
self.user_id = 'fake'
self.project_id = 'fake'
self.context = context.RequestContext(self.user_id, self.project_id)
self.network = utils.import_object(FLAGS.network_manager)
self.context = context.get_admin_context()
FLAGS.instances_path = ''
self.call_libvirt_dependant_setup = False
self.test_ip = '10.11.12.13'
def tearDown(self):
self.manager.delete_project(self.project)
self.manager.delete_user(self.user)
super(LibvirtConnTestCase, self).tearDown()
test_ip = '10.11.12.13'
test_instance = {'memory_kb': '1024000',
'basepath': '/some/path',
'bridge_name': 'br100',
@@ -233,6 +211,16 @@ class LibvirtConnTestCase(test.TestCase):
def setattr(self, key, val):
self.__setattr__(key, val)
def plug(self, instance, network, mapping):
return {
'id': 'fake',
'bridge_name': 'fake',
'mac_address': 'fake',
'ip_address': 'fake',
'dhcp_server': 'fake',
'extra_params': 'fake',
}
# Creating mocks
fake = FakeLibvirtConnection()
fakeip = FakeIptablesFirewallDriver
@@ -281,7 +269,6 @@ class LibvirtConnTestCase(test.TestCase):
return db.service_create(context.get_admin_context(), service_ref)
@test.skip_test("Please review this test to ensure intent")
def test_preparing_xml_info(self):
conn = connection.LibvirtConnection(True)
instance_ref = db.instance_create(self.context, self.test_instance)
@@ -297,27 +284,23 @@ class LibvirtConnTestCase(test.TestCase):
_create_network_info(2))
self.assertTrue(len(result['nics']) == 2)
@test.skip_test("skipping libvirt tests depends on get_network_info shim")
def test_xml_and_uri_no_ramdisk_no_kernel(self):
instance_data = dict(self.test_instance)
self._check_xml_and_uri(instance_data,
expect_kernel=False, expect_ramdisk=False)
@test.skip_test("skipping libvirt tests depends on get_network_info shim")
def test_xml_and_uri_no_ramdisk(self):
instance_data = dict(self.test_instance)
instance_data['kernel_id'] = 'aki-deadbeef'
self._check_xml_and_uri(instance_data,
expect_kernel=True, expect_ramdisk=False)
@test.skip_test("skipping libvirt tests depends on get_network_info shim")
def test_xml_and_uri_no_kernel(self):
instance_data = dict(self.test_instance)
instance_data['ramdisk_id'] = 'ari-deadbeef'
self._check_xml_and_uri(instance_data,
expect_kernel=False, expect_ramdisk=False)
@test.skip_test("skipping libvirt tests depends on get_network_info shim")
def test_xml_and_uri(self):
instance_data = dict(self.test_instance)
instance_data['ramdisk_id'] = 'ari-deadbeef'
@@ -325,7 +308,6 @@ class LibvirtConnTestCase(test.TestCase):
self._check_xml_and_uri(instance_data,
expect_kernel=True, expect_ramdisk=True)
@test.skip_test("skipping libvirt tests depends on get_network_info shim")
def test_xml_and_uri_rescue(self):
instance_data = dict(self.test_instance)
instance_data['ramdisk_id'] = 'ari-deadbeef'
@@ -333,7 +315,6 @@ class LibvirtConnTestCase(test.TestCase):
self._check_xml_and_uri(instance_data, expect_kernel=True,
expect_ramdisk=True, rescue=True)
@test.skip_test("skipping libvirt tests depends on get_network_info shim")
def test_lxc_container_and_uri(self):
instance_data = dict(self.test_instance)
self._check_xml_and_container(instance_data)
@@ -413,6 +394,15 @@ class LibvirtConnTestCase(test.TestCase):
self.assertEquals(snapshot['status'], 'active')
self.assertEquals(snapshot['name'], snapshot_name)
def test_attach_invalid_device(self):
self.create_fake_libvirt_mock()
connection.LibvirtConnection._conn.lookupByName = self.fake_lookup
self.mox.ReplayAll()
conn = connection.LibvirtConnection(False)
self.assertRaises(exception.InvalidDevicePath,
conn.attach_volume,
"fake", "bad/device/path", "/dev/fake")
def test_multi_nic(self):
instance_data = dict(self.test_instance)
network_info = _create_network_info(2)
@@ -430,27 +420,10 @@ class LibvirtConnTestCase(test.TestCase):
self.assertEquals(parameters[1].get('value'), 'fake')
def _check_xml_and_container(self, instance):
user_context = context.RequestContext(project=self.project,
user=self.user)
user_context = context.RequestContext(self.user_id,
self.project_id)
instance_ref = db.instance_create(user_context, instance)
# Re-get the instance so it's bound to an actual session
instance_ref = db.instance_get(user_context, instance_ref['id'])
network_ref = db.project_get_networks(context.get_admin_context(),
self.project.id)[0]
vif = {'address': '56:12:12:12:12:12',
'network_id': network_ref['id'],
'instance_id': instance_ref['id']}
vif_ref = db.virtual_interface_create(self.context, vif)
fixed_ip = {'address': self.test_ip,
'network_id': network_ref['id'],
'virtual_interface_id': vif_ref['id']}
ctxt = context.get_admin_context()
fixed_ip_ref = db.fixed_ip_create(ctxt, fixed_ip)
db.fixed_ip_update(ctxt, self.test_ip,
{'allocated': True,
'instance_id': instance_ref['id']})
_setup_networking(instance_ref['id'], self.test_ip)
self.flags(libvirt_type='lxc')
conn = connection.LibvirtConnection(True)
@@ -476,13 +449,12 @@ class LibvirtConnTestCase(test.TestCase):
def _check_xml_and_uri(self, instance, expect_ramdisk, expect_kernel,
rescue=False):
user_context = context.RequestContext(project=self.project,
user=self.user)
user_context = context.RequestContext(self.user_id, self.project_id)
instance_ref = db.instance_create(user_context, instance)
network_ref = db.project_get_networks(context.get_admin_context(),
self.project.id)[0]
self.project_id)[0]
_setup_networking(instance_ref['id'], ip=self.test_ip)
_setup_networking(instance_ref['id'], self.test_ip)
type_uri_map = {'qemu': ('qemu:///system',
[(lambda t: t.find('.').get('type'), 'qemu'),
@@ -836,7 +808,6 @@ class LibvirtConnTestCase(test.TestCase):
db.instance_destroy(self.context, instance_ref['id'])
@test.skip_test("test needs rewrite: instance no longer has mac_address")
def test_spawn_with_network_info(self):
# Skip if non-libvirt environment
if not self.lazy_load_library_exists():
@@ -855,17 +826,7 @@ class LibvirtConnTestCase(test.TestCase):
conn.firewall_driver.setattr('setup_basic_filtering', fake_none)
conn.firewall_driver.setattr('prepare_instance_filter', fake_none)
network = db.project_get_networks(context.get_admin_context(),
self.project.id)[0]
ip_dict = {'ip': self.test_ip,
'netmask': network['netmask'],
'enabled': '1'}
mapping = {'label': network['label'],
'gateway': network['gateway'],
'mac': instance['mac_address'],
'dns': [network['dns']],
'ips': [ip_dict]}
network_info = [(network, mapping)]
network_info = _create_network_info()
try:
conn.spawn(instance, network_info)
@@ -911,11 +872,9 @@ class IptablesFirewallTestCase(test.TestCase):
def setUp(self):
super(IptablesFirewallTestCase, self).setUp()
self.manager = manager.AuthManager()
self.user = self.manager.create_user('fake', 'fake', 'fake',
admin=True)
self.project = self.manager.create_project('fake', 'fake', 'fake')
self.context = context.RequestContext('fake', 'fake')
self.user_id = 'fake'
self.project_id = 'fake'
self.context = context.RequestContext(self.user_id, self.project_id)
self.network = utils.import_object(FLAGS.network_manager)
class FakeLibvirtConnection(object):
@@ -923,6 +882,7 @@ class IptablesFirewallTestCase(test.TestCase):
"""setup_basic_rules in nwfilter calls this."""
pass
self.fake_libvirt_connection = FakeLibvirtConnection()
self.test_ip = '10.11.12.13'
self.fw = firewall.IptablesFirewallDriver(
get_connection=lambda: self.fake_libvirt_connection)
@@ -940,11 +900,6 @@ class IptablesFirewallTestCase(test.TestCase):
connection.libxml2 = __import__('libxml2')
return True
def tearDown(self):
self.manager.delete_project(self.project)
self.manager.delete_user(self.user)
super(IptablesFirewallTestCase, self).tearDown()
in_nat_rules = [
'# Generated by iptables-save v1.4.10 on Sat Feb 19 00:03:19 2011',
'*nat',
@@ -988,27 +943,11 @@ class IptablesFirewallTestCase(test.TestCase):
'project_id': 'fake',
'instance_type_id': 1})
@test.skip_test("skipping libvirt tests depends on get_network_info shim")
def test_static_filters(self):
instance_ref = self._create_instance_ref()
ip = '10.11.12.13'
_setup_networking(instance_ref['id'], self.test_ip)
network_ref = db.project_get_networks(self.context,
'fake',
associate=True)[0]
vif = {'address': '56:12:12:12:12:12',
'network_id': network_ref['id'],
'instance_id': instance_ref['id']}
vif_ref = db.virtual_interface_create(self.context, vif)
fixed_ip = {'address': ip,
'network_id': network_ref['id'],
'virtual_interface_id': vif_ref['id']}
admin_ctxt = context.get_admin_context()
db.fixed_ip_create(admin_ctxt, fixed_ip)
db.fixed_ip_update(admin_ctxt, ip, {'allocated': True,
'instance_id': instance_ref['id']})
secgroup = db.security_group_create(admin_ctxt,
{'user_id': 'fake',
'project_id': 'fake',
@@ -1140,7 +1079,6 @@ class IptablesFirewallTestCase(test.TestCase):
self.assertEquals(ipv6_network_rules,
ipv6_rules_per_network * networks_count)
@test.skip_test("skipping libvirt tests")
def test_do_refresh_security_group_rules(self):
instance_ref = self._create_instance_ref()
self.mox.StubOutWithMock(self.fw,
@@ -1151,7 +1089,6 @@ class IptablesFirewallTestCase(test.TestCase):
self.mox.ReplayAll()
self.fw.do_refresh_security_group_rules("fake")
@test.skip_test("skip libvirt test project_get_network no longer exists")
def test_unfilter_instance_undefines_nwfilter(self):
# Skip if non-libvirt environment
if not self.lazy_load_library_exists():
@@ -1165,38 +1102,24 @@ class IptablesFirewallTestCase(test.TestCase):
self.fw.nwfilter._conn.nwfilterLookupByName =\
fakefilter.nwfilterLookupByName
instance_ref = self._create_instance_ref()
inst_id = instance_ref['id']
instance = db.instance_get(self.context, inst_id)
ip = '10.11.12.13'
network_ref = db.project_get_network(self.context, 'fake')
fixed_ip = {'address': ip, 'network_id': network_ref['id']}
db.fixed_ip_create(admin_ctxt, fixed_ip)
db.fixed_ip_update(admin_ctxt, ip, {'allocated': True,
'instance_id': inst_id})
self.fw.setup_basic_filtering(instance)
self.fw.prepare_instance_filter(instance)
self.fw.apply_instance_filter(instance)
_setup_networking(instance_ref['id'], self.test_ip)
self.fw.setup_basic_filtering(instance_ref)
self.fw.prepare_instance_filter(instance_ref)
self.fw.apply_instance_filter(instance_ref)
original_filter_count = len(fakefilter.filters)
self.fw.unfilter_instance(instance)
self.fw.unfilter_instance(instance_ref)
# should undefine just the instance filter
self.assertEqual(original_filter_count - len(fakefilter.filters), 1)
db.instance_destroy(admin_ctxt, instance_ref['id'])
@test.skip_test("skip libvirt test project_get_network no longer exists")
def test_provider_firewall_rules(self):
# setup basic instance data
instance_ref = self._create_instance_ref()
nw_info = _create_network_info(1)
ip = '10.11.12.13'
network_ref = db.project_get_network(self.context, 'fake')
admin_ctxt = context.get_admin_context()
fixed_ip = {'address': ip, 'network_id': network_ref['id']}
db.fixed_ip_create(admin_ctxt, fixed_ip)
db.fixed_ip_update(admin_ctxt, ip, {'allocated': True,
'instance_id': instance_ref['id']})
_setup_networking(instance_ref['id'], self.test_ip)
# FRAGILE: peeks at how the firewall names chains
chain_name = 'inst-%s' % instance_ref['id']
@@ -1208,6 +1131,7 @@ class IptablesFirewallTestCase(test.TestCase):
if rule.chain == 'provider']
self.assertEqual(0, len(rules))
admin_ctxt = context.get_admin_context()
# add a rule and send the update message, check for 1 rule
provider_fw0 = db.provider_fw_rule_create(admin_ctxt,
{'protocol': 'tcp',
@@ -1258,22 +1182,16 @@ class NWFilterTestCase(test.TestCase):
class Mock(object):
pass
self.manager = manager.AuthManager()
self.user = self.manager.create_user('fake', 'fake', 'fake',
admin=True)
self.project = self.manager.create_project('fake', 'fake', 'fake')
self.context = context.RequestContext(self.user, self.project)
self.user_id = 'fake'
self.project_id = 'fake'
self.context = context.RequestContext(self.user_id, self.project_id)
self.fake_libvirt_connection = Mock()
self.test_ip = '10.11.12.13'
self.fw = firewall.NWFilterFirewall(
lambda: self.fake_libvirt_connection)
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()
cloud_controller.create_security_group(self.context,
@@ -1352,7 +1270,6 @@ class NWFilterTestCase(test.TestCase):
inst.update(params)
return db.instance_type_create(context, inst)['id']
@test.skip_test('Skipping this test')
def test_creates_base_rule_first(self):
# These come pre-defined by libvirt
self.defined_filters = ['no-mac-spoofing',
@@ -1384,21 +1301,11 @@ class NWFilterTestCase(test.TestCase):
instance_ref = self._create_instance()
inst_id = instance_ref['id']
ip = '10.11.12.13'
#network_ref = db.project_get_networks(self.context, 'fake')[0]
#fixed_ip = {'address': ip, 'network_id': network_ref['id']}
#admin_ctxt = context.get_admin_context()
#db.fixed_ip_create(admin_ctxt, fixed_ip)
#db.fixed_ip_update(admin_ctxt, ip, {'allocated': True,
# 'instance_id': inst_id})
self._setup_networking(instance_ref['id'], ip=ip)
_setup_networking(instance_ref['id'], self.test_ip)
def _ensure_all_called():
instance_filter = 'nova-instance-%s-%s' % (instance_ref['name'],
'00A0C914C829')
'561212121212')
secgroup_filter = 'nova-secgroup-%s' % self.security_group['id']
for required in [secgroup_filter, 'allow-dhcp-server',
'no-arp-spoofing', 'no-ip-spoofing',
@@ -1419,7 +1326,7 @@ class NWFilterTestCase(test.TestCase):
self.fw.apply_instance_filter(instance)
_ensure_all_called()
self.teardown_security_group()
db.instance_destroy(admin_ctxt, instance_ref['id'])
db.instance_destroy(context.get_admin_context(), instance_ref['id'])
def test_create_network_filters(self):
instance_ref = self._create_instance()
@@ -1429,7 +1336,6 @@ class NWFilterTestCase(test.TestCase):
"fake")
self.assertEquals(len(result), 3)
@test.skip_test("skip libvirt test project_get_network no longer exists")
def test_unfilter_instance_undefines_nwfilters(self):
admin_ctxt = context.get_admin_context()
@@ -1447,12 +1353,7 @@ class NWFilterTestCase(test.TestCase):
instance = db.instance_get(self.context, inst_id)
ip = '10.11.12.13'
network_ref = db.project_get_network(self.context, 'fake')
fixed_ip = {'address': ip, 'network_id': network_ref['id']}
db.fixed_ip_create(admin_ctxt, fixed_ip)
db.fixed_ip_update(admin_ctxt, ip, {'allocated': True,
'instance_id': inst_id})
_setup_networking(instance_ref['id'], self.test_ip)
self.fw.setup_basic_filtering(instance)
self.fw.prepare_instance_filter(instance)
self.fw.apply_instance_filter(instance)

61
nova/tests/test_rpc.py
View File

@@ -33,11 +33,12 @@ LOG = logging.getLogger('nova.tests.rpc')
class RpcTestCase(test.TestCase):
def setUp(self):
super(RpcTestCase, self).setUp()
self.conn = rpc.Connection.instance(True)
self.conn = rpc.create_connection(True)
self.receiver = TestReceiver()
self.consumer = rpc.TopicAdapterConsumer(connection=self.conn,
topic='test',
proxy=self.receiver)
self.consumer = rpc.create_consumer(self.conn,
'test',
self.receiver,
False)
self.consumer.attach_to_eventlet()
self.context = context.get_admin_context()
@@ -129,6 +130,8 @@ class RpcTestCase(test.TestCase):
"""Calls echo in the passed queue"""
LOG.debug(_("Nested received %(queue)s, %(value)s")
% locals())
# TODO: so, it will replay the context and use the same REQID?
# that's bizarre.
ret = rpc.call(context,
queue,
{"method": "echo",
@@ -137,10 +140,11 @@ class RpcTestCase(test.TestCase):
return value
nested = Nested()
conn = rpc.Connection.instance(True)
consumer = rpc.TopicAdapterConsumer(connection=conn,
topic='nested',
proxy=nested)
conn = rpc.create_connection(True)
consumer = rpc.create_consumer(conn,
'nested',
nested,
False)
consumer.attach_to_eventlet()
value = 42
result = rpc.call(self.context,
@@ -149,47 +153,6 @@ class RpcTestCase(test.TestCase):
"value": value}})
self.assertEqual(value, result)
def test_connectionpool_single(self):
"""Test that ConnectionPool recycles a single connection."""
conn1 = rpc.ConnectionPool.get()
rpc.ConnectionPool.put(conn1)
conn2 = rpc.ConnectionPool.get()
rpc.ConnectionPool.put(conn2)
self.assertEqual(conn1, conn2)
def test_connectionpool_double(self):
"""Test that ConnectionPool returns and reuses separate connections.
When called consecutively we should get separate connections and upon
returning them those connections should be reused for future calls
before generating a new connection.
"""
conn1 = rpc.ConnectionPool.get()
conn2 = rpc.ConnectionPool.get()
self.assertNotEqual(conn1, conn2)
rpc.ConnectionPool.put(conn1)
rpc.ConnectionPool.put(conn2)
conn3 = rpc.ConnectionPool.get()
conn4 = rpc.ConnectionPool.get()
self.assertEqual(conn1, conn3)
self.assertEqual(conn2, conn4)
def test_connectionpool_limit(self):
"""Test connection pool limit and connection uniqueness."""
max_size = FLAGS.rpc_conn_pool_size
conns = []
for i in xrange(max_size):
conns.append(rpc.ConnectionPool.get())
self.assertFalse(rpc.ConnectionPool.free_items)
self.assertEqual(rpc.ConnectionPool.current_size,
rpc.ConnectionPool.max_size)
self.assertEqual(len(set(conns)), max_size)
class TestReceiver(object):
"""Simple Proxy class so the consumer has methods to call.

68
nova/tests/test_rpc_amqp.py Normal file
View File

@@ -0,0 +1,68 @@
from nova import context
from nova import flags
from nova import log as logging
from nova import rpc
from nova.rpc import amqp
from nova import test
FLAGS = flags.FLAGS
LOG = logging.getLogger('nova.tests.rpc')
class RpcAMQPTestCase(test.TestCase):
def setUp(self):
super(RpcAMQPTestCase, self).setUp()
self.conn = rpc.create_connection(True)
self.receiver = TestReceiver()
self.consumer = rpc.create_consumer(self.conn,
'test',
self.receiver,
False)
self.consumer.attach_to_eventlet()
self.context = context.get_admin_context()
def test_connectionpool_single(self):
"""Test that ConnectionPool recycles a single connection."""
conn1 = amqp.ConnectionPool.get()
amqp.ConnectionPool.put(conn1)
conn2 = amqp.ConnectionPool.get()
amqp.ConnectionPool.put(conn2)
self.assertEqual(conn1, conn2)
class TestReceiver(object):
"""Simple Proxy class so the consumer has methods to call.
Uses static methods because we aren't actually storing any state.
"""
@staticmethod
def echo(context, value):
"""Simply returns whatever value is sent in."""
LOG.debug(_("Received %s"), value)
return value
@staticmethod
def context(context, value):
"""Returns dictionary version of context."""
LOG.debug(_("Received %s"), context)
return context.to_dict()
@staticmethod
def echo_three_times(context, value):
context.reply(value)
context.reply(value + 1)
context.reply(value + 2)
@staticmethod
def echo_three_times_yield(context, value):
yield value
yield value + 1
yield value + 2
@staticmethod
def fail(context, value):
"""Raises an exception with the value sent in."""
raise Exception(value)

106
nova/tests/test_vmwareapi.py
View File

@@ -19,14 +19,11 @@
Test suite for VMWareAPI.
"""
import stubout
from nova import context
from nova import db
from nova import flags
from nova import test
from nova import utils
from nova.auth import manager
from nova.compute import power_state
from nova.tests.glance import stubs as glance_stubs
from nova.tests.vmwareapi import db_fakes
@@ -41,51 +38,63 @@ FLAGS = flags.FLAGS
class VMWareAPIVMTestCase(test.TestCase):
"""Unit tests for Vmware API connection calls."""
# NOTE(jkoelker): This is leaking stubs into the db module.
# Commenting out until updated for multi-nic.
#def setUp(self):
# super(VMWareAPIVMTestCase, self).setUp()
# self.flags(vmwareapi_host_ip='test_url',
# vmwareapi_host_username='test_username',
# vmwareapi_host_password='test_pass')
# self.manager = manager.AuthManager()
# self.user = self.manager.create_user('fake', 'fake', 'fake',
# admin=True)
# self.project = self.manager.create_project('fake', 'fake', 'fake')
# self.network = utils.import_object(FLAGS.network_manager)
# self.stubs = stubout.StubOutForTesting()
# vmwareapi_fake.reset()
# db_fakes.stub_out_db_instance_api(self.stubs)
# stubs.set_stubs(self.stubs)
# glance_stubs.stubout_glance_client(self.stubs,
# glance_stubs.FakeGlance)
# self.conn = vmwareapi_conn.get_connection(False)
def setUp(self):
super(VMWareAPIVMTestCase, self).setUp()
self.flags(vmwareapi_host_ip='test_url',
vmwareapi_host_username='test_username',
vmwareapi_host_password='test_pass')
self.user_id = 'fake'
self.project_id = 'fake'
self.context = context.RequestContext(self.user_id, self.project_id)
self.network = utils.import_object(FLAGS.network_manager)
vmwareapi_fake.reset()
db_fakes.stub_out_db_instance_api(self.stubs)
stubs.set_stubs(self.stubs)
glance_stubs.stubout_glance_client(self.stubs)
self.conn = vmwareapi_conn.get_connection(False)
# NOTE(vish): none of the network plugging code is actually
# being tested
self.network_info = [({'bridge': 'fa0',
'id': 0,
'vlan': None,
'bridge_interface': None,
'injected': True},
{'broadcast': '192.168.0.255',
'dns': ['192.168.0.1'],
'gateway': '192.168.0.1',
'gateway6': 'dead:beef::1',
'ip6s': [{'enabled': '1',
'ip': 'dead:beef::dcad:beff:feef:0',
'netmask': '64'}],
'ips': [{'enabled': '1',
'ip': '192.168.0.100',
'netmask': '255.255.255.0'}],
'label': 'fake',
'mac': 'DE:AD:BE:EF:00:00',
'rxtx_cap': 3})]
#def tearDown(self):
# super(VMWareAPIVMTestCase, self).tearDown()
# vmwareapi_fake.cleanup()
# self.manager.delete_project(self.project)
# self.manager.delete_user(self.user)
# self.stubs.UnsetAll()
def tearDown(self):
super(VMWareAPIVMTestCase, self).tearDown()
vmwareapi_fake.cleanup()
def _create_instance_in_the_db(self):
values = {'name': 1,
'id': 1,
'project_id': self.project.id,
'user_id': self.user.id,
'image_id': "1",
'project_id': self.project_id,
'user_id': self.user_id,
'image_ref': "1",
'kernel_id': "1",
'ramdisk_id': "1",
'mac_address': "de:ad:be:ef:be:ef",
'instance_type': 'm1.large',
'mac_address': 'aa:bb:cc:dd:ee:ff',
}
self.instance = db.instance_create(values)
self.instance = db.instance_create(None, values)
def _create_vm(self):
"""Create and spawn the VM."""
self._create_instance_in_the_db()
self.type_data = db.instance_type_get_by_name(None, 'm1.large')
self.conn.spawn(self.instance)
self.conn.spawn(self.instance, self.network_info)
self._check_vm_record()
def _check_vm_record(self):
@@ -129,24 +138,20 @@ class VMWareAPIVMTestCase(test.TestCase):
self.assertEquals(info["mem"], mem_kib)
self.assertEquals(info["num_cpu"], self.type_data['vcpus'])
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_list_instances(self):
instances = self.conn.list_instances()
self.assertEquals(len(instances), 0)
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_list_instances_1(self):
self._create_vm()
instances = self.conn.list_instances()
self.assertEquals(len(instances), 1)
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_spawn(self):
self._create_vm()
info = self.conn.get_info(1)
self._check_vm_info(info, power_state.RUNNING)
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_snapshot(self):
self._create_vm()
info = self.conn.get_info(1)
@@ -155,27 +160,23 @@ class VMWareAPIVMTestCase(test.TestCase):
info = self.conn.get_info(1)
self._check_vm_info(info, power_state.RUNNING)
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_snapshot_non_existent(self):
self._create_instance_in_the_db()
self.assertRaises(Exception, self.conn.snapshot, self.instance,
"Test-Snapshot")
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_reboot(self):
self._create_vm()
info = self.conn.get_info(1)
self._check_vm_info(info, power_state.RUNNING)
self.conn.reboot(self.instance)
self.conn.reboot(self.instance, self.network_info)
info = self.conn.get_info(1)
self._check_vm_info(info, power_state.RUNNING)
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_reboot_non_existent(self):
self._create_instance_in_the_db()
self.assertRaises(Exception, self.conn.reboot, self.instance)
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_reboot_not_poweredon(self):
self._create_vm()
info = self.conn.get_info(1)
@@ -185,7 +186,6 @@ class VMWareAPIVMTestCase(test.TestCase):
self._check_vm_info(info, power_state.PAUSED)
self.assertRaises(Exception, self.conn.reboot, self.instance)
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_suspend(self):
self._create_vm()
info = self.conn.get_info(1)
@@ -194,13 +194,11 @@ class VMWareAPIVMTestCase(test.TestCase):
info = self.conn.get_info(1)
self._check_vm_info(info, power_state.PAUSED)
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_suspend_non_existent(self):
self._create_instance_in_the_db()
self.assertRaises(Exception, self.conn.suspend, self.instance,
self.dummy_callback_handler)
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_resume(self):
self._create_vm()
info = self.conn.get_info(1)
@@ -212,13 +210,11 @@ class VMWareAPIVMTestCase(test.TestCase):
info = self.conn.get_info(1)
self._check_vm_info(info, power_state.RUNNING)
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_resume_non_existent(self):
self._create_instance_in_the_db()
self.assertRaises(Exception, self.conn.resume, self.instance,
self.dummy_callback_handler)
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_resume_not_suspended(self):
self._create_vm()
info = self.conn.get_info(1)
@@ -226,49 +222,41 @@ class VMWareAPIVMTestCase(test.TestCase):
self.assertRaises(Exception, self.conn.resume, self.instance,
self.dummy_callback_handler)
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_get_info(self):
self._create_vm()
info = self.conn.get_info(1)
self._check_vm_info(info, power_state.RUNNING)
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_destroy(self):
self._create_vm()
info = self.conn.get_info(1)
self._check_vm_info(info, power_state.RUNNING)
instances = self.conn.list_instances()
self.assertEquals(len(instances), 1)
self.conn.destroy(self.instance)
self.conn.destroy(self.instance, self.network_info)
instances = self.conn.list_instances()
self.assertEquals(len(instances), 0)
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_destroy_non_existent(self):
self._create_instance_in_the_db()
self.assertEquals(self.conn.destroy(self.instance), None)
self.assertEquals(self.conn.destroy(self.instance, self.network_info),
None)
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_pause(self):
pass
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_unpause(self):
pass
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_diagnostics(self):
pass
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_get_console_output(self):
pass
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def test_get_ajax_console(self):
pass
@test.skip_test("DB stubbing not removed, needs updating for multi-nic")
def dummy_callback_handler(self, ret):
"""
Dummy callback function to be passed to suspend, resume, etc., calls.

222
nova/tests/test_xenapi.py
View File

@@ -30,7 +30,6 @@ from nova import flags
from nova import log as logging
from nova import test
from nova import utils
from nova.auth import manager
from nova.compute import instance_types
from nova.compute import power_state
from nova import exception
@@ -69,7 +68,9 @@ class XenAPIVolumeTestCase(test.TestCase):
def setUp(self):
super(XenAPIVolumeTestCase, self).setUp()
self.stubs = stubout.StubOutForTesting()
self.context = context.RequestContext('fake', 'fake', False)
self.user_id = 'fake'
self.project_id = 'fake'
self.context = context.RequestContext(self.user_id, self.project_id)
FLAGS.target_host = '127.0.0.1'
FLAGS.xenapi_connection_url = 'test_url'
FLAGS.xenapi_connection_password = 'test_pass'
@@ -77,7 +78,7 @@ class XenAPIVolumeTestCase(test.TestCase):
stubs.stub_out_get_target(self.stubs)
xenapi_fake.reset()
self.values = {'id': 1,
'project_id': 'fake',
'project_id': self.user_id,
'user_id': 'fake',
'image_ref': 1,
'kernel_id': 2,
@@ -173,10 +174,6 @@ class XenAPIVMTestCase(test.TestCase):
"""Unit tests for VM operations."""
def setUp(self):
super(XenAPIVMTestCase, self).setUp()
self.manager = manager.AuthManager()
self.user = self.manager.create_user('fake', 'fake', 'fake',
admin=True)
self.project = self.manager.create_project('fake', 'fake', 'fake')
self.network = utils.import_object(FLAGS.network_manager)
self.stubs = stubout.StubOutForTesting()
self.flags(xenapi_connection_url='test_url',
@@ -195,7 +192,9 @@ class XenAPIVMTestCase(test.TestCase):
stubs.stub_out_vm_methods(self.stubs)
glance_stubs.stubout_glance_client(self.stubs)
fake_utils.stub_out_utils_execute(self.stubs)
self.context = context.RequestContext('fake', 'fake', False)
self.user_id = 'fake'
self.project_id = 'fake'
self.context = context.RequestContext(self.user_id, self.project_id)
self.conn = xenapi_conn.get_connection(False)
def test_parallel_builds(self):
@@ -229,8 +228,8 @@ class XenAPIVMTestCase(test.TestCase):
instance = db.instance_create(self.context, values)
self.conn.spawn(instance, network_info)
gt1 = eventlet.spawn(_do_build, 1, self.project.id, self.user.id)
gt2 = eventlet.spawn(_do_build, 2, self.project.id, self.user.id)
gt1 = eventlet.spawn(_do_build, 1, self.project_id, self.user_id)
gt2 = eventlet.spawn(_do_build, 2, self.project_id, self.user_id)
gt1.wait()
gt2.wait()
@@ -396,18 +395,22 @@ class XenAPIVMTestCase(test.TestCase):
def _test_spawn(self, image_ref, kernel_id, ramdisk_id,
instance_type_id="3", os_type="linux",
architecture="x86-64", instance_id=1,
check_injection=False):
check_injection=False,
create_record=True):
stubs.stubout_loopingcall_start(self.stubs)
values = {'id': instance_id,
'project_id': self.project.id,
'user_id': self.user.id,
'image_ref': image_ref,
'kernel_id': kernel_id,
'ramdisk_id': ramdisk_id,
'instance_type_id': instance_type_id,
'os_type': os_type,
'architecture': architecture}
instance = db.instance_create(self.context, values)
if create_record:
values = {'id': instance_id,
'project_id': self.project_id,
'user_id': self.user_id,
'image_ref': image_ref,
'kernel_id': kernel_id,
'ramdisk_id': ramdisk_id,
'instance_type_id': instance_type_id,
'os_type': os_type,
'architecture': architecture}
instance = db.instance_create(self.context, values)
else:
instance = db.instance_get(self.context, instance_id)
network_info = [({'bridge': 'fa0', 'id': 0, 'injected': True},
{'broadcast': '192.168.0.255',
'dns': ['192.168.0.1'],
@@ -465,12 +468,30 @@ class XenAPIVMTestCase(test.TestCase):
self._check_vdis(vdi_recs_start, vdi_recs_end)
def test_spawn_raw_objectstore(self):
FLAGS.xenapi_image_service = 'objectstore'
self._test_spawn(1, None, None)
# TODO(vish): deprecated
from nova.auth import manager
authman = manager.AuthManager()
authman.create_user('fake', 'fake')
authman.create_project('fake', 'fake')
try:
FLAGS.xenapi_image_service = 'objectstore'
self._test_spawn(1, None, None)
finally:
authman.delete_project('fake')
authman.delete_user('fake')
def test_spawn_objectstore(self):
FLAGS.xenapi_image_service = 'objectstore'
self._test_spawn(1, 2, 3)
# TODO(vish): deprecated
from nova.auth import manager
authman = manager.AuthManager()
authman.create_user('fake', 'fake')
authman.create_project('fake', 'fake')
try:
FLAGS.xenapi_image_service = 'objectstore'
self._test_spawn(1, 2, 3)
finally:
authman.delete_project('fake')
authman.delete_user('fake')
@stub_vm_utils_with_vdi_attached_here
def test_spawn_raw_glance(self):
@@ -599,41 +620,38 @@ class XenAPIVMTestCase(test.TestCase):
# guest agent is detected
self.assertFalse(self._tee_executed)
@test.skip_test("Never gets an address, not sure why")
def test_spawn_vlanmanager(self):
self.flags(xenapi_image_service='glance',
network_manager='nova.network.manager.VlanManager',
network_driver='nova.network.xenapi_net',
vlan_interface='fake0')
def dummy(*args, **kwargs):
pass
self.stubs.Set(VMOps, 'create_vifs', dummy)
self.stubs.Set(vmops.VMOps, 'create_vifs', dummy)
# Reset network table
xenapi_fake.reset_table('network')
# Instance id = 2 will use vlan network (see db/fakes.py)
ctxt = self.context.elevated()
instance_ref = self._create_instance(2)
network_bk = self.network
# Ensure we use xenapi_net driver
self.network = utils.import_object(FLAGS.network_manager)
instance = self._create_instance(2, False)
networks = self.network.db.network_get_all(ctxt)
for network in networks:
self.network.set_network_host(ctxt, network['id'])
self.network.set_network_host(ctxt, network)
self.network.allocate_for_instance(ctxt, instance_id=instance_ref.id,
instance_type_id=1, project_id=self.project.id)
self.network.setup_compute_network(ctxt, instance_ref.id)
self.network.allocate_for_instance(ctxt,
instance_id=2,
host=FLAGS.host,
vpn=None,
instance_type_id=1,
project_id=self.project_id)
self._test_spawn(glance_stubs.FakeGlance.IMAGE_MACHINE,
glance_stubs.FakeGlance.IMAGE_KERNEL,
glance_stubs.FakeGlance.IMAGE_RAMDISK,
instance_id=instance_ref.id,
instance_id=2,
create_record=False)
# TODO(salvatore-orlando): a complete test here would require
# a check for making sure the bridge for the VM's VIF is
# consistent with bridge specified in nova db
self.network = network_bk
def test_spawn_with_network_qos(self):
self._create_instance()
@@ -644,7 +662,7 @@ class XenAPIVMTestCase(test.TestCase):
str(3 * 1024))
def test_rescue(self):
self.flags(xenapi_inject_image=False)
self.flags(flat_injected=False)
instance = self._create_instance()
conn = xenapi_conn.get_connection(False)
conn.rescue(instance, None, [])
@@ -655,21 +673,13 @@ class XenAPIVMTestCase(test.TestCase):
# Ensure that it will not unrescue a non-rescued instance.
self.assertRaises(Exception, conn.unrescue, instance, None)
def tearDown(self):
super(XenAPIVMTestCase, self).tearDown()
self.manager.delete_project(self.project)
self.manager.delete_user(self.user)
self.vm_info = None
self.vm = None
self.stubs.UnsetAll()
def _create_instance(self, instance_id=1):
def _create_instance(self, instance_id=1, spawn=True):
"""Creates and spawns a test instance."""
stubs.stubout_loopingcall_start(self.stubs)
values = {
'id': instance_id,
'project_id': self.project.id,
'user_id': self.user.id,
'project_id': self.project_id,
'user_id': self.user_id,
'image_ref': 1,
'kernel_id': 2,
'ramdisk_id': 3,
@@ -691,7 +701,8 @@ class XenAPIVMTestCase(test.TestCase):
'label': 'fake',
'mac': 'DE:AD:BE:EF:00:00',
'rxtx_cap': 3})]
self.conn.spawn(instance, network_info)
if spawn:
self.conn.spawn(instance, network_info)
return instance
@@ -750,14 +761,12 @@ class XenAPIMigrateInstance(test.TestCase):
stubs.stub_out_get_target(self.stubs)
xenapi_fake.reset()
xenapi_fake.create_network('fake', FLAGS.flat_network_bridge)
self.manager = manager.AuthManager()
self.user = self.manager.create_user('fake', 'fake', 'fake',
admin=True)
self.project = self.manager.create_project('fake', 'fake', 'fake')
self.context = context.RequestContext('fake', 'fake', False)
self.user_id = 'fake'
self.project_id = 'fake'
self.context = context.RequestContext(self.user_id, self.project_id)
self.values = {'id': 1,
'project_id': self.project.id,
'user_id': self.user.id,
'project_id': self.project_id,
'user_id': self.user_id,
'image_ref': 1,
'kernel_id': None,
'ramdisk_id': None,
@@ -771,20 +780,21 @@ class XenAPIMigrateInstance(test.TestCase):
stubs.stubout_get_this_vm_uuid(self.stubs)
glance_stubs.stubout_glance_client(self.stubs)
def tearDown(self):
super(XenAPIMigrateInstance, self).tearDown()
self.manager.delete_project(self.project)
self.manager.delete_user(self.user)
self.stubs.UnsetAll()
def test_migrate_disk_and_power_off(self):
instance = db.instance_create(self.context, self.values)
stubs.stubout_session(self.stubs, stubs.FakeSessionForMigrationTests)
conn = xenapi_conn.get_connection(False)
conn.migrate_disk_and_power_off(instance, '127.0.0.1')
def test_finish_resize(self):
def test_finish_migrate(self):
instance = db.instance_create(self.context, self.values)
self.called = False
def fake_vdi_resize(*args, **kwargs):
self.called = True
self.stubs.Set(stubs.FakeSessionForMigrationTests,
"VDI_resize_online", fake_vdi_resize)
stubs.stubout_session(self.stubs, stubs.FakeSessionForMigrationTests)
stubs.stubout_loopingcall_start(self.stubs)
conn = xenapi_conn.get_connection(False)
@@ -802,8 +812,86 @@ class XenAPIMigrateInstance(test.TestCase):
'label': 'fake',
'mac': 'DE:AD:BE:EF:00:00',
'rxtx_cap': 3})]
conn.finish_resize(instance, dict(base_copy='hurr', cow='durr'),
network_info)
conn.finish_migration(instance, dict(base_copy='hurr', cow='durr'),
network_info, resize_instance=True)
self.assertEqual(self.called, True)
def test_finish_migrate_no_local_storage(self):
tiny_type_id = \
instance_types.get_instance_type_by_name('m1.tiny')['id']
self.values.update({'instance_type_id': tiny_type_id, 'local_gb': 0})
instance = db.instance_create(self.context, self.values)
def fake_vdi_resize(*args, **kwargs):
raise Exception("This shouldn't be called")
self.stubs.Set(stubs.FakeSessionForMigrationTests,
"VDI_resize_online", fake_vdi_resize)
stubs.stubout_session(self.stubs, stubs.FakeSessionForMigrationTests)
stubs.stubout_loopingcall_start(self.stubs)
conn = xenapi_conn.get_connection(False)
network_info = [({'bridge': 'fa0', 'id': 0, 'injected': False},
{'broadcast': '192.168.0.255',
'dns': ['192.168.0.1'],
'gateway': '192.168.0.1',
'gateway6': 'dead:beef::1',
'ip6s': [{'enabled': '1',
'ip': 'dead:beef::dcad:beff:feef:0',
'netmask': '64'}],
'ips': [{'enabled': '1',
'ip': '192.168.0.100',
'netmask': '255.255.255.0'}],
'label': 'fake',
'mac': 'DE:AD:BE:EF:00:00',
'rxtx_cap': 3})]
conn.finish_migration(instance, dict(base_copy='hurr', cow='durr'),
network_info, resize_instance=True)
def test_finish_migrate_no_resize_vdi(self):
instance = db.instance_create(self.context, self.values)
def fake_vdi_resize(*args, **kwargs):
raise Exception("This shouldn't be called")
self.stubs.Set(stubs.FakeSessionForMigrationTests,
"VDI_resize_online", fake_vdi_resize)
stubs.stubout_session(self.stubs, stubs.FakeSessionForMigrationTests)
stubs.stubout_loopingcall_start(self.stubs)
conn = xenapi_conn.get_connection(False)
network_info = [({'bridge': 'fa0', 'id': 0, 'injected': False},
{'broadcast': '192.168.0.255',
'dns': ['192.168.0.1'],
'gateway': '192.168.0.1',
'gateway6': 'dead:beef::1',
'ip6s': [{'enabled': '1',
'ip': 'dead:beef::dcad:beff:feef:0',
'netmask': '64'}],
'ips': [{'enabled': '1',
'ip': '192.168.0.100',
'netmask': '255.255.255.0'}],
'label': 'fake',
'mac': 'DE:AD:BE:EF:00:00',
'rxtx_cap': 3})]
# Resize instance would be determined by the compute call
conn.finish_migration(instance, dict(base_copy='hurr', cow='durr'),
network_info, resize_instance=False)
class XenAPIImageTypeTestCase(test.TestCase):
"""Test ImageType class."""
def test_to_string(self):
"""Can convert from type id to type string."""
self.assertEquals(
vm_utils.ImageType.to_string(vm_utils.ImageType.KERNEL),
vm_utils.ImageType.KERNEL_STR)
def test_from_string(self):
"""Can convert from string to type id."""
self.assertEquals(
vm_utils.ImageType.from_string(vm_utils.ImageType.KERNEL_STR),
vm_utils.ImageType.KERNEL)
class XenAPIDetermineDiskImageTestCase(test.TestCase):

10
nova/tests/vmwareapi/db_fakes.py
View File

@@ -70,8 +70,8 @@ def stub_out_db_instance_api(stubs):
'launch_time': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()),
'instance_type': values['instance_type'],
'memory_mb': type_data['memory_mb'],
'mac_address': values['mac_address'],
'vcpus': type_data['vcpus'],
'mac_addresses': [{'address': values['mac_address']}],
'local_gb': type_data['local_gb'],
}
return FakeModel(base_options)
@@ -83,6 +83,8 @@ def stub_out_db_instance_api(stubs):
'bridge': 'vmnet0',
'netmask': '255.255.255.0',
'gateway': '10.10.10.1',
'broadcast': '10.10.10.255',
'dns1': 'fake',
'vlan': 100}
return FakeModel(fields)
@@ -90,7 +92,7 @@ def stub_out_db_instance_api(stubs):
"""Stubs out the db.instance_action_create method."""
pass
def fake_instance_get_fixed_address(context, instance_id):
def fake_instance_get_fixed_addresses(context, instance_id):
"""Stubs out the db.instance_get_fixed_address method."""
return '10.10.10.10'
@@ -103,7 +105,7 @@ def stub_out_db_instance_api(stubs):
stubs.Set(db, 'instance_create', fake_instance_create)
stubs.Set(db, 'network_get_by_instance', fake_network_get_by_instance)
stubs.Set(db, 'instance_action_create', fake_instance_action_create)
stubs.Set(db, 'instance_get_fixed_address',
fake_instance_get_fixed_address)
stubs.Set(db, 'instance_get_fixed_addresses',
fake_instance_get_fixed_addresses)
stubs.Set(db, 'instance_type_get_all', fake_instance_type_get_all)
stubs.Set(db, 'instance_type_get_by_name', fake_instance_type_get_by_name)

7
nova/tests/vmwareapi/stubs.py
View File

@@ -22,6 +22,8 @@ Stubouts for the test suite
from nova.virt import vmwareapi_conn
from nova.virt.vmwareapi import fake
from nova.virt.vmwareapi import vmware_images
from nova.virt.vmwareapi import vmops
from nova.virt.vmwareapi import network_utils
def fake_get_vim_object(arg):
@@ -36,10 +38,15 @@ def fake_is_vim_object(arg, module):
def set_stubs(stubs):
"""Set the stubs."""
stubs.Set(vmops.VMWareVMOps, 'plug_vifs', fake.fake_plug_vifs)
stubs.Set(network_utils, 'get_network_with_the_name',
fake.fake_get_network)
stubs.Set(vmware_images, 'fetch_image', fake.fake_fetch_image)
stubs.Set(vmware_images, 'get_vmdk_size_and_properties',
fake.fake_get_vmdk_size_and_properties)
stubs.Set(vmware_images, 'upload_image', fake.fake_upload_image)
stubs.Set(vmwareapi_conn.VMWareAPISession, "_get_vim_object",
fake_get_vim_object)
stubs.Set(vmwareapi_conn.VMWareAPISession, "_get_vim_object",
fake_get_vim_object)
stubs.Set(vmwareapi_conn.VMWareAPISession, "_is_vim_object",