import os
import datetime
-import cStringIO
-import logging
import mox
import shutil
from cinder import db
from cinder import flags
from cinder.tests.image import fake as fake_image
-from cinder.openstack.common import log as os_logging
from cinder.openstack.common import importutils
from cinder.openstack.common.notifier import test_notifier
from cinder.openstack.common import rpc
import cinder.volume.api
FLAGS = flags.FLAGS
-LOG = os_logging.getLogger(__name__)
class VolumeTestCase(test.TestCase):
volume_id)
self.assert_(iscsi_target not in targets)
targets.append(iscsi_target)
- LOG.debug(_("Target %s allocated"), iscsi_target)
total_slots = FLAGS.iscsi_num_targets
for _index in xrange(total_slots):
volume = self._create_volume()
super(DriverTestCase, self).setUp()
vol_tmpdir = tempfile.mkdtemp()
self.flags(volume_driver=self.driver_name,
- volumes_dir=vol_tmpdir,
- logging_default_format_string="%(message)s")
+ volumes_dir=vol_tmpdir)
self.volume = importutils.import_object(FLAGS.volume_manager)
self.context = context.get_admin_context()
self.output = ""
return self.output, None
self.volume.driver.set_execute(_fake_execute)
- log = logging.getLogger()
- self.stream = cStringIO.StringIO()
- log.addHandler(logging.StreamHandler(self.stream))
-
def tearDown(self):
try:
shutil.rmtree(FLAGS.volumes_dir)
super(DriverTestCase, self).tearDown()
def _attach_volume(self):
- """Attach volumes to an instance. This function also sets
- a fake log message."""
+ """Attach volumes to an instance. """
return []
def _detach_volume(self, volume_id_list):
driver_name = "cinder.volume.driver.ISCSIDriver"
def _attach_volume(self):
- """Attach volumes to an instance. This function also sets
- a fake log message."""
+ """Attach volumes to an instance. """
volume_id_list = []
for index in xrange(3):
vol = {}
return volume_id_list
def test_check_for_export_with_no_volume(self):
- """No log message when no volume is attached to an instance."""
- self.stream.truncate(0)
instance_uuid = '12345678-1234-5678-1234-567812345678'
self.volume.check_for_export(self.context, instance_uuid)
- self.assertEqual(self.stream.getvalue(), '')
def test_check_for_export_with_all_volume_exported(self):
- """No log message when all the processes are running."""
volume_id_list = self._attach_volume()
self.mox.StubOutWithMock(self.volume.driver.tgtadm, 'show_target')
tid = db.volume_get_iscsi_target_num(self.context, i)
self.volume.driver.tgtadm.show_target(tid)
- self.stream.truncate(0)
self.mox.ReplayAll()
instance_uuid = '12345678-1234-5678-1234-567812345678'
self.volume.check_for_export(self.context, instance_uuid)
- self.assertEqual(self.stream.getvalue(), '')
self.mox.UnsetStubs()
self._detach_volume(volume_id_list)
self.volume.check_for_export,
self.context,
instance_uuid)
- msg = _("Cannot confirm exported volume id:%s.") % volume_id_list[0]
- self.assertTrue(0 <= self.stream.getvalue().find(msg))
self.mox.UnsetStubs()
self._detach_volume(volume_id_list)