Windows SMBFS: fix using share subdirs

When using Scale-Out File Server SMB shares, we have to use the local
share paths instead of the UNC paths if the host is part of the
cluster.

There's an os-win method that allows fetching the shared path but
it expects the share name, while we're passing subdirs as well,
for which reason it fails.

We'll have to make sure we're passing the share name only, appending
the subdir afterwards.

Closes-Bug: #1805341

Change-Id: I47d7e7afefe050cd4b25435cfcfc3e8963bd04bc
changes/53/620253/2
Lucian Petrut 4 years ago
parent bc1917dfee
commit d5c3f3a7f6
  1. 3
      os_brick/initiator/windows/smbfs.py
  2. 24
      os_brick/remotefs/windows_remotefs.py
  3. 42
      os_brick/tests/remotefs/test_windows_remotefs.py
  4. 4
      os_brick/tests/windows/test_smbfs.py

@ -102,8 +102,7 @@ class WindowsSMBFSConnector(win_conn_base.BaseWindowsConnector):
disk_dir = self._remotefsclient.get_mount_point(
export_path)
elif use_local_path:
share_name = self._remotefsclient.get_share_name(export_path)
disk_dir = self._remotefsclient.get_local_share_path(share_name)
disk_dir = self._remotefsclient.get_local_share_path(export_path)
disk_name = connection_properties['name']
disk_path = os.path.join(disk_dir, disk_name)

@ -57,19 +57,36 @@ class WindowsRemoteFsClient(remotefs.RemoteFsClient):
self._pathutils = utilsfactory.get_pathutils()
def get_local_share_path(self, share, expect_existing=True):
local_share_path = self._smbutils.get_smb_share_path(share)
share = self._get_share_norm_path(share)
share_name = self.get_share_name(share)
share_subdir = self.get_share_subdir(share)
is_local_share = self._smbutils.is_local_share(share)
if not is_local_share:
LOG.debug("Share '%s' is not exposed by the current host.", share)
local_share_path = None
else:
local_share_path = self._smbutils.get_smb_share_path(share_name)
if not local_share_path and expect_existing:
err_msg = _("Could not find the local "
"share path for %(share)s.")
raise exception.VolumePathsNotFound(err_msg % dict(share=share))
if local_share_path and share_subdir:
local_share_path = os.path.join(local_share_path, share_subdir)
return local_share_path
def _get_share_norm_path(self, share):
return share.replace('/', '\\')
def get_share_name(self, share):
return self._get_share_norm_path(share).lstrip('\\').split('\\', 1)[1]
return self._get_share_norm_path(share).lstrip('\\').split('\\')[1]
def get_share_subdir(self, share):
return "\\".join(
self._get_share_norm_path(share).lstrip('\\').split('\\')[2:])
def mount(self, share, flags=None):
share_norm_path = self._get_share_norm_path(share)
@ -102,9 +119,8 @@ class WindowsRemoteFsClient(remotefs.RemoteFsClient):
# what the caller will expect.
mnt_point = self.get_mount_point(share)
share_norm_path = self._get_share_norm_path(share)
share_name = self.get_share_name(share)
symlink_dest = (share_norm_path if not use_local_path
else self.get_local_share_path(share_name))
else self.get_local_share_path(share))
if not os.path.isdir(self._mount_base):
os.makedirs(self._mount_base)

@ -39,24 +39,44 @@ class WindowsRemotefsClientTestCase(base.TestCase):
self._smbutils = self._remotefs._smbutils
self._pathutils = self._remotefs._pathutils
@ddt.data({},
{'expect_existing': False},
{'local_path': mock.sentinel.local_path})
@ddt.data({'is_local_share': False},
{'expect_existing': False})
@ddt.unpack
def test_get_local_share_path(self, expect_existing=True,
local_path=None):
self._smbutils.get_smb_share_path.return_value = local_path
if not local_path and expect_existing:
def test_get_local_share_path_missing(self, expect_existing=True,
is_local_share=True):
self._smbutils.get_smb_share_path.return_value = None
self._smbutils.is_local_share.return_value = is_local_share
if expect_existing:
self.assertRaises(
exception.VolumePathsNotFound,
self._remotefs.get_local_share_path,
mock.sentinel.share_name,
self._FAKE_SHARE,
expect_existing=expect_existing)
else:
share_path = self._remotefs.get_local_share_path(
mock.sentinel.share_name,
self._FAKE_SHARE,
expect_existing=expect_existing)
self.assertEqual(local_path, share_path)
self.assertIsNone(share_path)
self.assertEqual(is_local_share,
self._smbutils.get_smb_share_path.called)
self._smbutils.is_local_share.assert_called_once_with(self._FAKE_SHARE)
@ddt.data({'share': '//addr/share_name/subdir_a/subdir_b',
'exp_path': r'C:\shared_dir\subdir_a\subdir_b'},
{'share': '//addr/share_name',
'exp_path': r'C:\shared_dir'})
@ddt.unpack
@mock.patch('os.path.join', lambda *args: '\\'.join(args))
def test_get_local_share_path(self, share, exp_path):
fake_local_path = 'C:\\shared_dir'
self._smbutils.get_smb_share_path.return_value = fake_local_path
share_path = self._remotefs.get_local_share_path(share)
self.assertEqual(exp_path, share_path)
self._smbutils.get_smb_share_path.assert_called_once_with(
'share_name')
def test_get_share_name(self):
resulted_name = self._remotefs.get_share_name(self._FAKE_SHARE)
@ -126,7 +146,7 @@ class WindowsRemotefsClientTestCase(base.TestCase):
if use_local_path:
mock_get_local_share_path.assert_called_once_with(
self._FAKE_SHARE_NAME)
self._FAKE_SHARE)
expected_symlink_target = mock_get_local_share_path.return_value
else:
expected_symlink_target = self._FAKE_SHARE.replace('/', '\\')

@ -149,10 +149,8 @@ class WindowsSMBFSConnectorTestCase(test_base.WindowsConnectorTestBase):
elif expecting_local:
self._connector._smbutils.is_local_share.assert_called_once_with(
fake_export_path)
self._remotefs.get_share_name.assert_called_once_with(
fake_export_path)
self._remotefs.get_local_share_path.assert_called_once_with(
fake_share_name)
fake_export_path)
def test_get_search_path(self):
search_path = self._connector.get_search_path()

Loading…
Cancel
Save