import fixtures
import mock
-import mox
from oslo_concurrency import lockutils
from oslo_config import cfg
from oslo_config import fixture as config_fixture
from oslo_messaging import conffixture as messaging_conffixture
from oslo_utils import strutils
from oslo_utils import timeutils
-import stubout
+from oslotest import moxstubout
import testtools
from cinder.common import config # noqa Need to register global_opts
# emulate some of the mox stuff, we can't use the metaclass
# because it screws with our generators
- self.mox = mox.Mox()
- self.stubs = stubout.StubOutForTesting()
+ mox_fixture = self.useFixture(moxstubout.MoxStubout())
+ self.mox = mox_fixture.mox
+ self.stubs = mox_fixture.stubs
self.addCleanup(CONF.reset)
- self.addCleanup(self.mox.UnsetStubs)
- self.addCleanup(self.stubs.UnsetAll)
- self.addCleanup(self.stubs.SmartUnsetAll)
- self.addCleanup(self.mox.VerifyAll)
self.addCleanup(self._common_cleanup)
self.injected = []
self._services = []
# 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 mox
+from mox3 import mox
from oslo_concurrency import processutils
from oslo_log import log as logging
class BrickLvmTestCase(test.TestCase):
def setUp(self):
- self._mox = mox.Mox()
self.configuration = mox.MockObject(conf.Configuration)
self.configuration.volume_group_name = 'fake-vg'
super(BrickLvmTestCase, self).setUp()
self.assertEqual(self.vg.create_lv_snapshot('snapshot-1', 'fake-1'),
None)
- self._mox.StubOutWithMock(self.vg, 'get_volume')
+ self.mox.StubOutWithMock(self.vg, 'get_volume')
self.vg.get_volume('fake-non-existent').AndReturn(None)
- self._mox.ReplayAll()
+ self.mox.ReplayAll()
try:
self.vg.create_lv_snapshot('snapshot-1', 'fake-non-existent')
except exception.VolumeDeviceNotFound as e:
self.assertFalse(self.vg.lv_has_snapshot('test-volumes'))
def test_activate_lv(self):
- self._mox.StubOutWithMock(self.vg, '_execute')
+ self.mox.StubOutWithMock(self.vg, '_execute')
self.vg._supports_lvchange_ignoreskipactivation = True
self.vg._execute('lvchange', '-a', 'y', '--yes', '-K',
'fake-vg/my-lv',
root_helper='sudo', run_as_root=True)
- self._mox.ReplayAll()
+ self.mox.ReplayAll()
self.vg.activate_lv('my-lv')
- self._mox.VerifyAll()
+ self.mox.VerifyAll()
def test_get_mirrored_available_capacity(self):
self.assertEqual(self.vg.vg_mirror_free_space(1), 2.0)
# under the License.
import mock
-import mox
from oslo_log import log as logging
from cinder.brick import exception
def setUp(self):
super(BrickRemoteFsTestCase, self).setUp()
- self._mox = mox.Mox()
self._nfsclient = remotefs.RemoteFsClient(
'nfs', 'sudo', nfs_mount_point_base=self.TEST_MNT_BASE)
- self.addCleanup(self._mox.UnsetStubs)
def test_get_hash_str(self):
"""_get_hash_str should calculation correct value."""
self.assertEqual(mnt_point, self.TEST_MNT_POINT)
def test_mount_nfs_should_mount_correctly(self):
- mox = self._mox
+ mox = self.mox
client = self._nfsclient
mox.StubOutWithMock(client, '_execute')
self.TEST_EXPORT)
def test_mount_nfs_should_not_remount(self):
- mox = self._mox
+ mox = self.mox
client = self._nfsclient
line = "%s on %s type nfs (rw)\n" % (self.TEST_EXPORT,
import os
import tempfile
-import mox
+from mox3 import mox
from cinder import test
from cinder.volume import configuration as conf
SimulatedHusBackend.alloc_lun = []
SimulatedHusBackend.connections = []
SimulatedHusBackend.out = ''
- self.mox = mox.Mox()
self.mox.StubOutWithMock(hds, 'factory_bend')
hds.factory_bend().AndReturn(SimulatedHusBackend())
self.mox.ReplayAll()
self.configuration = mox.MockObject(conf.Configuration)
self.configuration.hds_cinder_config_file = self.config_file
self.driver = hds.HUSDriver(configuration=self.configuration)
- self.addCleanup(self.mox.UnsetStubs)
def test_get_volume_stats(self):
stats = self.driver.get_volume_stats(True)
connector['host'] = 'dut_1.lab.hds.com'
vol = self.test_create_volume()
self.mox.StubOutWithMock(self.driver, '_update_vol_location')
+ self.driver._update_vol_location(vol['id'], mox.IgnoreArg())
+ self.mox.ReplayAll()
conn = self.driver.initialize_connection(vol, connector)
self.assertIn('hitachi', conn['data']['target_iqn'])
self.assertIn('3260', conn['data']['target_portal'])
because an error/exception return will only jeopardize the
connection tear down at a host.
"""
- (vol, conn) = self.test_initialize_connection()
+ connector = {}
+ connector['initiator'] = 'iqn.1993-08.org.debian:01:11f90746eb2'
+ connector['host'] = 'dut_1.lab.hds.com'
+ vol = self.test_create_volume()
+ self.mox.StubOutWithMock(self.driver, '_update_vol_location')
+ self.driver._update_vol_location(vol['id'], mox.IgnoreArg())
+ self.driver._update_vol_location(vol['id'], mox.IgnoreArg())
+
+ self.mox.ReplayAll()
+
+ conn = self.driver.initialize_connection(vol, connector)
+ vol['provider_location'] = conn['data']['provider_location']
num_conn_before = len(SimulatedHusBackend.connections)
- self.driver.terminate_connection(vol, conn)
+ self.driver.terminate_connection(vol, connector)
num_conn_after = len(SimulatedHusBackend.connections)
self.assertGreater(num_conn_before, num_conn_after)
import copy
-import mox
+from mox3 import mox
from oslo_config import cfg
from cinder import context
from lxml import etree
import mock
-import mox as mox_lib
+from mox3 import mox as mox_lib
from oslo_log import log as logging
import six
import httplib
from lxml import etree
-import mox
+from mox3 import mox
import six
from cinder import exception
import testtools
import mock
-import mox as mox_lib
-from mox import stubout
+from mox3 import mox as mox_lib
+from mox3.mox import stubout
from oslo_utils import units
from cinder import exception
def setUp(self):
super(RemoteFsDriverTestCase, self).setUp()
self._driver = remotefs.RemoteFSDriver()
- self._mox = mox_lib.Mox()
self.configuration = mox_lib.MockObject(conf.Configuration)
self.configuration.append_config_values(mox_lib.IgnoreArg())
self.configuration.nas_secure_file_permissions = 'false'
self.configuration.nas_secure_file_operations = 'false'
self._driver = remotefs.RemoteFSDriver(
configuration=self.configuration)
- self.addCleanup(self._mox.UnsetStubs)
def test_create_sparsed_file(self):
- (mox, drv) = self._mox, self._driver
+ (mox, drv) = self.mox, self._driver
mox.StubOutWithMock(drv, '_execute')
drv._execute('truncate', '-s', '1G', '/path', run_as_root=True).\
mox.VerifyAll()
def test_create_regular_file(self):
- (mox, drv) = self._mox, self._driver
+ (mox, drv) = self.mox, self._driver
mox.StubOutWithMock(drv, '_execute')
drv._execute('dd', 'if=/dev/zero', 'of=/path', 'bs=1M', 'count=1024',
mox.VerifyAll()
def test_create_qcow2_file(self):
- (mox, drv) = self._mox, self._driver
+ (mox, drv) = self.mox, self._driver
file_size = 1
mox.VerifyAll()
def test_set_rw_permissions_for_all(self):
- (mox, drv) = self._mox, self._driver
+ (mox, drv) = self.mox, self._driver
mox.StubOutWithMock(drv, '_execute')
drv._execute('chmod', 'ugo+rw', '/path', run_as_root=True)
def setUp(self):
super(NfsDriverTestCase, self).setUp()
- self._mox = mox_lib.Mox()
+ self.mox = mox_lib.Mox()
self.stubs = stubout.StubOutForTesting()
self.configuration = mox_lib.MockObject(conf.Configuration)
self.configuration.append_config_values(mox_lib.IgnoreArg())
self._driver = nfs.NfsDriver(configuration=self.configuration)
self._driver.shares = {}
self.addCleanup(self.stubs.UnsetAll)
- self.addCleanup(self._mox.UnsetStubs)
+ self.addCleanup(self.mox.UnsetStubs)
def stub_out_not_replaying(self, obj, attr_name):
attr_to_replace = getattr(obj, attr_name)
def test_copy_image_to_volume(self):
"""resize_image common case usage."""
- mox = self._mox
+ mox = self.mox
drv = self._driver
TEST_IMG_SOURCE = 'foo.img'
def test_get_capacity_info(self):
"""_get_capacity_info should calculate correct value."""
- mox = self._mox
+ mox = self.mox
drv = self._driver
stat_total_size = 2620544
def test_get_capacity_info_for_share_and_mount_point_with_spaces(self):
"""_get_capacity_info should calculate correct value."""
- mox = self._mox
+ mox = self.mox
drv = self._driver
stat_total_size = 2620544
mox.VerifyAll()
def test_load_shares_config(self):
- mox = self._mox
+ mox = self.mox
drv = self._driver
drv.configuration.nfs_shares_config = self.TEST_SHARES_CONFIG_FILE
mox.VerifyAll()
def test_load_shares_config_nas_opts(self):
- mox = self._mox
+ mox = self.mox
drv = self._driver
mox.StubOutWithMock(drv, '_read_config_file') # ensure not called
def test_ensure_shares_mounted_should_save_mounting_successfully(self):
"""_ensure_shares_mounted should save share if mounted with success."""
- mox = self._mox
+ mox = self.mox
drv = self._driver
mox.StubOutWithMock(drv, '_read_config_file')
def test_find_share(self):
"""_find_share simple use case."""
- mox = self._mox
+ mox = self.mox
drv = self._driver
drv._mounted_shares = [self.TEST_NFS_EXPORT1, self.TEST_NFS_EXPORT2]
def test_find_share_should_throw_error_if_there_is_no_enough_place(self):
"""_find_share should throw error if there is no share to host vol."""
- mox = self._mox
+ mox = self.mox
drv = self._driver
drv._mounted_shares = [self.TEST_NFS_EXPORT1, self.TEST_NFS_EXPORT2]
return volume
def test_create_sparsed_volume(self):
- mox = self._mox
+ mox = self.mox
drv = self._driver
volume = self._simple_volume()
mox.VerifyAll()
def test_create_nonsparsed_volume(self):
- mox = self._mox
+ mox = self.mox
drv = self._driver
self.configuration.nfs_sparsed_volumes = False
volume = self._simple_volume()
def test_create_volume_should_ensure_nfs_mounted(self):
"""create_volume ensures shares provided in config are mounted."""
- mox = self._mox
+ mox = self.mox
drv = self._driver
self.stub_out_not_replaying(nfs, 'LOG')
def test_create_volume_should_return_provider_location(self):
"""create_volume should return provider_location with found share."""
- mox = self._mox
+ mox = self.mox
drv = self._driver
self.stub_out_not_replaying(nfs, 'LOG')
def test_delete_volume(self):
"""delete_volume simple test case."""
- mox = self._mox
+ mox = self.mox
drv = self._driver
self.stub_out_not_replaying(drv, '_ensure_share_mounted')
def test_delete_should_ensure_share_mounted(self):
"""delete_volume should ensure that corresponding share is mounted."""
- mox = self._mox
+ mox = self.mox
drv = self._driver
self.stub_out_not_replaying(drv, '_execute')
def test_get_volume_stats(self):
"""get_volume_stats must fill the correct values."""
- mox = self._mox
+ mox = self.mox
drv = self._driver
drv._mounted_shares = [self.TEST_NFS_EXPORT1, self.TEST_NFS_EXPORT2]
import shutil
import tempfile
-import mox as mox_lib
+import mock
+from mox3 import mox as mox_lib
from oslo_utils import units
from cinder import context
self.TEST_VOLDIR,
self.TEST_CLONENAME)
- self._mox = mox_lib.Mox()
self.configuration = mox_lib.MockObject(conf.Configuration)
self._configure_driver()
super(ScalityDriverTestCase, self).setUp()
def test_create_snapshot(self):
"""Expected behaviour for create_snapshot."""
- mox = self._mox
+ mox = self.mox
vol_size = self._driver._size_bytes(self.TEST_VOLSIZE)
self._driver.create_snapshot(self.TEST_SNAPSHOT)
- mox.UnsetStubs()
- mox.VerifyAll()
-
def test_delete_snapshot(self):
"""Expected behaviour for delete_snapshot."""
- mox = self._mox
+ mox = self.mox
mox.StubOutWithMock(os, 'remove')
os.remove(self.TEST_SNAPPATH)
self._driver.delete_snapshot(self.TEST_SNAPSHOT)
- mox.UnsetStubs()
- mox.VerifyAll()
-
def test_initialize_connection(self):
"""Expected behaviour for initialize_connection."""
ret = self._driver.initialize_connection(self.TEST_VOLUME, None)
self._driver.extend_volume(self.TEST_VOLUME, self.TEST_NEWSIZE)
def test_backup_volume(self):
- self.mox.StubOutWithMock(self._driver, 'db')
+ self.mox = mox_lib.Mox()
+ self._driver.db = self.mox.CreateMockAnything()
self.mox.StubOutWithMock(self._driver.db, 'volume_get')
volume = {'id': '2', 'name': self.TEST_VOLNAME}
image_utils.qemu_img_info(self.TEST_VOLPATH).AndReturn(info)
self.mox.StubOutWithMock(utils, 'temporary_chown')
- mock_tempchown = self.mox.CreateMockAnything()
+ mock_tempchown = mock.MagicMock()
utils.temporary_chown(self.TEST_VOLPATH).AndReturn(mock_tempchown)
- mock_tempchown.__enter__()
- mock_tempchown.__exit__(None, None, None)
self.mox.StubOutWithMock(fileutils, 'file_open')
- mock_fileopen = self.mox.CreateMockAnything()
+ mock_fileopen = mock.MagicMock()
fileutils.file_open(self.TEST_VOLPATH).AndReturn(mock_fileopen)
- mock_fileopen.__enter__()
- mock_fileopen.__exit__(None, None, None)
backup = {'volume_id': volume['id']}
mock_servicebackup = self.mox.CreateMockAnything()
mock_servicebackup.backup(backup, mox_lib.IgnoreArg())
self.mox.ReplayAll()
+
self._driver.backup_volume(context, backup, mock_servicebackup)
- self.mox.VerifyAll()
def test_restore_backup(self):
volume = {'id': '2', 'name': self.TEST_VOLNAME}
self.mox.StubOutWithMock(utils, 'temporary_chown')
- mock_tempchown = self.mox.CreateMockAnything()
+ mock_tempchown = mock.MagicMock()
utils.temporary_chown(self.TEST_VOLPATH).AndReturn(mock_tempchown)
- mock_tempchown.__enter__()
- mock_tempchown.__exit__(None, None, None)
self.mox.StubOutWithMock(fileutils, 'file_open')
- mock_fileopen = self.mox.CreateMockAnything()
+ mock_fileopen = mock.MagicMock()
fileutils.file_open(self.TEST_VOLPATH, 'wb').AndReturn(mock_fileopen)
- mock_fileopen.__enter__()
- mock_fileopen.__exit__(None, None, None)
backup = {'id': 123, 'volume_id': volume['id']}
mock_servicebackup = self.mox.CreateMockAnything()
mock_servicebackup.restore(backup, volume['id'], mox_lib.IgnoreArg())
self.mox.ReplayAll()
+
self._driver.restore_backup(context, backup, volume,
mock_servicebackup)
- self.mox.VerifyAll()
import mock
-import mox
+from mox3 import mox
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_db import exception as db_exc
import datetime
import mock
-import mox
+from mox3 import mox
from oslo_log import log as logging
from oslo_utils import timeutils
from oslo_utils import units
class SolidFireVolumeTestCase(test.TestCase):
def setUp(self):
self.ctxt = context.get_admin_context()
- self._mox = mox.Mox()
self.configuration = mox.MockObject(conf.Configuration)
self.configuration.sf_allow_tenant_qos = True
self.configuration.san_is_local = True
from distutils import version as ver
import mock
-import mox
+from mox3 import mox
from oslo_utils import units
from oslo_vmware import api
from oslo_vmware import exceptions
import eventlet
import mock
-import mox
+from mox3 import mox
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_serialization import jsonutils
self.assertEqual(volume_id, volume_ref.id)
self.assertEqual("available", volume_ref.status)
- self.mox.UnsetStubs()
- self.volume.delete_volume(self.context, volume_id)
-
def test_get_volume_different_tenant(self):
"""Test can't get volume of another tenant when viewable_admin_meta."""
volume = tests_utils.create_volume(self.context,
self.assertEqual(snapshot_id, snapshot_ref.id)
self.assertEqual("available", snapshot_ref.status)
- self.mox.UnsetStubs()
- self.volume.delete_snapshot(self.context, snapshot_obj)
- self.volume.delete_volume(self.context, volume_id)
-
@test.testtools.skipIf(sys.platform == "darwin", "SKIP on OSX")
def test_delete_no_dev_fails(self):
"""Test delete snapshot with no dev file fails."""
import shutil
import tempfile
-import mox
+from mox3 import mox
from oslo_config import cfg
from cinder.image import image_utils
discover
fixtures>=0.3.14
mock>=1.0
-mox>=0.5.3
+mox3>=0.7.0
MySQL-python
psycopg2
oslotest>=1.5.1 # Apache-2.0
Code Review / openstack-build / cinder-build.git / commitdiff