mgr/smb: add heavily mocked unit test for fs.py module

Signed-off-by: John Mulligan <jmulligan@redhat.com>

diff --git a/src/pybind/mgr/smb/fs.py b/src/pybind/mgr/smb/fs.py
index 5a1ba2aabc716d0bbc63da98997514a31b2f384d..8aaa8bcde060abe2b743aae1fb05305b585d90d5 100644 (file)
--- a/src/pybind/mgr/smb/fs.py
+++ b/src/pybind/mgr/smb/fs.py
@@ -54,9 +54,11 @@ class CephFSPathResolver:
     map to real paths in the cephfs volume and determine if those paths exist.
     """
 
-    def __init__(self, mgr: Module_T) -> None:
+    def __init__(
+        self, mgr: Module_T, *, client: Optional[CephfsClient] = None
+    ) -> None:
         self._mgr = mgr
-        self._cephfs_client = CephfsClient(mgr)
+        self._cephfs_client = client or CephfsClient(mgr)
 
     def resolve(
         self, volume: str, subvolumegroup: str, subvolume: str, path: str

diff --git a/src/pybind/mgr/smb/tests/test_fs.py b/src/pybind/mgr/smb/tests/test_fs.py
new file mode 100644 (file)
index 0000000..7af20eb
--- /dev/null
+++ b/src/pybind/mgr/smb/tests/test_fs.py
@@ -0,0 +1,67 @@
+from unittest import mock
+
+import pytest
+
+import smb.fs
+
+
+def test_mocked_fs_authorizer():
+    def mmcmd(cmd):
+        assert cmd['filesystem'] == 'cephfs'
+        if 'kaboom' in cmd['entity']:
+            return -5, 'oops', 'fail'
+        return 0, 'ok', 'nice'
+
+    m = mock.MagicMock()
+    m.mon_command.side_effect = mmcmd
+
+    fsauth = smb.fs.FileSystemAuthorizer(m)
+    fsauth.authorize_entity('cephfs', 'client.smb.foo')
+    with pytest.raises(smb.fs.AuthorizationGrantError):
+        fsauth.authorize_entity('cephfs', 'client.smb.kaboom')
+
+
+def test_mocked_fs_path_resolver(monkeypatch):
+    # we have to "re-patch" whatever cephfs module gets mocked with because
+    # the ObjectNotFound attribute is not an exception in the test environment
+    monkeypatch.setattr('cephfs.ObjectNotFound', KeyError)
+
+    def mmcmd(cmd):
+        if cmd['prefix'] == 'fs subvolume getpath':
+            if cmd['vol_name'] == 'cephfs' and cmd['sub_name'] == 'beta':
+                return 0, '/volumes/cool/path/f00d-600d', ''
+        return -5, '', 'eek'
+
+    m = mock.MagicMock()
+    m.mon_command.side_effect = mmcmd
+
+    fspr = smb.fs.CephFSPathResolver(m, client=m)
+
+    # resolve
+    path = fspr.resolve('cephfs', '', '', '/zowie')
+    assert path == '/zowie'
+
+    path = fspr.resolve('cephfs', 'alpha', 'beta', '/zowie')
+    assert path == '/volumes/cool/path/f00d-600d/zowie'
+
+    with pytest.raises(smb.fs.CephFSSubvolumeResolutionError):
+        path = fspr.resolve('ouch', 'alpha', 'beta', '/zowie')
+
+    # resolve_exists
+    m.connection_pool.get_fs_handle.return_value.statx.return_value = {
+        'mode': 0o41777
+    }
+    path = fspr.resolve_exists('cephfs', 'alpha', 'beta', '/zowie')
+    assert path == '/volumes/cool/path/f00d-600d/zowie'
+
+    m.connection_pool.get_fs_handle.return_value.statx.return_value = {
+        'mode': 0o101777
+    }
+    with pytest.raises(NotADirectoryError):
+        fspr.resolve_exists('cephfs', 'alpha', 'beta', '/zowie')
+
+    m.connection_pool.get_fs_handle.return_value.statx.side_effect = (
+        mock.MagicMock(side_effect=OSError('nope'))
+    )
+    with pytest.raises(FileNotFoundError):
+        fspr.resolve_exists('cephfs', 'alpha', 'beta', '/zowie')