summaryrefslogtreecommitdiffstats
path: root/nova/tests
diff options
context:
space:
mode:
Diffstat (limited to 'nova/tests')
-rw-r--r--nova/tests/api/openstack/contrib/test_quotas.py152
-rw-r--r--nova/tests/api/openstack/test_extensions.py24
-rw-r--r--nova/tests/scheduler/test_abstract_scheduler.py (renamed from nova/tests/scheduler/test_zone_aware_scheduler.py)40
-rw-r--r--nova/tests/scheduler/test_least_cost_scheduler.py4
-rw-r--r--nova/tests/scheduler/test_scheduler.py4
-rw-r--r--nova/tests/test_compute.py63
-rw-r--r--nova/tests/test_metadata.py2
-rw-r--r--nova/tests/test_network.py72
-rw-r--r--nova/tests/xenapi/stubs.py4
9 files changed, 306 insertions, 59 deletions
diff --git a/nova/tests/api/openstack/contrib/test_quotas.py b/nova/tests/api/openstack/contrib/test_quotas.py
new file mode 100644
index 000000000..f6a25385f
--- /dev/null
+++ b/nova/tests/api/openstack/contrib/test_quotas.py
@@ -0,0 +1,152 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2011 OpenStack LLC.
+# All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+import json
+import webob
+
+from nova import context
+from nova import test
+from nova.tests.api.openstack import fakes
+
+from nova.api.openstack.contrib.quotas import QuotaSetsController
+
+
+def quota_set(id):
+ return {'quota_set': {'id': id, 'metadata_items': 128, 'volumes': 10,
+ 'gigabytes': 1000, 'ram': 51200, 'floating_ips': 10,
+ 'instances': 10, 'injected_files': 5, 'cores': 20,
+ 'injected_file_content_bytes': 10240}}
+
+
+def quota_set_list():
+ return {'quota_set_list': [quota_set('1234'), quota_set('5678'),
+ quota_set('update_me')]}
+
+
+class QuotaSetsTest(test.TestCase):
+
+ def setUp(self):
+ super(QuotaSetsTest, self).setUp()
+ self.controller = QuotaSetsController()
+ self.user_id = 'fake'
+ self.project_id = 'fake'
+ self.user_context = context.RequestContext(self.user_id,
+ self.project_id)
+ self.admin_context = context.RequestContext(self.user_id,
+ self.project_id,
+ is_admin=True)
+
+ def test_format_quota_set(self):
+ raw_quota_set = {
+ 'instances': 10,
+ 'cores': 20,
+ 'ram': 51200,
+ 'volumes': 10,
+ 'floating_ips': 10,
+ 'metadata_items': 128,
+ 'gigabytes': 1000,
+ 'injected_files': 5,
+ 'injected_file_content_bytes': 10240}
+
+ quota_set = QuotaSetsController()._format_quota_set('1234',
+ raw_quota_set)
+ qs = quota_set['quota_set']
+
+ self.assertEqual(qs['id'], '1234')
+ self.assertEqual(qs['instances'], 10)
+ self.assertEqual(qs['cores'], 20)
+ self.assertEqual(qs['ram'], 51200)
+ self.assertEqual(qs['volumes'], 10)
+ self.assertEqual(qs['gigabytes'], 1000)
+ self.assertEqual(qs['floating_ips'], 10)
+ self.assertEqual(qs['metadata_items'], 128)
+ self.assertEqual(qs['injected_files'], 5)
+ self.assertEqual(qs['injected_file_content_bytes'], 10240)
+
+ def test_quotas_defaults(self):
+ req = webob.Request.blank('/v1.1/os-quota-sets/fake_tenant/defaults')
+ req.method = 'GET'
+ req.headers['Content-Type'] = 'application/json'
+ res = req.get_response(fakes.wsgi_app())
+
+ self.assertEqual(res.status_int, 200)
+ expected = {'quota_set': {
+ 'id': 'fake_tenant',
+ 'instances': 10,
+ 'cores': 20,
+ 'ram': 51200,
+ 'volumes': 10,
+ 'gigabytes': 1000,
+ 'floating_ips': 10,
+ 'metadata_items': 128,
+ 'injected_files': 5,
+ 'injected_file_content_bytes': 10240}}
+
+ self.assertEqual(json.loads(res.body), expected)
+
+ def test_quotas_show_as_admin(self):
+ req = webob.Request.blank('/v1.1/os-quota-sets/1234')
+ req.method = 'GET'
+ req.headers['Content-Type'] = 'application/json'
+ res = req.get_response(fakes.wsgi_app(
+ fake_auth_context=self.admin_context))
+
+ self.assertEqual(res.status_int, 200)
+ self.assertEqual(json.loads(res.body), quota_set('1234'))
+
+ def test_quotas_show_as_unauthorized_user(self):
+ req = webob.Request.blank('/v1.1/os-quota-sets/1234')
+ req.method = 'GET'
+ req.headers['Content-Type'] = 'application/json'
+ res = req.get_response(fakes.wsgi_app(
+ fake_auth_context=self.user_context))
+
+ self.assertEqual(res.status_int, 403)
+
+ def test_quotas_update_as_admin(self):
+ updated_quota_set = {'quota_set': {'instances': 50,
+ 'cores': 50, 'ram': 51200, 'volumes': 10,
+ 'gigabytes': 1000, 'floating_ips': 10,
+ 'metadata_items': 128, 'injected_files': 5,
+ 'injected_file_content_bytes': 10240}}
+
+ req = webob.Request.blank('/v1.1/os-quota-sets/update_me')
+ req.method = 'PUT'
+ req.body = json.dumps(updated_quota_set)
+ req.headers['Content-Type'] = 'application/json'
+
+ res = req.get_response(fakes.wsgi_app(
+ fake_auth_context=self.admin_context))
+
+ self.assertEqual(json.loads(res.body), updated_quota_set)
+
+ def test_quotas_update_as_user(self):
+ updated_quota_set = {'quota_set': {'instances': 50,
+ 'cores': 50, 'ram': 51200, 'volumes': 10,
+ 'gigabytes': 1000, 'floating_ips': 10,
+ 'metadata_items': 128, 'injected_files': 5,
+ 'injected_file_content_bytes': 10240}}
+
+ req = webob.Request.blank('/v1.1/os-quota-sets/update_me')
+ req.method = 'PUT'
+ req.body = json.dumps(updated_quota_set)
+ req.headers['Content-Type'] = 'application/json'
+
+ res = req.get_response(fakes.wsgi_app(
+ fake_auth_context=self.user_context))
+
+ self.assertEqual(res.status_int, 403)
diff --git a/nova/tests/api/openstack/test_extensions.py b/nova/tests/api/openstack/test_extensions.py
index 8a17ee5b2..9c29363c6 100644
--- a/nova/tests/api/openstack/test_extensions.py
+++ b/nova/tests/api/openstack/test_extensions.py
@@ -84,6 +84,19 @@ class ExtensionControllerTest(test.TestCase):
super(ExtensionControllerTest, self).setUp()
ext_path = os.path.join(os.path.dirname(__file__), "extensions")
self.flags(osapi_extensions_path=ext_path)
+ self.ext_list = [
+ "Createserverext"
+ "FlavorExtraSpecs",
+ "Floating_ips",
+ "Fox In Socks",
+ "Hosts",
+ "Keypairs",
+ "Multinic",
+ "Quotas",
+ "SecurityGroups",
+ "Volumes",
+ ]
+ self.ext_list.sort()
def test_list_extensions_json(self):
app = openstack.APIRouterV11()
@@ -96,12 +109,10 @@ class ExtensionControllerTest(test.TestCase):
data = json.loads(response.body)
names = [x['name'] for x in data['extensions']]
names.sort()
- self.assertEqual(names, ["Createserverext", "FlavorExtraSpecs",
- "Floating_ips", "Fox In Socks", "Hosts", "Keypairs", "Multinic",
- "SecurityGroups", "Volumes"])
+ self.assertEqual(names, self.ext_list)
# Make sure that at least Fox in Sox is correct.
- (fox_ext,) = [
+ (fox_ext, ) = [
x for x in data['extensions'] if x['alias'] == 'FOXNSOX']
self.assertEqual(fox_ext, {
'namespace': 'http://www.fox.in.socks/api/ext/pie/v1.0',
@@ -143,10 +154,10 @@ class ExtensionControllerTest(test.TestCase):
# Make sure we have all the extensions.
exts = root.findall('{0}extension'.format(NS))
- self.assertEqual(len(exts), 9)
+ self.assertEqual(len(exts), len(self.ext_list))
# Make sure that at least Fox in Sox is correct.
- (fox_ext,) = [x for x in exts if x.get('alias') == 'FOXNSOX']
+ (fox_ext, ) = [x for x in exts if x.get('alias') == 'FOXNSOX']
self.assertEqual(fox_ext.get('name'), 'Fox In Socks')
self.assertEqual(fox_ext.get('namespace'),
'http://www.fox.in.socks/api/ext/pie/v1.0')
@@ -218,6 +229,7 @@ class ResourceExtensionTest(test.TestCase):
class InvalidExtension(object):
+
def get_alias(self):
return "THIRD"
diff --git a/nova/tests/scheduler/test_zone_aware_scheduler.py b/nova/tests/scheduler/test_abstract_scheduler.py
index 788efca52..f4f5cc233 100644
--- a/nova/tests/scheduler/test_zone_aware_scheduler.py
+++ b/nova/tests/scheduler/test_abstract_scheduler.py
@@ -13,7 +13,7 @@
# License for the specific language governing permissions and limitations
# under the License.
"""
-Tests For Zone Aware Scheduler.
+Tests For Abstract Scheduler.
"""
import json
@@ -25,7 +25,7 @@ from nova import rpc
from nova import test
from nova.compute import api as compute_api
from nova.scheduler import driver
-from nova.scheduler import zone_aware_scheduler
+from nova.scheduler import abstract_scheduler
from nova.scheduler import zone_manager
@@ -60,7 +60,7 @@ def fake_zone_manager_service_states(num_hosts):
return states
-class FakeZoneAwareScheduler(zone_aware_scheduler.ZoneAwareScheduler):
+class FakeAbstractScheduler(abstract_scheduler.AbstractScheduler):
# No need to stub anything at the moment
pass
@@ -161,15 +161,15 @@ def fake_zone_get_all(context):
]
-class ZoneAwareSchedulerTestCase(test.TestCase):
- """Test case for Zone Aware Scheduler."""
+class AbstractSchedulerTestCase(test.TestCase):
+ """Test case for Abstract Scheduler."""
- def test_zone_aware_scheduler(self):
+ def test_abstract_scheduler(self):
"""
Create a nested set of FakeZones, try to build multiple instances
and ensure that a select call returns the appropriate build plan.
"""
- sched = FakeZoneAwareScheduler()
+ sched = FakeAbstractScheduler()
self.stubs.Set(sched, '_call_zone_method', fake_call_zone_method)
self.stubs.Set(nova.db, 'zone_get_all', fake_zone_get_all)
@@ -194,7 +194,7 @@ class ZoneAwareSchedulerTestCase(test.TestCase):
properly adjusted based on the scale/offset in the zone
db entries.
"""
- sched = FakeZoneAwareScheduler()
+ sched = FakeAbstractScheduler()
child_results = fake_call_zone_method(None, None, None, None)
zones = fake_zone_get_all(None)
sched._adjust_child_weights(child_results, zones)
@@ -209,11 +209,11 @@ class ZoneAwareSchedulerTestCase(test.TestCase):
if zone == 'zone3': # Scale x1000
self.assertEqual(scaled.pop(0), w)
- def test_empty_zone_aware_scheduler(self):
+ def test_empty_abstract_scheduler(self):
"""
Ensure empty hosts & child_zones result in NoValidHosts exception.
"""
- sched = FakeZoneAwareScheduler()
+ sched = FakeAbstractScheduler()
self.stubs.Set(sched, '_call_zone_method', fake_empty_call_zone_method)
self.stubs.Set(nova.db, 'zone_get_all', fake_zone_get_all)
@@ -231,7 +231,7 @@ class ZoneAwareSchedulerTestCase(test.TestCase):
If the zone_blob hint was passed in, don't re-schedule.
"""
global was_called
- sched = FakeZoneAwareScheduler()
+ sched = FakeAbstractScheduler()
was_called = False
self.stubs.Set(sched, '_provision_resource', fake_provision_resource)
request_spec = {
@@ -248,7 +248,7 @@ class ZoneAwareSchedulerTestCase(test.TestCase):
def test_provision_resource_local(self):
"""Provision a resource locally or remotely."""
global was_called
- sched = FakeZoneAwareScheduler()
+ sched = FakeAbstractScheduler()
was_called = False
self.stubs.Set(sched, '_provision_resource_locally',
fake_provision_resource_locally)
@@ -260,7 +260,7 @@ class ZoneAwareSchedulerTestCase(test.TestCase):
def test_provision_resource_remote(self):
"""Provision a resource locally or remotely."""
global was_called
- sched = FakeZoneAwareScheduler()
+ sched = FakeAbstractScheduler()
was_called = False
self.stubs.Set(sched, '_provision_resource_from_blob',
fake_provision_resource_from_blob)
@@ -272,9 +272,9 @@ class ZoneAwareSchedulerTestCase(test.TestCase):
def test_provision_resource_from_blob_empty(self):
"""Provision a resource locally or remotely given no hints."""
global was_called
- sched = FakeZoneAwareScheduler()
+ sched = FakeAbstractScheduler()
request_spec = {}
- self.assertRaises(zone_aware_scheduler.InvalidBlob,
+ self.assertRaises(abstract_scheduler.InvalidBlob,
sched._provision_resource_from_blob,
None, {}, 1, {}, {})
@@ -283,7 +283,7 @@ class ZoneAwareSchedulerTestCase(test.TestCase):
Provision a resource locally or remotely when blob hint passed in.
"""
global was_called
- sched = FakeZoneAwareScheduler()
+ sched = FakeAbstractScheduler()
was_called = False
def fake_create_db_entry_for_new_instance(self, context,
@@ -317,7 +317,7 @@ class ZoneAwareSchedulerTestCase(test.TestCase):
passed in.
"""
global was_called
- sched = FakeZoneAwareScheduler()
+ sched = FakeAbstractScheduler()
self.stubs.Set(sched, '_decrypt_blob',
fake_decrypt_blob_returns_child_info)
was_called = False
@@ -336,7 +336,7 @@ class ZoneAwareSchedulerTestCase(test.TestCase):
from an immediate child.
"""
global was_called
- sched = FakeZoneAwareScheduler()
+ sched = FakeAbstractScheduler()
was_called = False
self.stubs.Set(sched, '_ask_child_zone_to_create_instance',
fake_ask_child_zone_to_create_instance)
@@ -350,14 +350,14 @@ class ZoneAwareSchedulerTestCase(test.TestCase):
def test_decrypt_blob(self):
"""Test that the decrypt method works."""
- fixture = FakeZoneAwareScheduler()
+ fixture = FakeAbstractScheduler()
test_data = {"foo": "bar"}
class StubDecryptor(object):
def decryptor(self, key):
return lambda blob: blob
- self.stubs.Set(zone_aware_scheduler, 'crypto',
+ self.stubs.Set(abstract_scheduler, 'crypto',
StubDecryptor())
self.assertEqual(fixture._decrypt_blob(test_data),
diff --git a/nova/tests/scheduler/test_least_cost_scheduler.py b/nova/tests/scheduler/test_least_cost_scheduler.py
index fbe6b2f77..de7581d0a 100644
--- a/nova/tests/scheduler/test_least_cost_scheduler.py
+++ b/nova/tests/scheduler/test_least_cost_scheduler.py
@@ -18,7 +18,7 @@ Tests For Least Cost Scheduler
from nova import test
from nova.scheduler import least_cost
-from nova.tests.scheduler import test_zone_aware_scheduler
+from nova.tests.scheduler import test_abstract_scheduler
MB = 1024 * 1024
@@ -70,7 +70,7 @@ class LeastCostSchedulerTestCase(test.TestCase):
zone_manager = FakeZoneManager()
- states = test_zone_aware_scheduler.fake_zone_manager_service_states(
+ states = test_abstract_scheduler.fake_zone_manager_service_states(
num_hosts=10)
zone_manager.service_states = states
diff --git a/nova/tests/scheduler/test_scheduler.py b/nova/tests/scheduler/test_scheduler.py
index 33461025f..158df2a27 100644
--- a/nova/tests/scheduler/test_scheduler.py
+++ b/nova/tests/scheduler/test_scheduler.py
@@ -257,7 +257,9 @@ class SimpleDriverTestCase(test.TestCase):
def _create_instance(self, **kwargs):
"""Create a test instance"""
inst = {}
- inst['image_id'] = 1
+ # NOTE(jk0): If an integer is passed as the image_ref, the image
+ # service will use the default image service (in this case, the fake).
+ inst['image_ref'] = '1'
inst['reservation_id'] = 'r-fakeres'
inst['user_id'] = self.user_id
inst['project_id'] = self.project_id
diff --git a/nova/tests/test_compute.py b/nova/tests/test_compute.py
index e2fa3b140..4f5d36f14 100644
--- a/nova/tests/test_compute.py
+++ b/nova/tests/test_compute.py
@@ -1341,6 +1341,69 @@ class ComputeTestCase(test.TestCase):
db.instance_destroy(c, instance_id2)
db.instance_destroy(c, instance_id3)
+ def test_get_all_by_metadata(self):
+ """Test searching instances by metadata"""
+
+ c = context.get_admin_context()
+ instance_id0 = self._create_instance()
+ instance_id1 = self._create_instance({
+ 'metadata': {'key1': 'value1'}})
+ instance_id2 = self._create_instance({
+ 'metadata': {'key2': 'value2'}})
+ instance_id3 = self._create_instance({
+ 'metadata': {'key3': 'value3'}})
+ instance_id4 = self._create_instance({
+ 'metadata': {'key3': 'value3',
+ 'key4': 'value4'}})
+
+ # get all instances
+ instances = self.compute_api.get_all(c,
+ search_opts={'metadata': {}})
+ self.assertEqual(len(instances), 5)
+
+ # wrong key/value combination
+ instances = self.compute_api.get_all(c,
+ search_opts={'metadata': {'key1': 'value3'}})
+ self.assertEqual(len(instances), 0)
+
+ # non-existing keys
+ instances = self.compute_api.get_all(c,
+ search_opts={'metadata': {'key5': 'value1'}})
+ self.assertEqual(len(instances), 0)
+
+ # find existing instance
+ instances = self.compute_api.get_all(c,
+ search_opts={'metadata': {'key2': 'value2'}})
+ self.assertEqual(len(instances), 1)
+ self.assertEqual(instances[0].id, instance_id2)
+
+ instances = self.compute_api.get_all(c,
+ search_opts={'metadata': {'key3': 'value3'}})
+ self.assertEqual(len(instances), 2)
+ instance_ids = [instance.id for instance in instances]
+ self.assertTrue(instance_id3 in instance_ids)
+ self.assertTrue(instance_id4 in instance_ids)
+
+ # multiple criterias as a dict
+ instances = self.compute_api.get_all(c,
+ search_opts={'metadata': {'key3': 'value3',
+ 'key4': 'value4'}})
+ self.assertEqual(len(instances), 1)
+ self.assertEqual(instances[0].id, instance_id4)
+
+ # multiple criterias as a list
+ instances = self.compute_api.get_all(c,
+ search_opts={'metadata': [{'key4': 'value4'},
+ {'key3': 'value3'}]})
+ self.assertEqual(len(instances), 1)
+ self.assertEqual(instances[0].id, instance_id4)
+
+ db.instance_destroy(c, instance_id0)
+ db.instance_destroy(c, instance_id1)
+ db.instance_destroy(c, instance_id2)
+ db.instance_destroy(c, instance_id3)
+ db.instance_destroy(c, instance_id4)
+
@staticmethod
def _parse_db_block_device_mapping(bdm_ref):
attr_list = ('delete_on_termination', 'device_name', 'no_device',
diff --git a/nova/tests/test_metadata.py b/nova/tests/test_metadata.py
index ad678714e..bfc7a6d44 100644
--- a/nova/tests/test_metadata.py
+++ b/nova/tests/test_metadata.py
@@ -39,7 +39,7 @@ class MetadataTestCase(test.TestCase):
'key_name': None,
'host': 'test',
'launch_index': 1,
- 'instance_type': 'm1.tiny',
+ 'instance_type': {'name': 'm1.tiny'},
'reservation_id': 'r-xxxxxxxx',
'user_data': '',
'image_ref': 7,
diff --git a/nova/tests/test_network.py b/nova/tests/test_network.py
index 1f01695a8..581a4765c 100644
--- a/nova/tests/test_network.py
+++ b/nova/tests/test_network.py
@@ -376,7 +376,7 @@ class CommonNetworkTestCase(test.TestCase):
raise exception.NetworkNotFoundForCidr()
def network_create_safe(self, context, net):
- fakenet = {}
+ fakenet = dict(net)
fakenet['id'] = 999
return fakenet
@@ -390,6 +390,9 @@ class CommonNetworkTestCase(test.TestCase):
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
@@ -407,16 +410,20 @@ class CommonNetworkTestCase(test.TestCase):
def test_validate_cidrs(self):
manager = self.FakeNetworkManager()
- nets = manager._validate_cidrs(None, '192.168.0.0/24', 1, 256)
+ nets = manager.create_networks(None, 'fake', '192.168.0.0/24',
+ False, 1, 256, None, None, None,
+ None)
self.assertEqual(1, len(nets))
- cidrs = [str(net) for net in nets]
+ cidrs = [str(net['cidr']) for net in nets]
self.assertTrue('192.168.0.0/24' in cidrs)
def test_validate_cidrs_split_exact_in_half(self):
manager = self.FakeNetworkManager()
- nets = manager._validate_cidrs(None, '192.168.0.0/24', 2, 128)
+ nets = manager.create_networks(None, 'fake', '192.168.0.0/24',
+ False, 2, 128, None, None, None,
+ None)
self.assertEqual(2, len(nets))
- cidrs = [str(net) for net in nets]
+ cidrs = [str(net['cidr']) for net in nets]
self.assertTrue('192.168.0.0/25' in cidrs)
self.assertTrue('192.168.0.128/25' in cidrs)
@@ -427,9 +434,11 @@ class CommonNetworkTestCase(test.TestCase):
manager.db.network_get_all(ctxt).AndReturn([{'id': 1,
'cidr': '192.168.2.0/24'}])
self.mox.ReplayAll()
- nets = manager._validate_cidrs(None, '192.168.0.0/16', 4, 256)
+ nets = manager.create_networks(None, 'fake', '192.168.0.0/16',
+ False, 4, 256, None, None, None,
+ None)
self.assertEqual(4, len(nets))
- cidrs = [str(net) for net in nets]
+ cidrs = [str(net['cidr']) for net in nets]
exp_cidrs = ['192.168.0.0/24', '192.168.1.0/24', '192.168.3.0/24',
'192.168.4.0/24']
for exp_cidr in exp_cidrs:
@@ -445,8 +454,9 @@ class CommonNetworkTestCase(test.TestCase):
self.mox.ReplayAll()
# ValueError: requested cidr (192.168.2.0/24) conflicts with
# existing smaller cidr
- args = [None, '192.168.2.0/24', 1, 256]
- self.assertRaises(ValueError, manager._validate_cidrs, *args)
+ args = (None, 'fake', '192.168.2.0/24', False, 1, 256, None, None,
+ None, None)
+ self.assertRaises(ValueError, manager.create_networks, *args)
def test_validate_cidrs_split_smaller_cidr_in_use(self):
manager = self.FakeNetworkManager()
@@ -455,9 +465,10 @@ class CommonNetworkTestCase(test.TestCase):
manager.db.network_get_all(ctxt).AndReturn([{'id': 1,
'cidr': '192.168.2.0/25'}])
self.mox.ReplayAll()
- nets = manager._validate_cidrs(None, '192.168.0.0/16', 4, 256)
+ nets = manager.create_networks(None, 'fake', '192.168.0.0/16',
+ False, 4, 256, None, None, None, None)
self.assertEqual(4, len(nets))
- cidrs = [str(net) for net in nets]
+ cidrs = [str(net['cidr']) for net in nets]
exp_cidrs = ['192.168.0.0/24', '192.168.1.0/24', '192.168.3.0/24',
'192.168.4.0/24']
for exp_cidr in exp_cidrs:
@@ -471,9 +482,10 @@ class CommonNetworkTestCase(test.TestCase):
manager.db.network_get_all(ctxt).AndReturn([{'id': 1,
'cidr': '192.168.2.9/29'}])
self.mox.ReplayAll()
- nets = manager._validate_cidrs(None, '192.168.2.0/24', 3, 32)
+ nets = manager.create_networks(None, 'fake', '192.168.2.0/24',
+ False, 3, 32, None, None, None, None)
self.assertEqual(3, len(nets))
- cidrs = [str(net) for net in nets]
+ cidrs = [str(net['cidr']) for net in nets]
exp_cidrs = ['192.168.2.32/27', '192.168.2.64/27', '192.168.2.96/27']
for exp_cidr in exp_cidrs:
self.assertTrue(exp_cidr in cidrs)
@@ -488,17 +500,19 @@ class CommonNetworkTestCase(test.TestCase):
{'id': 3, 'cidr': '192.168.2.128/26'}]
manager.db.network_get_all(ctxt).AndReturn(in_use)
self.mox.ReplayAll()
- args = [None, '192.168.2.0/24', 3, 64]
+ args = (None, 'fake', '192.168.2.0/24', False, 3, 64, None, None,
+ None, None)
# ValueError: Not enough subnets avail to satisfy requested num_
# networks - some subnets in requested range already
# in use
- self.assertRaises(ValueError, manager._validate_cidrs, *args)
+ self.assertRaises(ValueError, manager.create_networks, *args)
def test_validate_cidrs_one_in_use(self):
manager = self.FakeNetworkManager()
- args = [None, '192.168.0.0/24', 2, 256]
+ 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._validate_cidrs, *args)
+ self.assertRaises(ValueError, manager.create_networks, *args)
def test_validate_cidrs_already_used(self):
manager = self.FakeNetworkManager()
@@ -508,20 +522,23 @@ class CommonNetworkTestCase(test.TestCase):
'cidr': '192.168.0.0/24'}])
self.mox.ReplayAll()
# ValueError: cidr already in use
- args = [None, '192.168.0.0/24', 1, 256]
- self.assertRaises(ValueError, manager._validate_cidrs, *args)
+ args = (None, 'fake', '192.168.0.0/24', False, 1, 256, None, None,
+ None, None)
+ self.assertRaises(ValueError, manager.create_networks, *args)
def test_validate_cidrs_too_many(self):
manager = self.FakeNetworkManager()
- args = [None, '192.168.0.0/24', 200, 256]
+ args = (None, 'fake', '192.168.0.0/24', False, 200, 256, None, None,
+ None, None)
# ValueError: Not enough subnets avail to satisfy requested
# num_networks
- self.assertRaises(ValueError, manager._validate_cidrs, *args)
+ self.assertRaises(ValueError, manager.create_networks, *args)
def test_validate_cidrs_split_partial(self):
manager = self.FakeNetworkManager()
- nets = manager._validate_cidrs(None, '192.168.0.0/16', 2, 256)
- returned_cidrs = [str(net) for net in nets]
+ 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]
self.assertTrue('192.168.0.0/24' in returned_cidrs)
self.assertTrue('192.168.1.0/24' in returned_cidrs)
@@ -532,10 +549,11 @@ class CommonNetworkTestCase(test.TestCase):
fakecidr = [{'id': 1, 'cidr': '192.168.0.0/8'}]
manager.db.network_get_all(ctxt).AndReturn(fakecidr)
self.mox.ReplayAll()
- args = [None, '192.168.0.0/24', 1, 256]
+ args = (None, 'fake', '192.168.0.0/24', False, 1, 256, None, None,
+ None, None)
# ValueError: requested cidr (192.168.0.0/24) conflicts
# with existing supernet
- self.assertRaises(ValueError, manager._validate_cidrs, *args)
+ self.assertRaises(ValueError, manager.create_networks, *args)
def test_create_networks(self):
cidr = '192.168.0.0/24'
@@ -545,7 +563,7 @@ class CommonNetworkTestCase(test.TestCase):
args = [None, 'foo', cidr, None, 1, 256, 'fd00::/48', None, None,
None]
result = manager.create_networks(*args)
- self.assertEqual(manager.create_networks(*args), None)
+ self.assertTrue(manager.create_networks(*args))
def test_create_networks_cidr_already_used(self):
manager = self.FakeNetworkManager()
@@ -565,4 +583,4 @@ class CommonNetworkTestCase(test.TestCase):
self.fake_create_fixed_ips)
args = [None, 'foo', cidr, None, 10, 256, 'fd00::/48', None, None,
None]
- self.assertEqual(manager.create_networks(*args), None)
+ self.assertTrue(manager.create_networks(*args))
diff --git a/nova/tests/xenapi/stubs.py b/nova/tests/xenapi/stubs.py
index 0d0f84e32..a6a1febd6 100644
--- a/nova/tests/xenapi/stubs.py
+++ b/nova/tests/xenapi/stubs.py
@@ -28,10 +28,10 @@ from nova import utils
def stubout_instance_snapshot(stubs):
@classmethod
- def fake_fetch_image(cls, context, session, instance_id, image, user,
+ def fake_fetch_image(cls, context, session, instance, image, user,
project, type):
from nova.virt.xenapi.fake import create_vdi
- name_label = "instance-%s" % instance_id
+ name_label = "instance-%s" % instance.id
#TODO: create fake SR record
sr_ref = "fakesr"
vdi_ref = create_vdi(name_label=name_label, read_only=False,
generated by cgit (git 2.34.1) at 2026-04-24 15:00:31 +0000