x/python-ganttclient
Code Issues Proposed changes

Merging trunk

Browse Source
This commit is contained in:
Rick Harris
2011-09-21 03:33:55 +00:00
parent d29ed7086d 869ba4f976
commit 65b537378f
6 changed files with 249 additions and 293 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Authors
View File

@@ -12,6 +12,7 @@ Armando Migliaccio <Armando.Migliaccio@eu.citrix.com>
Arvind Somya <asomya@cisco.com>
Bilal Akhtar <bilalakhtar@ubuntu.com>
Brad Hall <brad@nicira.com>
Brad McConnell <bmcconne@rackspace.com>
Brian Lamar <brian.lamar@rackspace.com>
Brian Schott <bschott@isi.edu>
Brian Waldon <brian.waldon@rackspace.com>

68
bin/nova-manage
View File

@@ -61,6 +61,7 @@ import math
import netaddr
from optparse import OptionParser
import os
import StringIO
import sys
import time
@@ -274,6 +275,58 @@ class ShellCommands(object):
arguments: path"""
exec(compile(open(path).read(), path, 'exec'), locals(), globals())
@args('--filename', dest='filename', metavar='<path>', default=False,
help='Export file path')
def export(self, filename):
"""Export Nova users into a file that can be consumed by Keystone"""
def create_file(filename):
data = generate_data()
with open(filename, 'w') as f:
f.write(data.getvalue())
def tenants(data, am):
for project in am.get_projects():
print >> data, ("tenant add '%s'" %
(project.name))
for u in project.member_ids:
user = am.get_user(u)
print >> data, ("user add '%s' '%s' '%s'" %
(user.name, user.access, project.name))
print >> data, ("credentials add 'EC2' '%s:%s' '%s' '%s'" %
(user.access, project.id, user.secret, project.id))
def roles(data, am):
for role in am.get_roles():
print >> data, ("role add '%s'" % (role))
def grant_roles(data, am):
roles = am.get_roles()
for project in am.get_projects():
for u in project.member_ids:
user = am.get_user(u)
for role in db.user_get_roles_for_project(ctxt, u,
project.id):
print >> data, ("role grant '%s', '%s', '%s')," %
(user.name, role, project.name))
print >> data
def generate_data():
data = StringIO.StringIO()
am = manager.AuthManager()
tenants(data, am)
roles(data, am)
grant_roles(data, am)
data.seek(0)
return data
ctxt = context.get_admin_context()
if filename:
create_file(filename)
else:
data = generate_data()
print data.getvalue()
class RoleCommands(object):
"""Class for managing roles."""
@@ -685,7 +738,7 @@ class NetworkCommands(object):
help='Multi host')
@args('--dns1', dest="dns1", metavar="<DNS Address>", help='First DNS')
@args('--dns2', dest="dns2", metavar="<DNS Address>", help='Second DNS')
@args('--uuid', dest="net_uuid", metavar="<network uuid>",
@args('--uuid', dest="uuid", metavar="<network uuid>",
help='Network UUID')
@args('--project_id', dest="project_id", metavar="<project id>",
help='Project id')
@@ -710,16 +763,7 @@ class NetworkCommands(object):
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')
raise exception.NetworkNotCreated(req='--bridge')
bridge_interface = bridge_interface or FLAGS.flat_interface or \
FLAGS.vlan_interface
@@ -1648,7 +1692,7 @@ class InstanceTypeCommands(object):
def _print_instance_types(self, name, val):
deleted = ('', ', inactive')[val["deleted"] == 1]
print ("%s: Memory: %sMB, VCPUS: %s, Storage: %sGB, FlavorID: %s, "
"Swap: %sGB, RXTX Quota: %sGB, RXTX Cap: %sMB%s") % (
"Swap: %sMB, RXTX Quota: %sGB, RXTX Cap: %sMB%s") % (
name, val["memory_mb"], val["vcpus"], val["local_gb"],
val["flavorid"], val["swap"], val["rxtx_quota"],
val["rxtx_cap"], deleted)

253
nova/tests/test_compute.py
View File

@@ -21,22 +21,24 @@ Tests For Compute
"""
from nova import compute
from nova.compute import instance_types
from nova.compute import manager as compute_manager
from nova.compute import power_state
from nova.compute import vm_states
from nova import context
from nova import db
from nova.db.sqlalchemy import models
from nova.db.sqlalchemy import api as sqlalchemy_api
from nova import exception
from nova import flags
import nova.image.fake
from nova import log as logging
from nova import rpc
from nova import test
from nova import utils
from nova.compute import instance_types
from nova.compute import manager as compute_manager
from nova.compute import power_state
from nova.compute import vm_states
from nova.db.sqlalchemy import models
from nova.image import fake as fake_image
from nova.notifier import test_notifier
from nova.tests import fake_network
LOG = logging.getLogger('nova.tests.compute')
FLAGS = flags.FLAGS
@@ -74,7 +76,7 @@ class ComputeTestCase(test.TestCase):
def fake_show(meh, context, id):
return {'id': 1, 'properties': {'kernel_id': 1, 'ramdisk_id': 1}}
self.stubs.Set(nova.image.fake._FakeImageService, 'show', fake_show)
self.stubs.Set(fake_image._FakeImageService, 'show', fake_show)
def _create_instance(self, params=None):
"""Create a test instance"""
@@ -1023,190 +1025,19 @@ class ComputeTestCase(test.TestCase):
db.instance_destroy(c, instance_id2)
db.instance_destroy(c, instance_id3)
def test_get_by_fixed_ip(self):
"""Test getting 1 instance by Fixed IP"""
c = context.get_admin_context()
instance_id1 = self._create_instance()
instance_id2 = self._create_instance({'id': 20})
instance_id3 = self._create_instance({'id': 30})
vif_ref1 = db.virtual_interface_create(c,
{'address': '12:34:56:78:90:12',
'instance_id': instance_id1,
'network_id': 1})
vif_ref2 = db.virtual_interface_create(c,
{'address': '90:12:34:56:78:90',
'instance_id': instance_id2,
'network_id': 1})
db.fixed_ip_create(c,
{'address': '1.1.1.1',
'instance_id': instance_id1,
'virtual_interface_id': vif_ref1['id']})
db.fixed_ip_create(c,
{'address': '1.1.2.1',
'instance_id': instance_id2,
'virtual_interface_id': vif_ref2['id']})
# regex not allowed
instances = self.compute_api.get_all(c,
search_opts={'fixed_ip': '.*'})
self.assertEqual(len(instances), 0)
instances = self.compute_api.get_all(c,
search_opts={'fixed_ip': '1.1.3.1'})
self.assertEqual(len(instances), 0)
instances = self.compute_api.get_all(c,
search_opts={'fixed_ip': '1.1.1.1'})
self.assertEqual(len(instances), 1)
self.assertEqual(instances[0].id, instance_id1)
instances = self.compute_api.get_all(c,
search_opts={'fixed_ip': '1.1.2.1'})
self.assertEqual(len(instances), 1)
self.assertEqual(instances[0].id, instance_id2)
db.virtual_interface_delete(c, vif_ref1['id'])
db.virtual_interface_delete(c, vif_ref2['id'])
db.instance_destroy(c, instance_id1)
db.instance_destroy(c, instance_id2)
def test_get_all_by_ip_regexp(self):
"""Test searching by Floating and Fixed IP"""
c = context.get_admin_context()
instance_id1 = self._create_instance({'display_name': 'woot'})
instance_id2 = self._create_instance({
'display_name': 'woo',
'id': 20})
instance_id3 = self._create_instance({
'display_name': 'not-woot',
'id': 30})
vif_ref1 = db.virtual_interface_create(c,
{'address': '12:34:56:78:90:12',
'instance_id': instance_id1,
'network_id': 1})
vif_ref2 = db.virtual_interface_create(c,
{'address': '90:12:34:56:78:90',
'instance_id': instance_id2,
'network_id': 1})
vif_ref3 = db.virtual_interface_create(c,
{'address': '34:56:78:90:12:34',
'instance_id': instance_id3,
'network_id': 1})
db.fixed_ip_create(c,
{'address': '1.1.1.1',
'instance_id': instance_id1,
'virtual_interface_id': vif_ref1['id']})
db.fixed_ip_create(c,
{'address': '1.1.2.1',
'instance_id': instance_id2,
'virtual_interface_id': vif_ref2['id']})
fix_addr = db.fixed_ip_create(c,
{'address': '1.1.3.1',
'instance_id': instance_id3,
'virtual_interface_id': vif_ref3['id']})
fix_ref = db.fixed_ip_get_by_address(c, fix_addr)
flo_ref = db.floating_ip_create(c,
{'address': '10.0.0.2',
'fixed_ip_id': fix_ref['id']})
# ends up matching 2nd octet here.. so all 3 match
instances = self.compute_api.get_all(c,
search_opts={'ip': '.*\.1'})
self.assertEqual(len(instances), 3)
instances = self.compute_api.get_all(c,
search_opts={'ip': '1.*'})
self.assertEqual(len(instances), 3)
instances = self.compute_api.get_all(c,
search_opts={'ip': '.*\.1.\d+$'})
self.assertEqual(len(instances), 1)
instance_ids = [instance.id for instance in instances]
self.assertTrue(instance_id1 in instance_ids)
instances = self.compute_api.get_all(c,
search_opts={'ip': '.*\.2.+'})
self.assertEqual(len(instances), 1)
self.assertEqual(instances[0].id, instance_id2)
instances = self.compute_api.get_all(c,
search_opts={'ip': '10.*'})
self.assertEqual(len(instances), 1)
self.assertEqual(instances[0].id, instance_id3)
db.virtual_interface_delete(c, vif_ref1['id'])
db.virtual_interface_delete(c, vif_ref2['id'])
db.virtual_interface_delete(c, vif_ref3['id'])
db.floating_ip_destroy(c, '10.0.0.2')
db.instance_destroy(c, instance_id1)
db.instance_destroy(c, instance_id2)
db.instance_destroy(c, instance_id3)
def test_get_all_by_ipv6_regexp(self):
"""Test searching by IPv6 address"""
c = context.get_admin_context()
instance_id1 = self._create_instance({'display_name': 'woot'})
instance_id2 = self._create_instance({
'display_name': 'woo',
'id': 20})
instance_id3 = self._create_instance({
'display_name': 'not-woot',
'id': 30})
vif_ref1 = db.virtual_interface_create(c,
{'address': '12:34:56:78:90:12',
'instance_id': instance_id1,
'network_id': 1})
vif_ref2 = db.virtual_interface_create(c,
{'address': '90:12:34:56:78:90',
'instance_id': instance_id2,
'network_id': 1})
vif_ref3 = db.virtual_interface_create(c,
{'address': '34:56:78:90:12:34',
'instance_id': instance_id3,
'network_id': 1})
# This will create IPv6 addresses of:
# 1: fd00::1034:56ff:fe78:9012
# 20: fd00::9212:34ff:fe56:7890
# 30: fd00::3656:78ff:fe90:1234
instances = self.compute_api.get_all(c,
search_opts={'ip6': '.*1034.*'})
self.assertEqual(len(instances), 1)
self.assertEqual(instances[0].id, instance_id1)
instances = self.compute_api.get_all(c,
search_opts={'ip6': '^fd00.*'})
self.assertEqual(len(instances), 3)
instance_ids = [instance.id for instance in instances]
self.assertTrue(instance_id1 in instance_ids)
self.assertTrue(instance_id2 in instance_ids)
self.assertTrue(instance_id3 in instance_ids)
instances = self.compute_api.get_all(c,
search_opts={'ip6': '^.*12.*34.*'})
self.assertEqual(len(instances), 2)
instance_ids = [instance.id for instance in instances]
self.assertTrue(instance_id2 in instance_ids)
self.assertTrue(instance_id3 in instance_ids)
db.virtual_interface_delete(c, vif_ref1['id'])
db.virtual_interface_delete(c, vif_ref2['id'])
db.virtual_interface_delete(c, vif_ref3['id'])
db.instance_destroy(c, instance_id1)
db.instance_destroy(c, instance_id2)
db.instance_destroy(c, instance_id3)
def test_get_all_by_multiple_options_at_once(self):
"""Test searching by multiple options at once"""
c = context.get_admin_context()
instance_id1 = self._create_instance({'display_name': 'woot'})
network_manager = fake_network.FakeNetworkManager()
self.stubs.Set(self.compute_api.network_api,
'get_instance_uuids_by_ip_filter',
network_manager.get_instance_uuids_by_ip_filter)
self.stubs.Set(network_manager.db,
'instance_get_id_to_uuid_mapping',
db.instance_get_id_to_uuid_mapping)
instance_id1 = self._create_instance({'display_name': 'woot',
'id': 0})
instance_id2 = self._create_instance({
'display_name': 'woo',
'id': 20})
@@ -1214,36 +1045,6 @@ class ComputeTestCase(test.TestCase):
'display_name': 'not-woot',
'id': 30})
vif_ref1 = db.virtual_interface_create(c,
{'address': '12:34:56:78:90:12',
'instance_id': instance_id1,
'network_id': 1})
vif_ref2 = db.virtual_interface_create(c,
{'address': '90:12:34:56:78:90',
'instance_id': instance_id2,
'network_id': 1})
vif_ref3 = db.virtual_interface_create(c,
{'address': '34:56:78:90:12:34',
'instance_id': instance_id3,
'network_id': 1})
db.fixed_ip_create(c,
{'address': '1.1.1.1',
'instance_id': instance_id1,
'virtual_interface_id': vif_ref1['id']})
db.fixed_ip_create(c,
{'address': '1.1.2.1',
'instance_id': instance_id2,
'virtual_interface_id': vif_ref2['id']})
fix_addr = db.fixed_ip_create(c,
{'address': '1.1.3.1',
'instance_id': instance_id3,
'virtual_interface_id': vif_ref3['id']})
fix_ref = db.fixed_ip_get_by_address(c, fix_addr)
flo_ref = db.floating_ip_create(c,
{'address': '10.0.0.2',
'fixed_ip_id': fix_ref['id']})
# ip ends up matching 2nd octet here.. so all 3 match ip
# but 'name' only matches one
instances = self.compute_api.get_all(c,
@@ -1251,18 +1052,18 @@ class ComputeTestCase(test.TestCase):
self.assertEqual(len(instances), 1)
self.assertEqual(instances[0].id, instance_id3)
# ip ends up matching any ip with a '2' in it.. so instance
# 2 and 3.. but name should only match #2
# ip ends up matching any ip with a '1' in the last octet..
# so instance 1 and 3.. but name should only match #1
# but 'name' only matches one
instances = self.compute_api.get_all(c,
search_opts={'ip': '.*2', 'name': '^woo.*'})
search_opts={'ip': '.*\.1$', 'name': '^woo.*'})
self.assertEqual(len(instances), 1)
self.assertEqual(instances[0].id, instance_id2)
self.assertEqual(instances[0].id, instance_id1)
# same as above but no match on name (name matches instance_id1
# but the ip query doesn't
instances = self.compute_api.get_all(c,
search_opts={'ip': '.*2.*', 'name': '^woot.*'})
search_opts={'ip': '.*\.2$', 'name': '^woot.*'})
self.assertEqual(len(instances), 0)
# ip matches all 3... ipv6 matches #2+#3...name matches #3
@@ -1273,10 +1074,6 @@ class ComputeTestCase(test.TestCase):
self.assertEqual(len(instances), 1)
self.assertEqual(instances[0].id, instance_id3)
db.virtual_interface_delete(c, vif_ref1['id'])
db.virtual_interface_delete(c, vif_ref2['id'])
db.virtual_interface_delete(c, vif_ref3['id'])
db.floating_ip_destroy(c, '10.0.0.2')
db.instance_destroy(c, instance_id1)
db.instance_destroy(c, instance_id2)
db.instance_destroy(c, instance_id3)

32
nova/tests/test_db_api.py
View File

@@ -18,6 +18,8 @@
"""Unit tests for the DB API"""
import datetime
from nova import test
from nova import context
from nova import db
@@ -92,6 +94,32 @@ class DbApiTestCase(test.TestCase):
db.instance_destroy(self.context, inst1.id)
result = db.instance_get_all_by_filters(self.context.elevated(), {})
self.assertEqual(2, len(result))
self.assertEqual(result[0].id, inst2.id)
self.assertEqual(result[1].id, inst1.id)
self.assertIn(inst1.id, [result[0].id, result[1].id])
self.assertIn(inst2.id, [result[0].id, result[1].id])
if inst1.id == result[0].id:
self.assertTrue(result[0].deleted)
else:
self.assertTrue(result[1].deleted)
def test_migration_get_all_unconfirmed(self):
ctxt = context.get_admin_context()
# Ensure no migrations are returned.
results = db.migration_get_all_unconfirmed(ctxt, 10)
self.assertEqual(0, len(results))
# Ensure one migration older than 10 seconds is returned.
updated_at = datetime.datetime(2000, 01, 01, 12, 00, 00)
values = {"status": "FINISHED", "updated_at": updated_at}
migration = db.migration_create(ctxt, values)
results = db.migration_get_all_unconfirmed(ctxt, 10)
self.assertEqual(1, len(results))
db.migration_update(ctxt, migration.id, {"status": "CONFIRMED"})
# Ensure the new migration is not returned.
updated_at = datetime.datetime.utcnow()
values = {"status": "FINISHED", "updated_at": updated_at}
migration = db.migration_create(ctxt, values)
results = db.migration_get_all_unconfirmed(ctxt, 10)
self.assertEqual(0, len(results))
db.migration_update(ctxt, migration.id, {"status": "CONFIRMED"})

182
nova/tests/test_network.py
View File

@@ -438,55 +438,23 @@ class VlanNetworkTestCase(test.TestCase):
class CommonNetworkTestCase(test.TestCase):
class FakeNetworkManager(network_manager.NetworkManager):
"""This NetworkManager doesn't call the base class so we can bypass all
inherited service cruft and just perform unit tests.
"""
class FakeDB:
def fixed_ip_get_by_instance(self, context, instance_id):
return [dict(address='10.0.0.0'), dict(address='10.0.0.1'),
dict(address='10.0.0.2')]
def network_get_by_cidr(self, context, cidr):
raise exception.NetworkNotFoundForCidr()
def network_create_safe(self, context, net):
fakenet = dict(net)
fakenet['id'] = 999
return fakenet
def network_get_all(self, context):
raise exception.NoNetworksFound()
def __init__(self):
self.db = self.FakeDB()
self.deallocate_called = None
def deallocate_fixed_ip(self, context, address):
self.deallocate_called = address
def _create_fixed_ips(self, context, network_id):
pass
def fake_create_fixed_ips(self, context, network_id):
return None
def test_remove_fixed_ip_from_instance(self):
manager = self.FakeNetworkManager()
manager = fake_network.FakeNetworkManager()
manager.remove_fixed_ip_from_instance(None, 99, '10.0.0.1')
self.assertEquals(manager.deallocate_called, '10.0.0.1')
def test_remove_fixed_ip_from_instance_bad_input(self):
manager = self.FakeNetworkManager()
manager = fake_network.FakeNetworkManager()
self.assertRaises(exception.FixedIpNotFoundForSpecificInstance,
manager.remove_fixed_ip_from_instance,
None, 99, 'bad input')
def test_validate_cidrs(self):
manager = self.FakeNetworkManager()
manager = fake_network.FakeNetworkManager()
nets = manager.create_networks(None, 'fake', '192.168.0.0/24',
False, 1, 256, None, None, None,
None)
@@ -495,7 +463,7 @@ class CommonNetworkTestCase(test.TestCase):
self.assertTrue('192.168.0.0/24' in cidrs)
def test_validate_cidrs_split_exact_in_half(self):
manager = self.FakeNetworkManager()
manager = fake_network.FakeNetworkManager()
nets = manager.create_networks(None, 'fake', '192.168.0.0/24',
False, 2, 128, None, None, None,
None)
@@ -505,7 +473,7 @@ class CommonNetworkTestCase(test.TestCase):
self.assertTrue('192.168.0.128/25' in cidrs)
def test_validate_cidrs_split_cidr_in_use_middle_of_range(self):
manager = self.FakeNetworkManager()
manager = fake_network.FakeNetworkManager()
self.mox.StubOutWithMock(manager.db, 'network_get_all')
ctxt = mox.IgnoreArg()
manager.db.network_get_all(ctxt).AndReturn([{'id': 1,
@@ -523,7 +491,7 @@ class CommonNetworkTestCase(test.TestCase):
self.assertFalse('192.168.2.0/24' in cidrs)
def test_validate_cidrs_smaller_subnet_in_use(self):
manager = self.FakeNetworkManager()
manager = fake_network.FakeNetworkManager()
self.mox.StubOutWithMock(manager.db, 'network_get_all')
ctxt = mox.IgnoreArg()
manager.db.network_get_all(ctxt).AndReturn([{'id': 1,
@@ -536,7 +504,7 @@ class CommonNetworkTestCase(test.TestCase):
self.assertRaises(ValueError, manager.create_networks, *args)
def test_validate_cidrs_split_smaller_cidr_in_use(self):
manager = self.FakeNetworkManager()
manager = fake_network.FakeNetworkManager()
self.mox.StubOutWithMock(manager.db, 'network_get_all')
ctxt = mox.IgnoreArg()
manager.db.network_get_all(ctxt).AndReturn([{'id': 1,
@@ -553,7 +521,7 @@ class CommonNetworkTestCase(test.TestCase):
self.assertFalse('192.168.2.0/24' in cidrs)
def test_validate_cidrs_split_smaller_cidr_in_use2(self):
manager = self.FakeNetworkManager()
manager = fake_network.FakeNetworkManager()
self.mox.StubOutWithMock(manager.db, 'network_get_all')
ctxt = mox.IgnoreArg()
manager.db.network_get_all(ctxt).AndReturn([{'id': 1,
@@ -569,7 +537,7 @@ class CommonNetworkTestCase(test.TestCase):
self.assertFalse('192.168.2.0/27' in cidrs)
def test_validate_cidrs_split_all_in_use(self):
manager = self.FakeNetworkManager()
manager = fake_network.FakeNetworkManager()
self.mox.StubOutWithMock(manager.db, 'network_get_all')
ctxt = mox.IgnoreArg()
in_use = [{'id': 1, 'cidr': '192.168.2.9/29'},
@@ -585,14 +553,14 @@ class CommonNetworkTestCase(test.TestCase):
self.assertRaises(ValueError, manager.create_networks, *args)
def test_validate_cidrs_one_in_use(self):
manager = self.FakeNetworkManager()
manager = fake_network.FakeNetworkManager()
args = (None, 'fake', '192.168.0.0/24', False, 2, 256, None, None,
None, None)
# ValueError: network_size * num_networks exceeds cidr size
self.assertRaises(ValueError, manager.create_networks, *args)
def test_validate_cidrs_already_used(self):
manager = self.FakeNetworkManager()
manager = fake_network.FakeNetworkManager()
self.mox.StubOutWithMock(manager.db, 'network_get_all')
ctxt = mox.IgnoreArg()
manager.db.network_get_all(ctxt).AndReturn([{'id': 1,
@@ -604,7 +572,7 @@ class CommonNetworkTestCase(test.TestCase):
self.assertRaises(ValueError, manager.create_networks, *args)
def test_validate_cidrs_too_many(self):
manager = self.FakeNetworkManager()
manager = fake_network.FakeNetworkManager()
args = (None, 'fake', '192.168.0.0/24', False, 200, 256, None, None,
None, None)
# ValueError: Not enough subnets avail to satisfy requested
@@ -612,7 +580,7 @@ class CommonNetworkTestCase(test.TestCase):
self.assertRaises(ValueError, manager.create_networks, *args)
def test_validate_cidrs_split_partial(self):
manager = self.FakeNetworkManager()
manager = fake_network.FakeNetworkManager()
nets = manager.create_networks(None, 'fake', '192.168.0.0/16',
False, 2, 256, None, None, None, None)
returned_cidrs = [str(net['cidr']) for net in nets]
@@ -620,7 +588,7 @@ class CommonNetworkTestCase(test.TestCase):
self.assertTrue('192.168.1.0/24' in returned_cidrs)
def test_validate_cidrs_conflict_existing_supernet(self):
manager = self.FakeNetworkManager()
manager = fake_network.FakeNetworkManager()
self.mox.StubOutWithMock(manager.db, 'network_get_all')
ctxt = mox.IgnoreArg()
fakecidr = [{'id': 1, 'cidr': '192.168.0.0/8'}]
@@ -634,16 +602,15 @@ class CommonNetworkTestCase(test.TestCase):
def test_create_networks(self):
cidr = '192.168.0.0/24'
manager = self.FakeNetworkManager()
manager = fake_network.FakeNetworkManager()
self.stubs.Set(manager, '_create_fixed_ips',
self.fake_create_fixed_ips)
args = [None, 'foo', cidr, None, 1, 256, 'fd00::/48', None, None,
None]
result = manager.create_networks(*args)
self.assertTrue(manager.create_networks(*args))
def test_create_networks_cidr_already_used(self):
manager = self.FakeNetworkManager()
manager = fake_network.FakeNetworkManager()
self.mox.StubOutWithMock(manager.db, 'network_get_all')
ctxt = mox.IgnoreArg()
fakecidr = [{'id': 1, 'cidr': '192.168.0.0/24'}]
@@ -655,9 +622,124 @@ class CommonNetworkTestCase(test.TestCase):
def test_create_networks_many(self):
cidr = '192.168.0.0/16'
manager = self.FakeNetworkManager()
manager = fake_network.FakeNetworkManager()
self.stubs.Set(manager, '_create_fixed_ips',
self.fake_create_fixed_ips)
args = [None, 'foo', cidr, None, 10, 256, 'fd00::/48', None, None,
None]
self.assertTrue(manager.create_networks(*args))
def test_get_instance_uuids_by_ip_regex(self):
manager = fake_network.FakeNetworkManager()
_vifs = manager.db.virtual_interface_get_all(None)
# Greedy get eveything
res = manager.get_instance_uuids_by_ip_filter(None, {'ip': '.*'})
self.assertEqual(len(res), len(_vifs))
# Doesn't exist
res = manager.get_instance_uuids_by_ip_filter(None, {'ip': '10.0.0.1'})
self.assertFalse(res)
# Get instance 1
res = manager.get_instance_uuids_by_ip_filter(None,
{'ip': '172.16.0.2'})
self.assertTrue(res)
self.assertEqual(len(res), 1)
self.assertEqual(res[0]['instance_id'], _vifs[1]['instance_id'])
# Get instance 2
res = manager.get_instance_uuids_by_ip_filter(None,
{'ip': '173.16.0.2'})
self.assertTrue(res)
self.assertEqual(len(res), 1)
self.assertEqual(res[0]['instance_id'], _vifs[2]['instance_id'])
# Get instance 0 and 1
res = manager.get_instance_uuids_by_ip_filter(None,
{'ip': '172.16.0.*'})
self.assertTrue(res)
self.assertEqual(len(res), 2)
self.assertEqual(res[0]['instance_id'], _vifs[0]['instance_id'])
self.assertEqual(res[1]['instance_id'], _vifs[1]['instance_id'])
# Get instance 1 and 2
res = manager.get_instance_uuids_by_ip_filter(None,
{'ip': '17..16.0.2'})
self.assertTrue(res)
self.assertEqual(len(res), 2)
self.assertEqual(res[0]['instance_id'], _vifs[1]['instance_id'])
self.assertEqual(res[1]['instance_id'], _vifs[2]['instance_id'])
def test_get_instance_uuids_by_ipv6_regex(self):
manager = fake_network.FakeNetworkManager()
_vifs = manager.db.virtual_interface_get_all(None)
# Greedy get eveything
res = manager.get_instance_uuids_by_ip_filter(None, {'ip6': '.*'})
self.assertEqual(len(res), len(_vifs))
# Doesn't exist
res = manager.get_instance_uuids_by_ip_filter(None,
{'ip6': '.*1034.*'})
self.assertFalse(res)
# Get instance 1
res = manager.get_instance_uuids_by_ip_filter(None,
{'ip6': '2001:.*:2'})
self.assertTrue(res)
self.assertEqual(len(res), 1)
self.assertEqual(res[0]['instance_id'], _vifs[1]['instance_id'])
# Get instance 2
ip6 = '2002:db8::dcad:beff:feef:2'
res = manager.get_instance_uuids_by_ip_filter(None, {'ip6': ip6})
self.assertTrue(res)
self.assertEqual(len(res), 1)
self.assertEqual(res[0]['instance_id'], _vifs[2]['instance_id'])
# Get instance 0 and 1
res = manager.get_instance_uuids_by_ip_filter(None, {'ip6': '2001:.*'})
self.assertTrue(res)
self.assertEqual(len(res), 2)
self.assertEqual(res[0]['instance_id'], _vifs[0]['instance_id'])
self.assertEqual(res[1]['instance_id'], _vifs[1]['instance_id'])
# Get instance 1 and 2
ip6 = '200.:db8::dcad:beff:feef:2'
res = manager.get_instance_uuids_by_ip_filter(None, {'ip6': ip6})
self.assertTrue(res)
self.assertEqual(len(res), 2)
self.assertEqual(res[0]['instance_id'], _vifs[1]['instance_id'])
self.assertEqual(res[1]['instance_id'], _vifs[2]['instance_id'])
def test_get_instance_uuids_by_ip(self):
manager = fake_network.FakeNetworkManager()
_vifs = manager.db.virtual_interface_get_all(None)
# No regex for you!
res = manager.get_instance_uuids_by_ip_filter(None,
{'fixed_ip': '.*'})
self.assertFalse(res)
# Doesn't exist
ip = '10.0.0.1'
res = manager.get_instance_uuids_by_ip_filter(None,
{'fixed_ip': ip})
self.assertFalse(res)
# Get instance 1
ip = '172.16.0.2'
res = manager.get_instance_uuids_by_ip_filter(None,
{'fixed_ip': ip})
self.assertTrue(res)
self.assertEqual(len(res), 1)
self.assertEqual(res[0]['instance_id'], _vifs[1]['instance_id'])
# Get instance 2
ip = '173.16.0.2'
res = manager.get_instance_uuids_by_ip_filter(None,
{'fixed_ip': ip})
self.assertTrue(res)
self.assertEqual(len(res), 1)
self.assertEqual(res[0]['instance_id'], _vifs[2]['instance_id'])

4
nova/tests/test_virt_drivers.py
View File

@@ -176,6 +176,10 @@ class _VirtDriverTestCase(test.TestCase):
def test_poll_rescued_instances(self):
self.connection.poll_rescued_instances(10)
@catch_notimplementederror
def test_poll_unconfirmed_resizes(self):
self.connection.poll_unconfirmed_resizes(10)
@catch_notimplementederror
def test_migrate_disk_and_power_off(self):
instance_ref = test_utils.get_test_instance()