x/python-ganttclient
Code Issues Proposed changes

Removed FakeInstance and introduced stubout for DB. Code clean-up

Browse Source
This commit is contained in:
Armando Migliaccio
2010-12-16 16:33:38 +00:00
parent cad11d8ad0
commit 331d83b5a2
3 changed files with 173 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
nova/tests/xenapi/__init__.py Normal file
View File

@@ -0,0 +1,33 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright (c) 2010 Citrix Systems, Inc.
#
# 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.
# vim: tabstop=4 shiftwidth=4 softtabstop=4
#
# Copyright (c) 2010 Citrix Systems, Inc.
#
# 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.
"""Stubouts, mocks and fixtures for the test suite"""

98
nova/tests/xenapi/stubs.py Normal file
View File

@@ -0,0 +1,98 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright (c) 2010 Citrix Systems, Inc.
#
# 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.
"""Stubouts, mocks and fixtures for the test suite"""
from nova.virt import xenapi_conn
from nova.virt.xenapi import fake
def stubout_session(stubs, cls):
""" Stubs out two methods from XenAPISession """
def fake_import(self):
""" Stubs out get_imported_xenapi of XenAPISession """
fake_module = 'nova.virt.xenapi.fake'
from_list = ['fake']
return __import__(fake_module, globals(), locals(), from_list, -1)
stubs.Set(xenapi_conn.XenAPISession, '_create_session',
lambda s, url: cls(url))
stubs.Set(xenapi_conn.XenAPISession, 'get_imported_xenapi',
fake_import)
class FakeSessionForVMTests(fake.SessionBase):
""" Stubs out a XenAPISession for VM tests """
def __init__(self, uri):
super(FakeSessionForVMTests, self).__init__(uri)
def network_get_all_records_where(self, _1, _2):
return self.xenapi.network.get_all_records()
def host_call_plugin(self, _1, _2, _3, _4, _5):
return ''
def VM_start(self, _1, ref, _2, _3):
vm = fake.get_record('VM', ref)
if vm['power_state'] != 'Halted':
raise fake.Failure(['VM_BAD_POWER_STATE', ref, 'Halted',
vm['power_state']])
vm['power_state'] = 'Running'
class FakeSessionForVolumeTests(fake.SessionBase):
""" Stubs out a XenAPISession for Volume tests """
def __init__(self, uri):
super(FakeSessionForVolumeTests, self).__init__(uri)
def VBD_plug(self, _1, _2):
#FIXME(armando):make proper plug
pass
def PBD_unplug(self, _1, _2):
#FIXME(armando):make proper unplug
pass
def SR_forget(self, _1, _2):
#FIXME(armando):make proper forget
pass
def VDI_introduce(self, _1, uuid, _2, _3, _4, _5,
_6, _7, _8, _9, _10, _11):
#FIXME(armando):make proper introduce
valid_vdi = False
refs = fake.get_all('VDI')
for ref in refs:
rec = fake.get_record('VDI', ref)
if rec['uuid'] == uuid:
valid_vdi = True
if not valid_vdi:
raise fake.Failure([['INVALID_VDI', 'session', self._session]])
class FakeSessionForVolumeFailedTests(FakeSessionForVolumeTests):
""" Stubs out a XenAPISession for Volume tests: it injects failures """
def __init__(self, uri):
super(FakeSessionForVolumeFailedTests, self).__init__(uri)
def VDI_introduce(self, _1, uuid, _2, _3, _4, _5,
_6, _7, _8, _9, _10, _11):
# This is for testing failure
raise fake.Failure([['INVALID_VDI', 'session', self._session]])
def VBD_plug(self, _1, _2):
# This is for testing failure
raise fake.Failure([['INVALID_VBD', 'session', self._session]])

165
nova/tests/xenapi_unittest.py
View File

@@ -14,32 +14,14 @@
# License for the specific language governing permissions and limitations
# under the License.
# vim: tabstop=4 shiftwidth=4 softtabstop=4
#
# Copyright (c) 2010 Citrix Systems, Inc.
#
# 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.
"""
Test suite for XenAPI
"""
import stubout
import uuid
from twisted.internet import defer
from twisted.internet import threads
from nova import db
from nova import context
from nova import exception
from nova import flags
from nova import test
from nova import utils
@@ -49,28 +31,15 @@ from nova.compute import power_state
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
from nova.virt.xenapi import volumeops
from boto.ec2.volume import Volume
from nova.tests.db import fakes
from nova.tests.xenapi import stubs
FLAGS = flags.FLAGS
def stubout_session(stubs, cls):
def fake_import(self):
fake_module = 'nova.virt.xenapi.fake'
from_list = ['fake']
return __import__(fake_module, globals(), locals(), from_list, -1)
stubs.Set(xenapi_conn.XenAPISession, '_create_session',
lambda s, url: cls(url))
stubs.Set(xenapi_conn.XenAPISession, 'get_imported_xenapi',
fake_import)
class XenAPIVolumeTestCase(test.TrialTestCase):
"""
Unit tests for VM operations
Unit tests for Volume operations
"""
def setUp(self):
super(XenAPIVolumeTestCase, self).setUp()
@@ -78,7 +47,17 @@ class XenAPIVolumeTestCase(test.TrialTestCase):
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)
fake.reset()
self.values = {'name': 1,
'project_id': 'fake',
'user_id': 'fake',
'image_id': 1,
'kernel_id': 2,
'ramdisk_id': 3,
'instance_type': 'm1.large',
'mac_address': 'aa:bb:cc:dd:ee:ff',
}
def _create_volume(self, size='0'):
"""Create a volume object."""
@@ -94,7 +73,7 @@ class XenAPIVolumeTestCase(test.TrialTestCase):
def test_create_iscsi_storage(self):
""" This shows how to test helper classes' methods """
stubout_session(self.stubs, FakeSessionForVolumeTests)
stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests)
session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass')
helper = volume_utils.VolumeHelper
helper.XenAPI = session.get_imported_xenapi()
@@ -106,11 +85,13 @@ class XenAPIVolumeTestCase(test.TrialTestCase):
info,
label,
description)
srs = fake.get_all('SR')
self.assertEqual(sr_ref, srs[0])
db.volume_destroy(context.get_admin_context(), vol['id'])
def test_parse_volume_info_raise_exception(self):
""" This shows how to test helper classes' methods """
stubout_session(self.stubs, FakeSessionForVolumeTests)
stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests)
session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass')
helper = volume_utils.VolumeHelper
helper.XenAPI = session.get_imported_xenapi()
@@ -119,6 +100,7 @@ class XenAPIVolumeTestCase(test.TrialTestCase):
info = helper.parse_volume_info(vol['ec2_id'], '/dev/sd')
def check(exc):
""" handler """
self.assertIsInstance(exc.value, volume_utils.StorageError)
info.addErrback(check)
@@ -126,16 +108,16 @@ class XenAPIVolumeTestCase(test.TrialTestCase):
def test_attach_volume(self):
""" This shows how to test Ops classes' methods """
stubout_session(self.stubs, FakeSessionForVolumeTests)
stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests)
conn = xenapi_conn.get_connection(False)
volume = self._create_volume()
instance = FakeInstance(1, 'fake', 'fake', 1, 2, 3,
'm1.large', 'aa:bb:cc:dd:ee:ff')
instance = db.instance_create(self.values)
fake.create_vm(instance.name, 'Running')
result = conn.attach_volume(instance.name, volume['ec2_id'],
'/dev/sdc')
def check(_):
""" handler """
# check that the VM has a VBD attached to it
# Get XenAPI reference for the VM
vms = fake.get_all('VM')
@@ -150,16 +132,17 @@ class XenAPIVolumeTestCase(test.TrialTestCase):
def test_attach_volume_raise_exception(self):
""" This shows how to test when exceptions are raised """
stubout_session(self.stubs, FakeSessionForVolumeFailedTests)
stubs.stubout_session(self.stubs,
stubs.FakeSessionForVolumeFailedTests)
conn = xenapi_conn.get_connection(False)
volume = self._create_volume()
instance = FakeInstance(1, 'fake', 'fake', 1, 2, 3,
'm1.large', 'aa:bb:cc:dd:ee:ff')
instance = db.instance_create(self.values)
fake.create_vm(instance.name, 'Running')
result = conn.attach_volume(instance.name, volume['ec2_id'],
'/dev/sdc')
def check(exc):
""" handler """
if exc:
pass
else:
@@ -188,22 +171,32 @@ class XenAPIVMTestCase(test.TrialTestCase):
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)
def test_list_instances_0(self):
stubout_session(self.stubs, FakeSessionForVMTests)
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
conn = xenapi_conn.get_connection(False)
instances = conn.list_instances()
self.assertEquals(instances, [])
def test_spawn(self):
stubout_session(self.stubs, FakeSessionForVMTests)
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
values = {'name': 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 = FakeInstance(1, self.project.id, self.user.id, 1, 2, 3,
'm1.large', 'aa:bb:cc:dd:ee:ff')
instance = db.instance_create(values)
result = conn.spawn(instance)
def check(_):
""" handler """
instances = conn.list_instances()
self.assertEquals(instances, [1])
@@ -241,77 +234,3 @@ class XenAPIVMTestCase(test.TrialTestCase):
self.manager.delete_project(self.project)
self.manager.delete_user(self.user)
self.stubs.UnsetAll()
class FakeInstance():
def __init__(self, name, project_id, user_id, image_id, kernel_id,
ramdisk_id, instance_type, mac_address):
self.name = name
self.project_id = project_id
self.user_id = user_id
self.image_id = image_id
self.kernel_id = kernel_id
self.ramdisk_id = ramdisk_id
self.instance_type = instance_type
self.mac_address = mac_address
class FakeSessionForVMTests(fake.SessionBase):
def __init__(self, uri):
super(FakeSessionForVMTests, self).__init__(uri)
def network_get_all_records_where(self, _1, _2):
return self.xenapi.network.get_all_records()
def host_call_plugin(self, _1, _2, _3, _4, _5):
return ''
def VM_start(self, _1, ref, _2, _3):
vm = fake.get_record('VM', ref)
if vm['power_state'] != 'Halted':
raise fake.Failure(['VM_BAD_POWER_STATE', ref, 'Halted',
vm['power_state']])
vm['power_state'] = 'Running'
class FakeSessionForVolumeTests(fake.SessionBase):
def __init__(self, uri):
super(FakeSessionForVolumeTests, self).__init__(uri)
def VBD_plug(self, _1, _2):
#FIXME(armando):make proper plug
pass
def PBD_unplug(self, _1, _2):
#FIXME(armando):make proper unplug
pass
def SR_forget(self, _1, _2):
#FIXME(armando):make proper forget
pass
def VDI_introduce(self, _1, uuid, _2, _3, _4, _5,
_6, _7, _8, _9, _10, _11):
#FIXME(armando):make proper introduce
valid_vdi = False
refs = fake.get_all('VDI')
for ref in refs:
rec = fake.get_record('VDI', ref)
if rec['uuid'] == uuid:
valid_vdi = True
if not valid_vdi:
raise fake.Failure([['INVALID_VDI', 'session', self._session]])
class FakeSessionForVolumeFailedTests(FakeSessionForVolumeTests):
def __init__(self, uri):
super(FakeSessionForVolumeFailedTests, self).__init__(uri)
def VDI_introduce(self, _1, uuid, _2, _3, _4, _5,
_6, _7, _8, _9, _10, _11):
# test failure
raise fake.Failure([['INVALID_VDI', 'session', self._session]])
def VBD_plug(self, _1, _2):
# test failure
raise fake.Failure([['INVALID_VBD', 'session', self._session]])