NAS-130922 / 25.04 / Remove REST from test_999_pool_dataset_unlock.py

src/middlewared/middlewared/test/integration/assets/pool.py

@@ -67,11 +67,10 @@ def dataset(name, data=None, pool=pool, **kwargs):
     call("pool.dataset.create", {"name": dataset, **data})
 
     try:
-        if "acl" in kwargs or "mode" in kwargs:
-            if "acl" in kwargs:
-                call("filesystem.setacl", {'path': f"/mnt/{dataset}", "dacl": kwargs['acl']})
-            else:
-                call("filesystem.setperm", {'path': f"/mnt/{dataset}", "mode": kwargs['mode'] or "777"})
+        if "acl" in kwargs:
+            call("filesystem.setacl", {'path': f"/mnt/{dataset}", "dacl": kwargs['acl']})
+        elif "mode" in kwargs:
+            call("filesystem.setperm", {'path': f"/mnt/{dataset}", "mode": kwargs['mode'] or "777"})
 
         yield dataset
     finally:

tests/api2/test_pool_dataset_unlock.py

@@ -0,0 +1,131 @@
+import contextlib
+
+import pytest
+
+from middlewared.test.integration.assets.account import user
+from middlewared.test.integration.assets.pool import dataset
+from middlewared.test.integration.assets.smb import smb_share
+from middlewared.test.integration.utils import call, ssh
+from protocols import SMB
+from samba import ntstatus, NTSTATUSError
+
+
+SMB_PASSWORD = 'Abcd1234'
+SMB_USER = 'smbuser999'
+
+
+def passphrase_encryption():
+    return {
+        'encryption_options': {
+            'generate_key': False,
+            'pbkdf2iters': 100000,
+            'algorithm': 'AES-128-CCM',
+            'passphrase': 'passphrase',
+        },
+        'encryption': True,
+        'inherit_encryption': False,
+    }
+
+def lock_dataset(name):
+    payload = {
+        'force_umount': True
+    }
+    assert call('pool.dataset.lock', name, payload, job=True)
+
+
+def unlock_dataset(name, options=None):
+    payload = {
+        'recursive': True,
+        'datasets': [
+            {
+                'name': name,
+                'passphrase': 'passphrase'
+            }
+        ],
+        **(options or {}),
+    }
+    result = call('pool.dataset.unlock', name, payload, job=True)
+    assert result['unlocked'] == [name], str(result)
+
+
+@contextlib.contextmanager
+def smb_connection(**kwargs):
+    c = SMB()
+    c.connect(**kwargs)
+
+    try:
+        yield c
+    finally:
+        c.disconnect()
+
+
+@pytest.fixture(scope='module')
+def smb_user():
+    with user({
+        'username': SMB_USER,
+        'full_name': 'doug',
+        'group_create': True,
+        'password': SMB_PASSWORD,
+        'smb': True
+    }, get_instance=True) as u:
+        yield u
+
+
+@pytest.mark.parametrize('toggle_attachments', [True, False])
+def test_pool_dataset_unlock_smb(smb_user, toggle_attachments):
+    with (
+        # Prepare test SMB share
+        dataset('normal', mode='777') as normal,
+        smb_share(f'/mnt/{normal}', 'normal', {'guestok': True}),
+        # Create an encrypted SMB share, unlocking which might lead to SMB service interruption
+        dataset('encrypted', passphrase_encryption(), mode='777') as encrypted,
+        smb_share(f'/mnt/{encrypted}', 'encrypted', {'guestok': True})
+    ):
+        ssh(f'touch /mnt/{encrypted}/secret')
+        assert call('service.start', 'cifs')
+        lock_dataset(encrypted)
+        # Mount test SMB share
+        with smb_connection(
+            share='normal',
+            username=SMB_USER,
+            password=SMB_PASSWORD
+        ) as normal_connection:
+            # Locked share should not be mountable
+            with pytest.raises(NTSTATUSError) as e:
+                with smb_connection(
+                    share='encrypted',
+                    username=SMB_USER,
+                    password=SMB_PASSWORD
+                ):
+                    pass
+
+            assert e.value.args[0] == ntstatus.NT_STATUS_BAD_NETWORK_NAME
+
+            conn = normal_connection.show_connection()
+            assert conn['connected'], conn
+            unlock_dataset(encrypted, {'toggle_attachments': toggle_attachments})
+
+            conn = normal_connection.show_connection()
+            assert conn['connected'], conn
+
+        if toggle_attachments:
+            # We should be able to mount encrypted share
+            with smb_connection(
+                share='encrypted',
+                username=SMB_USER,
+                password=SMB_PASSWORD
+            ) as encrypted_connection:
+                assert [x['name'] for x in encrypted_connection.ls('')] == ['secret']
+        else:
+            # We should still not be able to mount encrypted share as we did not reload attachments
+            with pytest.raises(NTSTATUSError) as e:
+                with smb_connection(
+                    share='encrypted',
+                    username=SMB_USER,
+                    password=SMB_PASSWORD
+                ):
+                    pass
+
+            assert e.value.args[0] == ntstatus.NT_STATUS_BAD_NETWORK_NAME
+
+    assert call('service.stop', 'cifs') is False  # successfully stopped