super(TestBaseISCSITargetDriver, self).setUp()
self.target = fake.FakeTarget(root_helper=utils.get_root_helper(),
configuration=self.configuration)
+ self.target.db = mock.MagicMock(
+ volume_get=lambda x, y: {'provider_auth': 'CHAP otzL 234Z'})
def test_abc_methods_not_present_fails(self):
configuration = conf.Configuration(cfg.StrOpt('iscsi_target_prefix',
location = self.target._iscsi_location('portal', 1, 'target', 2,
['portal2'])
self.assertEqual('portal:3260;portal2:3260,1 target 2', location)
+
+ def test_get_target_chap_auth(self):
+ ctxt = context.get_admin_context()
+ self.assertEqual(('otzL', '234Z'),
+ self.target._get_target_chap_auth(ctxt,
+ self.test_vol))
# License for the specific language governing permissions and limitations
# under the License.
-import contextlib
import os
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 = six.StringIO()
- tmp_file.write(
- 'target:\n'
- ' TargetName=iqn.2010-10.org.openstack:volume-%(id)s\n'
- ' TargetDevice=/dev/%(vg)s/volume-%(id)s\n'
- ' PortalGroup=1@10.9.8.7:3260\n'
- ' AuthMethod=CHAP\n'
- ' Auth_CHAP_Policy=Oneway\n'
- ' Auth_CHAP_Initiator="otzL":"234Z"\n' %
- {'id': self.VOLUME_ID, 'vg': self.VG}
- )
- tmp_file.seek(0)
-
- expected = ('otzL', '234Z')
- with mock.patch('six.moves.builtins.open') as mock_open:
- ctx = context.get_admin_context()
- mock_open.return_value = contextlib.closing(tmp_file)
- self.assertEqual(expected,
- self.target._get_target_chap_auth(ctx,
- self.test_vol))
- self.assertTrue(mock_open.called)
-
- def test_get_target_chap_auth_negative(self):
- with mock.patch('six.moves.builtins.open') as mock_open:
- e = IOError()
- e.errno = 123
- mock_open.side_effect = e
- ctxt = context.get_admin_context()
- self.assertRaises(IOError,
- self.target._get_target_chap_auth,
- ctxt, self.test_vol)
- mock_open.side_effect = ZeroDivisionError()
- self.assertRaises(ZeroDivisionError,
- self.target._get_target_chap_auth,
- ctxt, self.test_vol)
-
@mock.patch('cinder.volume.targets.cxt.CxtAdm._get_target',
return_value=1)
@mock.patch('cinder.utils.execute')
self.target._get_target,
'')
- @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 = six.StringIO()
- tmp_file.write(
- 'Target iqn.2010-10.org.openstack:volume-83c2e877-feed-46be-8435-77884fe55b45\n' # noqa
- ' IncomingUser otzLy2UYbYfnP4zXLG5z 234Zweo38VGBBvrpK9nt\n'
- ' Lun 0 Path=/dev/stack-volumes-lvmdriver-1/volume-83c2e877-feed-46be-8435-77884fe55b45,Type=fileio\n' # noqa
- )
- tmp_file.seek(0)
- expected = ('otzLy2UYbYfnP4zXLG5z', '234Zweo38VGBBvrpK9nt')
- with mock.patch('__builtin__.open') as mock_open:
- ictx = context.get_admin_context()
- mock_open.return_value = contextlib.closing(tmp_file)
- self.assertEqual(expected,
- self.target._get_target_chap_auth(ictx,
- self.test_vol))
- self.assertTrue(mock_open.called)
-
- # Test the failure case: Failed to handle the config file
- mock_open.side_effect = StandardError()
- self.assertRaises(StandardError,
- self.target._get_target_chap_auth,
- ictx,
- self.test_vol)
-
@mock.patch('cinder.volume.targets.iet.IetAdm._get_target',
return_value=0)
@mock.patch('cinder.utils.execute')
with mock.patch.object(lio.LioAdm, '_verify_rtstool'):
self.target = lio.LioAdm(root_helper=utils.get_root_helper(),
configuration=self.configuration)
- self.target.db = mock.MagicMock(
- volume_get=lambda x, y: {'provider_auth': 'IncomingUser foo bar'})
@mock.patch.object(lio.LioAdm, '_execute', side_effect=lio.LioAdm._execute)
@mock.patch.object(lio.LioAdm, '_persist_configuration')
self.assertEqual(expected,
self.target._get_target_and_lun(ctxt, self.testvol))
- def test_get_target_chap_auth(self):
- ctxt = context.get_admin_context()
- self.assertEqual(('foo', 'bar'),
- self.target._get_target_chap_auth(ctxt,
- self.test_vol))
-
@mock.patch.object(lio.LioAdm, '_execute', side_effect=lio.LioAdm._execute)
@mock.patch.object(lio.LioAdm, '_persist_configuration')
@mock.patch('cinder.utils.execute')
self.assertEqual(expected,
self.target._get_target_and_lun(ctxt, self.testvol))
- def test_get_target_chap_auth(self):
- persist_file =\
- '<target iqn.2010-10.org.openstack:volume-%(id)s>\n'\
- ' backing-store %(bspath)s\n'\
- ' driver iscsi\n'\
- ' incominguser otzL 234Z\n'\
- ' write-cache on\n'\
- '</target>' % {'id': self.VOLUME_ID,
- 'bspath': self.testvol_path}
- with open(os.path.join(self.fake_volumes_dir,
- self.test_vol.split(':')[1]),
- 'w') as tmp_file:
- tmp_file.write(persist_file)
- ctxt = context.get_admin_context()
- expected = ('otzL', '234Z')
- self.assertEqual(expected,
- self.target._get_target_chap_auth(ctxt,
- self.test_vol))
-
- def test_get_target_chap_auth_negative(self):
- with mock.patch('six.moves.builtins.open') as mock_open:
- e = IOError()
- e.errno = 123
- mock_open.side_effect = e
- ctxt = context.get_admin_context()
- self.assertRaises(IOError,
- self.target._get_target_chap_auth,
- ctxt, self.test_vol)
- mock_open.side_effect = ZeroDivisionError()
- self.assertRaises(ZeroDivisionError,
- self.target._get_target_chap_auth,
- ctxt, self.test_vol)
-
def test_create_iscsi_target(self):
with mock.patch('cinder.utils.execute', return_value=('', '')),\
mock.patch.object(self.target, '_get_target',
image_id='fake_id',
source_volume='fake_id')
+ @mock.patch.object(cinder.volume.targets.iscsi.ISCSITarget,
+ '_get_target_chap_auth')
@mock.patch.object(db, 'volume_admin_metadata_get')
@mock.patch.object(db, 'volume_get')
@mock.patch.object(db, 'volume_update')
def test_initialize_connection_fetchqos(self,
_mock_volume_update,
_mock_volume_get,
- _mock_volume_admin_metadata_get):
+ _mock_volume_admin_metadata_get,
+ mock_get_target):
"""Make sure initialize_connection returns correct information."""
_fake_admin_meta = {'fake-key': 'fake-value'}
_fake_volume = {'volume_type_id': 'fake_type_id',
'initialize_connection') as driver_init:
type_qos.return_value = dict(qos_specs=qos_values)
driver_init.return_value = {'data': {}}
+ mock_get_target.return_value = None
qos_specs_expected = {'key1': 'value1',
'key2': 'value2'}
# initialize_connection() passes qos_specs that is designated to
'fake_volume_id',
connector)
+ @mock.patch.object(cinder.volume.targets.iscsi.ISCSITarget,
+ '_get_target_chap_auth')
@mock.patch.object(db, 'volume_admin_metadata_get')
@mock.patch.object(db, 'volume_update')
@mock.patch.object(db, 'volume_get')
mock_driver_init,
mock_volume_get,
mock_volume_update,
- mock_metadata_get):
+ mock_metadata_get,
+ mock_get_target):
fake_admin_meta = {'fake-key': 'fake-value'}
fake_volume = {'volume_type_id': None,
mock_volume_get.return_value = fake_volume
mock_volume_update.return_value = fake_volume
+ mock_get_target.return_value = None
connector = {'ip': 'IP', 'initiator': 'INITIATOR'}
mock_driver_init.return_value = {
'driver_volume_type': 'iscsi',
import os
-import re
from oslo_concurrency import processutils as putils
from oslo_log import log as logging
iscsi_target = 1
return iscsi_target, lun
- def _get_target_chap_auth(self, context, name):
- volumes_dir = self._get_volumes_dir()
- vol_id = name.split(':')[1]
- volume_path = os.path.join(volumes_dir, vol_id)
-
- try:
- with open(volume_path, 'r') as f:
- volume_conf = f.read()
- except IOError as e_fnf:
- LOG.debug('Failed to open config for %(vol_id)s: %(e)s',
- {'vol_id': vol_id, 'e': e_fnf})
- # We don't run on anything non-linux
- if e_fnf.errno == 2:
- return None
- else:
- raise
- except Exception as e_vol:
- LOG.error(_LE('Failed to open config for %(vol_id)s: %(e)s'),
- {'vol_id': vol_id, 'e': e_vol})
- raise
-
- m = re.search('Auth_CHAP_Initiator="(\w+)":"(\w+)"', volume_conf)
- if m:
- return (m.group(1), m.group(2))
- LOG.debug('Failed to find CHAP auth from config for %s', vol_id)
- return None
-
@staticmethod
def _get_portal(ip, port=None):
# ipv6 addresses use [ip]:port format, ipv4 use ip:port
def _get_target_and_lun(self, context, volume):
return(0, 0)
- def _get_target_chap_auth(self, context, iscsi_name):
- pass
-
def create_iscsi_target(self, name, tid, lun, path,
chap_auth, **kwargs):
pass
return iscsi_target, lun
- def _get_target_chap_auth(self, context, name):
-
- vol_id = name.split(':')[1]
- if os.path.exists(self.iet_conf):
- try:
- with utils.temporary_chown(self.iet_conf):
- with open(self.iet_conf, 'r') as f:
- iet_conf_text = f.readlines()
- except Exception:
- # If we fail to handle config file, raise exception here to
- # prevent unexpected behavior during subsequent operations.
- LOG.exception(_LE("Failed to open config for %s."), vol_id)
- raise
-
- target_found = False
- for line in iet_conf_text:
- if target_found:
- m = re.search('(\w+) (\w+) (\w+)', line)
- if m:
- return (m.group(2), m.group(3))
- else:
- LOG.debug("Failed to find CHAP auth from config "
- "for %s", vol_id)
- return None
- elif name in line:
- target_found = True
- else:
- # Missing config file is unxepected sisuation. But we will create
- # new config file during create_iscsi_target(). Just we warn the
- # operator here.
- LOG.warning(_LW("Failed to find CHAP auth from config for "
- "%(vol_id)s. Config file %(conf)s does not "
- "exist."),
- {'vol_id': vol_id, 'conf': self.iet_conf})
- return None
-
def create_iscsi_target(self, name, tid, lun, path,
chap_auth=None, **kwargs):
if tid is None:
raise exception.NotFound()
+ def _get_target_chap_auth(self, context, iscsi_name):
+ """Get the current chap auth username and password."""
+ try:
+ # 'iscsi_name': 'iqn.2010-10.org.openstack:volume-00000001'
+ vol_id = iscsi_name.split(':volume-')[1]
+ volume_info = self.db.volume_get(context, vol_id)
+ # 'provider_auth': 'CHAP user_id password'
+ if volume_info['provider_auth']:
+ return tuple(volume_info['provider_auth'].split(' ', 3)[1:])
+ except exception.NotFound:
+ LOG.debug('Failed to get CHAP auth from DB for %s.', vol_id)
+
@abc.abstractmethod
def _get_target_and_lun(self, context, volume):
"""Get iscsi target and lun."""
pass
- @abc.abstractmethod
- def _get_target_chap_auth(self, context, iscsi_name):
- pass
-
@abc.abstractmethod
def create_iscsi_target(self, name, tid, lun, path,
chap_auth, **kwargs):
self._verify_rtstool()
- def _get_target_chap_auth(self, context, iscsi_name):
- """Get the current chap auth username and password."""
- try:
- # 'iscsi_name': 'iqn.2010-10.org.openstack:volume-00000001'
- vol_id = iscsi_name.split(':volume-')[1]
- volume_info = self.db.volume_get(context, vol_id)
- # 'provider_auth': 'CHAP user_id password'
- if volume_info['provider_auth']:
- return tuple(volume_info['provider_auth'].split(' ', 3)[1:])
- except exception.NotFound:
- LOG.debug('Failed to get CHAP auth from DB for %s', vol_id)
-
def _verify_rtstool(self):
try:
# This call doesn't need locking
# under the License.
import os
-import re
import textwrap
import time
iscsi_target = 0 # NOTE(jdg): Not used by tgtadm
return iscsi_target, lun
- def _get_target_chap_auth(self, context, iscsi_name):
- """Get the current chap auth username and password."""
- volumes_dir = self.volumes_dir
- vol_id = iscsi_name.split(':')[1]
- volume_path = os.path.join(volumes_dir, vol_id)
-
- try:
- with open(volume_path, 'r') as f:
- volume_conf = f.read()
- except IOError as e_fnf:
- LOG.debug('Failed to open config for Volume %(vol_id)s: %(e)s',
- {'vol_id': vol_id, 'e': e_fnf})
- # tgt is linux specific
- if e_fnf.errno == 2:
- return None
- else:
- raise
- except Exception as e_vol:
- LOG.error(_LE('Failed to open config for %(vol_id)s: %(e)s'),
- {'vol_id': vol_id, 'e': e_vol})
- raise
-
- m = re.search('incominguser (\w+) (\w+)', volume_conf)
- if m:
- return (m.group(1), m.group(2))
- LOG.debug('Failed to find CHAP auth from config for %s', vol_id)
- return None
-
@utils.retry(putils.ProcessExecutionError)
def _do_tgt_update(self, name):
(out, err) = utils.execute('tgt-admin', '--update', name,
Code Review / openstack-build / cinder-build.git / commitdiff