import contextlib
import os
-import StringIO
import mock
+import six
from cinder import context
from cinder.tests.unit.targets import targets_fixture as tf
self.assertTrue(m_exec.called)
def test_get_target_chap_auth(self):
- tmp_file = StringIO.StringIO()
+ tmp_file = six.StringIO()
tmp_file.write(
'target:\n'
' TargetName=iqn.2010-10.org.openstack:volume-83c2e877-feed-46be-8435-77884fe55b45\n' # noqa
import contextlib
-import StringIO
-
-
import mock
from oslo_concurrency import processutils as putils
+import six
from cinder import context
from cinder import exception
configuration=self.configuration)
def test_get_target(self):
- tmp_file = StringIO.StringIO()
+ tmp_file = six.StringIO()
tmp_file.write(
'tid:1 name:iqn.2010-10.org.openstack:volume-83c2e877-feed-46be-8435-77884fe55b45\n' # noqa
' sid:844427031282176 initiator:iqn.1994-05.com.redhat:5a6894679665\n' # noqa
@mock.patch('os.path.exists', return_value=True)
@mock.patch('cinder.utils.temporary_chown')
def test_get_target_chap_auth(self, mock_chown, mock_exists):
- tmp_file = StringIO.StringIO()
+ tmp_file = six.StringIO()
tmp_file.write(
'Target iqn.2010-10.org.openstack:volume-83c2e877-feed-46be-8435-77884fe55b45\n' # noqa
' IncomingUser otzLy2UYbYfnP4zXLG5z 234Zweo38VGBBvrpK9nt\n'
def test_create_iscsi_target(self, mock_chown, mock_exists,
mock_execute, mock_get_targ):
mock_execute.return_value = ('', '')
- tmp_file = StringIO.StringIO()
+ tmp_file = six.StringIO()
with mock.patch('__builtin__.open') as mock_open:
mock_open.return_value = contextlib.closing(tmp_file)
self.assertEqual(
self.testvol['name'])
def test_find_sid_cid_for_target(self):
- tmp_file = StringIO.StringIO()
+ tmp_file = six.StringIO()
tmp_file.write(
'tid:1 name:iqn.2010-10.org.openstack:volume-83c2e877-feed-46be-8435-77884fe55b45\n' # noqa
' sid:844427031282176 initiator:iqn.1994-05.com.redhat:5a6894679665\n' # noqa
"""
import os
-import StringIO
import tempfile
import mock
from oslo_log import log as logging
+import six
from cinder import exception
from cinder import test
def test_read_config(self, m_access, m_open):
# Test exception when file is not found
m_access.return_value = False
- m_open.return_value = StringIO.StringIO(HNASCONF)
+ m_open.return_value = six.StringIO(HNASCONF)
self.assertRaises(exception.NotFound, iscsi._read_config, '')
# Test exception when config file has parsing errors
# due to missing <svc> tag
m_access.return_value = True
- m_open.return_value = StringIO.StringIO(HNAS_WRONG_CONF1)
+ m_open.return_value = six.StringIO(HNAS_WRONG_CONF1)
self.assertRaises(exception.ConfigNotFound, iscsi._read_config, '')
# Test exception when config file has parsing errors
# due to missing <hdp> tag
- m_open.return_value = StringIO.StringIO(HNAS_WRONG_CONF2)
+ m_open.return_value = six.StringIO(HNAS_WRONG_CONF2)
self.configuration.hds_hnas_iscsi_config_file = ''
self.assertRaises(exception.ParameterNotFound, iscsi._read_config, '')
#
import os
-import StringIO
import tempfile
import mock
from oslo_log import log as logging
+import six
from cinder import exception
from cinder import test
def test_read_config(self, m_access, m_open):
# Test exception when file is not found
m_access.return_value = False
- m_open.return_value = StringIO.StringIO(HNASCONF)
+ m_open.return_value = six.StringIO(HNASCONF)
self.assertRaises(exception.NotFound, nfs._read_config, '')
# Test exception when config file has parsing errors
# due to missing <svc> tag
m_access.return_value = True
- m_open.return_value = StringIO.StringIO(HNAS_WRONG_CONF1)
+ m_open.return_value = six.StringIO(HNAS_WRONG_CONF1)
self.assertRaises(exception.ConfigNotFound, nfs._read_config, '')
# Test exception when config file has parsing errors
# due to missing <hdp> tag
- m_open.return_value = StringIO.StringIO(HNAS_WRONG_CONF2)
+ m_open.return_value = six.StringIO(HNAS_WRONG_CONF2)
self.configuration.hds_hnas_iscsi_config_file = ''
self.assertRaises(exception.ParameterNotFound, nfs._read_config, '')
import errno
import os
import six
-import StringIO
import traceback
import mock
mock.patch('cinder.volume.drivers.quobyte.QuobyteDriver'
'.read_proc_mount') as mock_open:
# Content of /proc/mount (not mounted yet).
- mock_open.return_value = StringIO.StringIO(
+ mock_open.return_value = six.StringIO(
"/dev/sda5 / ext4 rw,relatime,data=ordered 0 0")
self._driver._mount_quobyte(self.TEST_QUOBYTE_VOLUME,
mock.patch('cinder.volume.drivers.quobyte.QuobyteDriver'
'.read_proc_mount') as mock_open:
# Content of /proc/mount (already mounted).
- mock_open.return_value = StringIO.StringIO(
+ mock_open.return_value = six.StringIO(
"quobyte@%s %s fuse rw,nosuid,nodev,noatime,user_id=1000"
",group_id=100,default_permissions,allow_other 0 0"
% (self.TEST_QUOBYTE_VOLUME, self.TEST_MNT_POINT))
'.read_proc_mount') as mock_open, \
mock.patch('cinder.volume.drivers.quobyte.LOG') as mock_LOG:
# Content of /proc/mount (empty).
- mock_open.return_value = StringIO.StringIO()
+ mock_open.return_value = six.StringIO()
mock_execute.side_effect = [None, putils.ProcessExecutionError(
stderr='is busy or already mounted')]
with mock.patch.object(self._driver, '_execute') as mock_execute, \
mock.patch('cinder.volume.drivers.quobyte.QuobyteDriver'
'.read_proc_mount') as mock_open:
- mock_open.return_value = StringIO.StringIO()
+ mock_open.return_value = six.StringIO()
mock_execute.side_effect = [
None, # mkdir
putils.ProcessExecutionError( # mount
"""
import json
-import StringIO
import time
from oslo_log import log
+import six
from six.moves import http_client
from six.moves import urllib
:cmd_params args: The path part
:cmd_params kwargs: The query part
"""
- buf = StringIO.StringIO()
+ buf = six.StringIO()
query = "?"
for arg in args:
buf.write("/")