x/python-ganttclient
Code Issues Proposed changes

pep8 fixes

Browse Source
This commit is contained in:
Todd Willey
2011-01-03 14:29:39 -05:00
parent 022ba7b277 8215a6b6ea
commit 872c11fa29
9 changed files with 229 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.bzrignore
View File

@@ -6,6 +6,7 @@ keys
networks
nova.sqlite
CA/cacert.pem
CA/crl.pem
CA/index.txt*
CA/openssl.cnf
CA/serial*

1
.mailmap
View File

@@ -27,5 +27,6 @@
<vishvananda@gmail.com> <root@ubuntu>
<sleepsonthefloor@gmail.com> <root@tonbuntu>
<rlane@wikimedia.org> <laner@controller>
<rconradharris@gmail.com> <rick.harris@rackspace.com>
<corywright@gmail.com> <cory.wright@rackspace.com>
<ant@openstack.org> <amesserl@rackspace.com>

1
Authors
View File

@@ -27,6 +27,7 @@ Michael Gundlach <michael.gundlach@rackspace.com>
Monty Taylor <mordred@inaugust.com>
Paul Voccio <paul@openstack.org>
Rick Clark <rick@openstack.org>
Rick Harris <rconradharris@gmail.com>
Ryan Lane <rlane@wikimedia.org>
Ryan Lucio <rlucio@internap.com>
Salvatore Orlando <salvatore.orlando@eu.citrix.com>

3
nova/flags.py
View File

@@ -212,6 +212,8 @@ DEFINE_list('region_list',
DEFINE_string('connection_type', 'libvirt', 'libvirt, xenapi or fake')
DEFINE_string('aws_access_key_id', 'admin', 'AWS Access ID')
DEFINE_string('aws_secret_access_key', 'admin', 'AWS Access Key')
DEFINE_integer('glance_port', 9292, 'glance port')
DEFINE_string('glance_host', utils.get_my_ip(), 'glance host')
DEFINE_integer('s3_port', 3333, 's3 port')
DEFINE_string('s3_host', utils.get_my_ip(), 's3 host (for infrastructure)')
DEFINE_string('s3_dmz', utils.get_my_ip(), 's3 dmz ip (for instances)')
@@ -239,6 +241,7 @@ DEFINE_string('cc_dmz', utils.get_my_ip(), 'internal ip of api server')
DEFINE_integer('cc_port', 8773, 'cloud controller port')
DEFINE_string('ec2_suffix', '/services/Cloud', 'suffix for ec2')
DEFINE_string('default_project', 'openstack', 'default project for openstack')
DEFINE_string('default_image', 'ami-11111',
'default image to use, testing only')
DEFINE_string('default_instance_type', 'm1.small',

8
nova/tests/test_compute.py
View File

@@ -151,6 +151,14 @@ class ComputeTestCase(test.TestCase):
self.compute.reboot_instance(self.context, instance_id)
self.compute.terminate_instance(self.context, instance_id)
def test_snapshot(self):
"""Ensure instance can be snapshotted"""
instance_id = self._create_instance()
name = "myfakesnapshot"
self.compute.run_instance(self.context, instance_id)
self.compute.snapshot_instance(self.context, instance_id, name)
self.compute.terminate_instance(self.context, instance_id)
def test_console_output(self):
"""Make sure we can get console output from instance"""
instance_id = self._create_instance()

1
nova/tests/test_scheduler.py
View File

@@ -35,6 +35,7 @@ from nova.scheduler import driver
FLAGS = flags.FLAGS
flags.DECLARE('max_cores', 'nova.scheduler.simple')
flags.DECLARE('stub_network', 'nova.compute.manager')
class TestDriver(driver.Scheduler):

112
nova/tests/test_xenapi.py
View File

@@ -29,9 +29,9 @@ from nova.auth import manager
from nova.compute import instance_types
from nova.compute import power_state
from nova.virt import xenapi_conn
from nova.virt.xenapi import fake
from nova.virt.xenapi import fake as xenapi_fake
from nova.virt.xenapi import volume_utils
from nova.tests.db import fakes
from nova.tests.db import fakes as db_fakes
from nova.tests.xenapi import stubs
FLAGS = flags.FLAGS
@@ -47,9 +47,9 @@ class XenAPIVolumeTestCase(test.TestCase):
FLAGS.target_host = '127.0.0.1'
FLAGS.xenapi_connection_url = 'test_url'
FLAGS.xenapi_connection_password = 'test_pass'
fakes.stub_out_db_instance_api(self.stubs)
db_fakes.stub_out_db_instance_api(self.stubs)
stubs.stub_out_get_target(self.stubs)
fake.reset()
xenapi_fake.reset()
self.values = {'name': 1, 'id': 1,
'project_id': 'fake',
'user_id': 'fake',
@@ -83,7 +83,7 @@ class XenAPIVolumeTestCase(test.TestCase):
label = 'SR-%s' % vol['ec2_id']
description = 'Test-SR'
sr_ref = helper.create_iscsi_storage(session, info, label, description)
srs = fake.get_all('SR')
srs = xenapi_fake.get_all('SR')
self.assertEqual(sr_ref, srs[0])
db.volume_destroy(context.get_admin_context(), vol['id'])
@@ -107,17 +107,17 @@ class XenAPIVolumeTestCase(test.TestCase):
conn = xenapi_conn.get_connection(False)
volume = self._create_volume()
instance = db.instance_create(self.values)
fake.create_vm(instance.name, 'Running')
xenapi_fake.create_vm(instance.name, 'Running')
result = conn.attach_volume(instance.name, volume['ec2_id'],
'/dev/sdc')
def check():
# check that the VM has a VBD attached to it
# Get XenAPI reference for the VM
vms = fake.get_all('VM')
vms = xenapi_fake.get_all('VM')
# Get XenAPI record for VBD
vbds = fake.get_all('VBD')
vbd = fake.get_record('VBD', vbds[0])
vbds = xenapi_fake.get_all('VBD')
vbd = xenapi_fake.get_record('VBD', vbds[0])
vm_ref = vbd['VM']
self.assertEqual(vm_ref, vms[0])
@@ -130,7 +130,7 @@ class XenAPIVolumeTestCase(test.TestCase):
conn = xenapi_conn.get_connection(False)
volume = self._create_volume()
instance = db.instance_create(self.values)
fake.create_vm(instance.name, 'Running')
xenapi_fake.create_vm(instance.name, 'Running')
self.assertRaises(Exception,
conn.attach_volume,
instance.name,
@@ -156,41 +156,70 @@ class XenAPIVMTestCase(test.TestCase):
self.stubs = stubout.StubOutForTesting()
FLAGS.xenapi_connection_url = 'test_url'
FLAGS.xenapi_connection_password = 'test_pass'
fake.reset()
fakes.stub_out_db_instance_api(self.stubs)
fake.create_network('fake', FLAGS.flat_network_bridge)
xenapi_fake.reset()
db_fakes.stub_out_db_instance_api(self.stubs)
xenapi_fake.create_network('fake', FLAGS.flat_network_bridge)
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
self.conn = xenapi_conn.get_connection(False)
def test_list_instances_0(self):
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
conn = xenapi_conn.get_connection(False)
instances = conn.list_instances()
instances = self.conn.list_instances()
self.assertEquals(instances, [])
def test_spawn(self):
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
values = {'name': 1, 'id': 1,
'project_id': self.project.id,
'user_id': self.user.id,
'image_id': 1,
'kernel_id': 2,
'ramdisk_id': 3,
'instance_type': 'm1.large',
'mac_address': 'aa:bb:cc:dd:ee:ff',
}
conn = xenapi_conn.get_connection(False)
instance = db.instance_create(values)
conn.spawn(instance)
def test_get_diagnostics(self):
instance = self._create_instance()
self.conn.get_diagnostics(instance)
def test_instance_snapshot(self):
stubs.stubout_instance_snapshot(self.stubs)
instance = self._create_instance()
name = "MySnapshot"
template_vm_ref = self.conn.snapshot(instance, name)
def ensure_vm_was_torn_down():
vm_labels = []
for vm_ref in xenapi_fake.get_all('VM'):
vm_rec = xenapi_fake.get_record('VM', vm_ref)
if not vm_rec["is_control_domain"]:
vm_labels.append(vm_rec["name_label"])
self.assertEquals(vm_labels, [1])
def ensure_vbd_was_torn_down():
vbd_labels = []
for vbd_ref in xenapi_fake.get_all('VBD'):
vbd_rec = xenapi_fake.get_record('VBD', vbd_ref)
vbd_labels.append(vbd_rec["vm_name_label"])
self.assertEquals(vbd_labels, [1])
def ensure_vdi_was_torn_down():
for vdi_ref in xenapi_fake.get_all('VDI'):
vdi_rec = xenapi_fake.get_record('VDI', vdi_ref)
name_label = vdi_rec["name_label"]
self.assert_(not name_label.endswith('snapshot'))
def check():
instances = conn.list_instances()
ensure_vm_was_torn_down()
ensure_vbd_was_torn_down()
ensure_vdi_was_torn_down()
check()
def test_spawn(self):
instance = self._create_instance()
def check():
instances = self.conn.list_instances()
self.assertEquals(instances, [1])
# Get Nova record for VM
vm_info = conn.get_info(1)
vm_info = self.conn.get_info(1)
# Get XenAPI record for VM
vms = fake.get_all('VM')
vm = fake.get_record('VM', vms[0])
vms = xenapi_fake.get_all('VM')
vm = xenapi_fake.get_record('VM', vms[0])
# Check that m1.large above turned into the right thing.
instance_type = instance_types.INSTANCE_TYPES['m1.large']
@@ -218,3 +247,18 @@ class XenAPIVMTestCase(test.TestCase):
self.manager.delete_project(self.project)
self.manager.delete_user(self.user)
self.stubs.UnsetAll()
def _create_instance(self):
"""Creates and spawns a test instance"""
values = {'name': 1, 'id': 1,
'project_id': self.project.id,
'user_id': self.user.id,
'image_id': 1,
'kernel_id': 2,
'ramdisk_id': 3,
'instance_type': 'm1.large',
'mac_address': 'aa:bb:cc:dd:ee:ff'
}
instance = db.instance_create(values)
self.conn.spawn(instance)
return instance

66
nova/tests/xenapi/stubs.py
View File

@@ -19,6 +19,54 @@
from nova.virt import xenapi_conn
from nova.virt.xenapi import fake
from nova.virt.xenapi import volume_utils
from nova.virt.xenapi import vm_utils
def stubout_instance_snapshot(stubs):
@classmethod
def fake_fetch_image(cls, session, instance_id, image, user, project,
type):
# Stubout wait_for_task
def fake_wait_for_task(self, id, task):
class FakeEvent:
def send(self, value):
self.rv = value
def wait(self):
return self.rv
done = FakeEvent()
self._poll_task(id, task, done)
rv = done.wait()
return rv
stubs.Set(xenapi_conn.XenAPISession, 'wait_for_task',
fake_wait_for_task)
from nova.virt.xenapi.fake import create_vdi
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,
sr_ref=sr_ref, sharable=False)
vdi_rec = session.get_xenapi().VDI.get_record(vdi_ref)
vdi_uuid = vdi_rec['uuid']
return vdi_uuid
stubs.Set(vm_utils.VMHelper, 'fetch_image', fake_fetch_image)
def fake_parse_xmlrpc_value(val):
return val
stubs.Set(xenapi_conn, '_parse_xmlrpc_value', fake_parse_xmlrpc_value)
def fake_wait_for_vhd_coalesce(session, instance_id, sr_ref, vdi_ref,
original_parent_uuid):
#TODO(sirp): Should we actually fake out the data here
return "fakeparent"
stubs.Set(vm_utils, 'wait_for_vhd_coalesce', fake_wait_for_vhd_coalesce)
def stubout_session(stubs, cls):
@@ -63,6 +111,24 @@ class FakeSessionForVMTests(fake.SessionBase):
vm['is_a_template'] = False
vm['is_control_domain'] = False
def VM_snapshot(self, session_ref, vm_ref, label):
status = "Running"
template_vm_ref = fake.create_vm(label, status, is_a_template=True,
is_control_domain=False)
sr_ref = "fakesr"
template_vdi_ref = fake.create_vdi(label, read_only=True,
sr_ref=sr_ref, sharable=False)
template_vbd_ref = fake.create_vbd(template_vm_ref, template_vdi_ref)
return template_vm_ref
def VDI_destroy(self, session_ref, vdi_ref):
fake.destroy_vdi(vdi_ref)
def VM_destroy(self, session_ref, vm_ref):
fake.destroy_vm(vm_ref)
class FakeSessionForVolumeTests(fake.SessionBase):
""" Stubs out a XenAPISession for Volume tests """

68
run_tests.py Normal file
View File

@@ -0,0 +1,68 @@
#!/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.
import os
import unittest
import sys
from nose import config
from nose import result
from nose import core
class NovaTestResult(result.TextTestResult):
def __init__(self, *args, **kw):
result.TextTestResult.__init__(self, *args, **kw)
self._last_case = None
def getDescription(self, test):
return str(test)
def startTest(self, test):
unittest.TestResult.startTest(self, test)
current_case = test.test.__class__.__name__
if self.showAll:
if current_case != self._last_case:
self.stream.writeln(current_case)
self._last_case = current_case
self.stream.write(
' %s' % str(test.test._testMethodName).ljust(60))
self.stream.flush()
class NovaTestRunner(core.TextTestRunner):
def _makeResult(self):
return NovaTestResult(self.stream,
self.descriptions,
self.verbosity,
self.config)
if __name__ == '__main__':
c = config.Config(stream=sys.stdout,
env=os.environ,
verbosity=3)
runner = NovaTestRunner(stream=c.stream,
verbosity=c.verbosity,
config=c)
sys.exit(not core.run(config=c, testRunner=runner))