x/python-ganttclient
Code Issues Proposed changes

merged trunk

Browse Source
This commit is contained in:
Brian Lamar
2011-08-11 11:28:11 -04:00
parent 799f9cd875 1b6b8e12c1
commit 3ba12c6790
12 changed files with 516 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.mailmap
View File

@@ -18,6 +18,8 @@
<devin.carlen@gmail.com> <devcamcar@illian.local>
<ewan.mellor@citrix.com> <emellor@silver>
<itoumsn@nttdata.co.jp> <itoumsn@shayol>
<jake@ansolabs.com> <jake@markupisart.com>
<jake@ansolabs.com> <admin@jakedahn.com>
<jaypipes@gmail.com> <jpipes@serialcoder>
<jmckenty@gmail.com> <jmckenty@joshua-mckentys-macbook-pro.local>
<jmckenty@gmail.com> <jmckenty@yyj-dhcp171.corp.flock.com>

1
Authors
View File

@@ -37,6 +37,7 @@ Hisaharu Ishii <ishii.hisaharu@lab.ntt.co.jp>
Hisaki Ohara <hisaki.ohara@intel.com>
Ilya Alekseyev <ilyaalekseyev@acm.org>
Isaku Yamahata <yamahata@valinux.co.jp>
Jake Dahn <jake@ansolabs.com>
Jason Cannavale <jason.cannavale@rackspace.com>
Jason Koelker <jason@koelker.net>
Jay Pipes <jaypipes@gmail.com>

110
bin/nova-import-canonical-imagestore
View File

@@ -1,110 +0,0 @@
#!/usr/bin/env python
# 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.
"""
Download images from Canonical Image Store
"""
import gettext
import json
import os
import tempfile
import shutil
import subprocess
import sys
import urllib2
# If ../nova/__init__.py exists, add ../ to Python search path, so that
# it will override what happens to be installed in /usr/(local/)lib/python...
possible_topdir = os.path.normpath(os.path.join(os.path.abspath(sys.argv[0]),
os.pardir,
os.pardir))
if os.path.exists(os.path.join(possible_topdir, 'nova', '__init__.py')):
sys.path.insert(0, possible_topdir)
gettext.install('nova', unicode=1)
from nova import flags
from nova import log as logging
from nova import utils
from nova.objectstore import image
FLAGS = flags.FLAGS
API_URL = 'https://imagestore.canonical.com/api/dashboard'
def get_images():
"""Get a list of the images from the imagestore URL."""
images = json.load(urllib2.urlopen(API_URL))['images']
images = [img for img in images if img['title'].find('amd64') > -1]
return images
def download(img):
"""Download an image to the local filesystem."""
# FIXME(ja): add checksum/signature checks
tempdir = tempfile.mkdtemp(prefix='cis-')
kernel_id = None
ramdisk_id = None
for f in img['files']:
if f['kind'] == 'kernel':
dest = os.path.join(tempdir, 'kernel')
subprocess.call(['curl', '--fail', f['url'], '-o', dest])
kernel_id = image.Image.add(dest,
description='kernel/' + img['title'], kernel=True)
for f in img['files']:
if f['kind'] == 'ramdisk':
dest = os.path.join(tempdir, 'ramdisk')
subprocess.call(['curl', '--fail', f['url'], '-o', dest])
ramdisk_id = image.Image.add(dest,
description='ramdisk/' + img['title'], ramdisk=True)
for f in img['files']:
if f['kind'] == 'image':
dest = os.path.join(tempdir, 'image')
subprocess.call(['curl', '--fail', f['url'], '-o', dest])
ramdisk_id = image.Image.add(dest,
description=img['title'], kernel=kernel_id, ramdisk=ramdisk_id)
shutil.rmtree(tempdir)
def main():
"""Main entry point."""
utils.default_flagfile()
argv = FLAGS(sys.argv)
logging.setup()
images = get_images()
if len(argv) == 2:
for img in images:
if argv[1] == 'all' or argv[1] == img['title']:
download(img)
else:
print 'usage: %s (title|all)'
print 'available images:'
for img in images:
print img['title']
if __name__ == '__main__':
main()

3
nova/flags.py
View File

@@ -398,3 +398,6 @@ DEFINE_bool('start_guests_on_host_boot', False,
'Whether to restart guests when the host reboots')
DEFINE_bool('resume_guests_state_on_host_boot', False,
'Whether to start guests, that was running before the host reboot')
DEFINE_string('root_helper', 'sudo',
'Command prefix to use for running commands as root')

3
nova/scheduler/least_cost.py
View File

@@ -96,7 +96,8 @@ class LeastCostScheduler(zone_aware_scheduler.ZoneAwareScheduler):
cost_fn_str=cost_fn_str)
try:
weight = getattr(FLAGS, "%s_weight" % cost_fn.__name__)
flag_name = "%s_weight" % cost_fn.__name__
weight = getattr(FLAGS, flag_name)
except AttributeError:
raise exception.SchedulerWeightFlagNotFound(
flag_name=flag_name)

4
nova/scheduler/zone_aware_scheduler.py
View File

@@ -266,8 +266,8 @@ class ZoneAwareScheduler(driver.Scheduler):
"""
if topic != "compute":
raise NotImplemented(_("Zone Aware Scheduler only understands "
"Compute nodes (for now)"))
raise NotImplementedError(_("Zone Aware Scheduler only understands"
" Compute nodes (for now)"))
num_instances = request_spec.get('num_instances', 1)
instance_type = request_spec['instance_type']

2
nova/scheduler/zone_manager.py
View File

@@ -198,7 +198,7 @@ class ZoneManager(object):
def update_service_capabilities(self, service_name, host, capabilities):
"""Update the per-service capabilities based on this notification."""
logging.debug(_("Received %(service_name)s service update from "
"%(host)s: %(capabilities)s") % locals())
"%(host)s.") % locals())
service_caps = self.service_states.get(host, {})
capabilities["timestamp"] = utils.utcnow() # Reported time
service_caps[service_name] = capabilities

7
nova/tests/test_auth.py
View File

@@ -62,7 +62,12 @@ class project_generator(object):
class user_and_project_generator(object):
def __init__(self, manager, user_state={}, project_state={}):
def __init__(self, manager, user_state=None, project_state=None):
if not user_state:
user_state = {}
if not project_state:
project_state = {}
self.manager = manager
if 'name' not in user_state:
user_state['name'] = 'test1'

463
nova/tests/test_compute.py
View File

@@ -26,6 +26,7 @@ from nova.compute import power_state
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
@@ -73,8 +74,11 @@ class ComputeTestCase(test.TestCase):
self.stubs.Set(nova.image.fake._FakeImageService, 'show', fake_show)
def _create_instance(self, params={}):
def _create_instance(self, params=None):
"""Create a test instance"""
if not params:
params = {}
inst = {}
inst['image_ref'] = 1
inst['reservation_id'] = 'r-fakeres'
@@ -87,8 +91,11 @@ class ComputeTestCase(test.TestCase):
inst.update(params)
return db.instance_create(self.context, inst)['id']
def _create_instance_type(self, params={}):
def _create_instance_type(self, params=None):
"""Create a test instance"""
if not params:
params = {}
context = self.context.elevated()
inst = {}
inst['name'] = 'm1.small'
@@ -864,6 +871,458 @@ class ComputeTestCase(test.TestCase):
self.assertEqual(len(instances), 1)
self.assertEqual(power_state.SHUTOFF, instances[0]['state'])
def test_get_all_by_name_regexp(self):
"""Test searching instances by name (display_name)"""
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})
instances = self.compute_api.get_all(c,
search_opts={'name': 'woo.*'})
self.assertEqual(len(instances), 2)
instance_ids = [instance.id for instance in instances]
self.assertTrue(instance_id1 in instance_ids)
self.assertTrue(instance_id2 in instance_ids)
instances = self.compute_api.get_all(c,
search_opts={'name': 'woot.*'})
instance_ids = [instance.id for instance in instances]
self.assertEqual(len(instances), 1)
self.assertTrue(instance_id1 in instance_ids)
instances = self.compute_api.get_all(c,
search_opts={'name': '.*oot.*'})
self.assertEqual(len(instances), 2)
instance_ids = [instance.id for instance in instances]
self.assertTrue(instance_id1 in instance_ids)
self.assertTrue(instance_id3 in instance_ids)
instances = self.compute_api.get_all(c,
search_opts={'name': 'n.*'})
self.assertEqual(len(instances), 1)
instance_ids = [instance.id for instance in instances]
self.assertTrue(instance_id3 in instance_ids)
instances = self.compute_api.get_all(c,
search_opts={'name': 'noth.*'})
self.assertEqual(len(instances), 0)
db.instance_destroy(c, instance_id1)
db.instance_destroy(c, instance_id2)
db.instance_destroy(c, instance_id3)
def test_get_all_by_instance_name_regexp(self):
"""Test searching instances by name"""
self.flags(instance_name_template='instance-%d')
c = context.get_admin_context()
instance_id1 = self._create_instance()
instance_id2 = self._create_instance({'id': 2})
instance_id3 = self._create_instance({'id': 10})
instances = self.compute_api.get_all(c,
search_opts={'instance_name': 'instance.*'})
self.assertEqual(len(instances), 3)
instances = self.compute_api.get_all(c,
search_opts={'instance_name': '.*\-\d$'})
self.assertEqual(len(instances), 2)
instance_ids = [instance.id for instance in instances]
self.assertTrue(instance_id1 in instance_ids)
self.assertTrue(instance_id2 in instance_ids)
instances = self.compute_api.get_all(c,
search_opts={'instance_name': 'i.*2'})
self.assertEqual(len(instances), 1)
self.assertEqual(instances[0].id, instance_id2)
db.instance_destroy(c, instance_id1)
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'})
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']})
# ip ends up matching 2nd octet here.. so all 3 match ip
# but 'name' only matches one
instances = self.compute_api.get_all(c,
search_opts={'ip': '.*\.1', 'name': 'not.*'})
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
# but 'name' only matches one
instances = self.compute_api.get_all(c,
search_opts={'ip': '.*2', 'name': '^woo.*'})
self.assertEqual(len(instances), 1)
self.assertEqual(instances[0].id, instance_id2)
# 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.*'})
self.assertEqual(len(instances), 0)
# ip matches all 3... ipv6 matches #2+#3...name matches #3
instances = self.compute_api.get_all(c,
search_opts={'ip': '.*\.1',
'name': 'not.*',
'ip6': '^.*12.*34.*'})
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_image(self):
"""Test searching instances by image"""
c = context.get_admin_context()
instance_id1 = self._create_instance({'image_ref': '1234'})
instance_id2 = self._create_instance({
'id': 2,
'image_ref': '4567'})
instance_id3 = self._create_instance({
'id': 10,
'image_ref': '4567'})
instances = self.compute_api.get_all(c,
search_opts={'image': '123'})
self.assertEqual(len(instances), 0)
instances = self.compute_api.get_all(c,
search_opts={'image': '1234'})
self.assertEqual(len(instances), 1)
self.assertEqual(instances[0].id, instance_id1)
instances = self.compute_api.get_all(c,
search_opts={'image': '4567'})
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)
# Test passing a list as search arg
instances = self.compute_api.get_all(c,
search_opts={'image': ['1234', '4567']})
self.assertEqual(len(instances), 3)
db.instance_destroy(c, instance_id1)
db.instance_destroy(c, instance_id2)
db.instance_destroy(c, instance_id3)
def test_get_all_by_flavor(self):
"""Test searching instances by image"""
c = context.get_admin_context()
instance_id1 = self._create_instance({'instance_type_id': 1})
instance_id2 = self._create_instance({
'id': 2,
'instance_type_id': 2})
instance_id3 = self._create_instance({
'id': 10,
'instance_type_id': 2})
# NOTE(comstud): Migrations set up the instance_types table
# for us. Therefore, we assume the following is true for
# these tests:
# instance_type_id 1 == flavor 3
# instance_type_id 2 == flavor 1
# instance_type_id 3 == flavor 4
# instance_type_id 4 == flavor 5
# instance_type_id 5 == flavor 2
instances = self.compute_api.get_all(c,
search_opts={'flavor': 5})
self.assertEqual(len(instances), 0)
self.assertRaises(exception.FlavorNotFound,
self.compute_api.get_all,
c, search_opts={'flavor': 99})
instances = self.compute_api.get_all(c,
search_opts={'flavor': 3})
self.assertEqual(len(instances), 1)
self.assertEqual(instances[0].id, instance_id1)
instances = self.compute_api.get_all(c,
search_opts={'flavor': 1})
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.instance_destroy(c, instance_id1)
db.instance_destroy(c, instance_id2)
db.instance_destroy(c, instance_id3)
def test_get_all_by_state(self):
"""Test searching instances by state"""
c = context.get_admin_context()
instance_id1 = self._create_instance({'state': power_state.SHUTDOWN})
instance_id2 = self._create_instance({
'id': 2,
'state': power_state.RUNNING})
instance_id3 = self._create_instance({
'id': 10,
'state': power_state.RUNNING})
instances = self.compute_api.get_all(c,
search_opts={'state': power_state.SUSPENDED})
self.assertEqual(len(instances), 0)
instances = self.compute_api.get_all(c,
search_opts={'state': power_state.SHUTDOWN})
self.assertEqual(len(instances), 1)
self.assertEqual(instances[0].id, instance_id1)
instances = self.compute_api.get_all(c,
search_opts={'state': power_state.RUNNING})
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)
# Test passing a list as search arg
instances = self.compute_api.get_all(c,
search_opts={'state': [power_state.SHUTDOWN,
power_state.RUNNING]})
self.assertEqual(len(instances), 3)
db.instance_destroy(c, instance_id1)
db.instance_destroy(c, instance_id2)
db.instance_destroy(c, instance_id3)
@staticmethod
def _parse_db_block_device_mapping(bdm_ref):
attr_list = ('delete_on_termination', 'device_name', 'no_device',

15
nova/tests/test_libvirt.py
View File

@@ -921,18 +921,18 @@ class IptablesFirewallTestCase(test.TestCase):
# self.fw.add_instance(instance_ref)
def fake_iptables_execute(*cmd, **kwargs):
process_input = kwargs.get('process_input', None)
if cmd == ('sudo', 'ip6tables-save', '-t', 'filter'):
if cmd == ('ip6tables-save', '-t', 'filter'):
return '\n'.join(self.in6_filter_rules), None
if cmd == ('sudo', 'iptables-save', '-t', 'filter'):
if cmd == ('iptables-save', '-t', 'filter'):
return '\n'.join(self.in_filter_rules), None
if cmd == ('sudo', 'iptables-save', '-t', 'nat'):
if cmd == ('iptables-save', '-t', 'nat'):
return '\n'.join(self.in_nat_rules), None
if cmd == ('sudo', 'iptables-restore'):
if cmd == ('iptables-restore',):
lines = process_input.split('\n')
if '*filter' in lines:
self.out_rules = lines
return '', ''
if cmd == ('sudo', 'ip6tables-restore'):
if cmd == ('ip6tables-restore',):
lines = process_input.split('\n')
if '*filter' in lines:
self.out6_rules = lines
@@ -1194,8 +1194,11 @@ class NWFilterTestCase(test.TestCase):
'project_id': 'fake',
'instance_type_id': 1})
def _create_instance_type(self, params={}):
def _create_instance_type(self, params=None):
"""Create a test instance"""
if not params:
params = {}
context = self.context.elevated()
inst = {}
inst['name'] = 'm1.small'

10
nova/tests/test_volume.py
View File

@@ -414,8 +414,9 @@ class ISCSITestCase(DriverTestCase):
self.mox.StubOutWithMock(self.volume.driver, '_execute')
for i in volume_id_list:
tid = db.volume_get_iscsi_target_num(self.context, i)
self.volume.driver._execute("sudo", "ietadm", "--op", "show",
"--tid=%(tid)d" % locals())
self.volume.driver._execute("ietadm", "--op", "show",
"--tid=%(tid)d" % locals(),
run_as_root=True)
self.stream.truncate(0)
self.mox.ReplayAll()
@@ -433,8 +434,9 @@ class ISCSITestCase(DriverTestCase):
# the first vblade process isn't running
tid = db.volume_get_iscsi_target_num(self.context, volume_id_list[0])
self.mox.StubOutWithMock(self.volume.driver, '_execute')
self.volume.driver._execute("sudo", "ietadm", "--op", "show",
"--tid=%(tid)d" % locals()).AndRaise(
self.volume.driver._execute("ietadm", "--op", "show",
"--tid=%(tid)d" % locals(),
run_as_root=True).AndRaise(
exception.ProcessExecutionError())
self.mox.ReplayAll()

28
nova/tests/test_xenapi.py
View File

@@ -548,8 +548,8 @@ class XenAPIVMTestCase(test.TestCase):
return '', ''
fake_utils.fake_execute_set_repliers([
# Capture the sudo tee .../etc/network/interfaces command
(r'(sudo\s+)?tee.*interfaces', _tee_handler),
# Capture the tee .../etc/network/interfaces command
(r'tee.*interfaces', _tee_handler),
])
self._test_spawn(glance_stubs.FakeGlance.IMAGE_MACHINE,
glance_stubs.FakeGlance.IMAGE_KERNEL,
@@ -592,9 +592,9 @@ class XenAPIVMTestCase(test.TestCase):
return '', ''
fake_utils.fake_execute_set_repliers([
(r'(sudo\s+)?mount', _mount_handler),
(r'(sudo\s+)?umount', _umount_handler),
(r'(sudo\s+)?tee.*interfaces', _tee_handler)])
(r'mount', _mount_handler),
(r'umount', _umount_handler),
(r'tee.*interfaces', _tee_handler)])
self._test_spawn(1, 2, 3, check_injection=True)
# tee must not run in this case, where an injection-capable
@@ -654,6 +654,24 @@ class XenAPIVMTestCase(test.TestCase):
# Ensure that it will not unrescue a non-rescued instance.
self.assertRaises(Exception, conn.unrescue, instance, None)
def test_revert_migration(self):
instance = self._create_instance()
class VMOpsMock():
def __init__(self):
self.revert_migration_called = False
def revert_migration(self, instance):
self.revert_migration_called = True
stubs.stubout_session(self.stubs, stubs.FakeSessionForMigrationTests)
conn = xenapi_conn.get_connection(False)
conn._vmops = VMOpsMock()
conn.revert_migration(instance)
self.assertTrue(conn._vmops.revert_migration_called)
def _create_instance(self, instance_id=1, spawn=True):
"""Creates and spawns a test instance."""
stubs.stubout_loopingcall_start(self.stubs)