Edge Share - complete CRUD

Access Type - add get and set
Screened File Types - Add get, set, add and remove

Author: ygalblum

cterasdk/edge/shares.py

@@ -94,6 +94,24 @@ def set_share_winacls(self, name):
         logging.getLogger().error("Updating Windows file sharing access mode. %s", {'share': name, 'access': enum.Acl.WindowsNT})
         self._gateway.put('/config/fileservices/share/' + name + '/access', enum.Acl.WindowsNT)
 
+    def get_access_type(self, name):
+        """
+        Get the network share Windows File Sharing authentication mode
+
+        :param str name: The share name
+        """
+        return self._gateway.get('/config/fileservices/share/' + name + '/access')
+
+    def set_access_type(self, name, access):
+        """
+        Set the network share Windows File Sharing authentication mode
+
+        :param str name: The share name
+        :param cterasdk.edge.enum.Acl access: The Windows File Sharing authentication mode
+        """
+        logging.getLogger().info("Updating Windows file sharing access mode. %s", {'share': name, 'access': access})
+        self._gateway.put('/config/fileservices/share/' + name + '/access', access)
+
     def block_files(self, name, extensions):
         """
         Configure a share to block one or more file extensions
@@ -336,6 +354,68 @@ def entry_not_removed(entry):
         param = list(filter(entry_not_removed, self.get_trusted_nfs_clients(name)))
         self._gateway.put('/config/fileservices/share/' + name + '/trustedNFSClients', param)
 
+    def get_screened_file_types(self, name):
+        """
+        Get the share's current list of blocked file extensions
+
+        :param str name: The share name
+        """
+        return self._gateway.get('/config/fileservices/share/' + name + '/screenedFileTypes')
+
+    def set_screened_file_types(self, name, extensions):
+        """
+        Set the share's current list of blocked file extensions (override the current list)
+
+        :param str name: The share name
+        :param list[str] extensions: List of file extensions to block
+        """
+        share = self.get(name)
+        if share.access != enum.Acl.WindowsNT:
+            raise CTERAException('Cannot block file types on non Windows-ACL enabled shares', None, share=share.name, access=share.access)
+        logging.getLogger().info(
+            "Updating the list of blocked file extensions. %s",
+            {'share': name, 'extensions': extensions, 'access': enum.Acl.WindowsNT}
+        )
+        self._gateway.put('/config/fileservices/share/' + share.name + '/screenedFileTypes', extensions)
+
+    def add_screened_file_types(self, name, extensions):
+        """
+        Add extensions to the share's current list of blocked file extensions
+
+        :param str name: The share name
+        :param list[str] extensions: List of file extensions to add
+        """
+        share = self.get(name)
+        if share.access != enum.Acl.WindowsNT:
+            raise CTERAException('Cannot block file types on non Windows-ACL enabled shares', None, share=share.name, access=share.access)
+
+        new_list = list(set(share.screenedFileTypes + extensions))
+
+        logging.getLogger().info(
+            "Updating the list of blocked file extensions. %s",
+            {'share': name, 'extensions': new_list, 'access': enum.Acl.WindowsNT}
+        )
+        self._gateway.put('/config/fileservices/share/' + share.name + '/screenedFileTypes', new_list)
+
+    def remove_screened_file_types(self, name, extensions):
+        """
+        Remove extensions from the share's current list of blocked file extensions
+
+        :param str name: The share name
+        :param list[str] extensions: List of file extensions to remove
+        """
+        share = self.get(name)
+        if share.access != enum.Acl.WindowsNT:
+            raise CTERAException('Cannot block file types on non Windows-ACL enabled shares', None, share=share.name, access=share.access)
+
+        new_list = list(set(share.screenedFileTypes) - set(extensions))
+
+        logging.getLogger().info(
+            "Updating the list of blocked file extensions. %s",
+            {'share': name, 'extensions': new_list, 'access': enum.Acl.WindowsNT}
+        )
+        self._gateway.put('/config/fileservices/share/' + share.name + '/screenedFileTypes', new_list)
+
     def _validate_root_directory(self, name):
         param = Object()
         param.path = '/'

tests/ut/test_edge_shares.py

@@ -314,3 +314,106 @@ def test_remove_trusted_nfs_clients(self):
     @staticmethod
     def _get_get_trusted_nfs_client_object(address=None):
         return NFSv3AccessControlEntry(address=address or '192.168.68.0', netmask='255.255.255.0', perm=FileAccessMode.RO)
+
+    def _test_get_access_type(self, expected_access):
+        share_name = 'share'
+        self._init_filer(get_response=expected_access)
+        actual_access = shares.Shares(self._filer).get_access_type(share_name)
+        self._filer.get.assert_called_once_with('/config/fileservices/share/' + share_name + '/access')
+        self.assertEqual(actual_access, expected_access)
+
+    def test_get_access_type(self):
+        for access in [k for k in Acl.__dict__ if not k.startswith('_')]:
+            self._test_get_access_type(access)
+
+    def _test_set_access_type(self, access):
+        share_name = 'share'
+        self._init_filer()
+        shares.Shares(self._filer).set_access_type(share_name, access)
+        self._filer.put.assert_called_once_with('/config/fileservices/share/' + share_name + '/access', access)
+
+    def test_set_access_type(self):
+        for access in [k for k in Acl.__dict__ if not k.startswith('_')]:
+            self._test_set_access_type(access)
+
+    def test_get_screened_file_types(self):
+        share_name = 'share'
+        get_response = ['exe', 'sh']
+        self._init_filer(get_response=get_response)
+        screened_file_types = shares.Shares(self._filer).get_screened_file_types(share_name)
+        self._filer.get.assert_called_once_with('/config/fileservices/share/' + share_name + '/screenedFileTypes')
+        self.assertListEqual(get_response, screened_file_types)
+
+    def test_set_screened_file_types(self):
+        share_name = 'share'
+        current_screened_file_type = 'old'
+        current_share = self._test_screened_file_types_get_current_share(share_name, Acl.WindowsNT, [current_screened_file_type])
+        self._init_filer(get_response=current_share)
+        new_screened_file_type = ['new']
+        shares.Shares(self._filer).set_screened_file_types(share_name, new_screened_file_type)
+        self._filer.get.assert_called_once_with('/config/fileservices/share/' + share_name)
+        self._filer.put.assert_called_once_with('/config/fileservices/share/' + share_name + '/screenedFileTypes', mock.ANY)
+        self.assertListEqual(new_screened_file_type, self._filer.put.call_args[0][1])
+
+    def test_set_screened_file_types_invalid_access(self):
+        share_name = 'share'
+        current_share = self._test_screened_file_types_get_current_share(
+            share_name,
+            Acl.OnlyAuthenticatedUsers,
+            ['old']
+        )
+        self._init_filer(get_response=current_share)
+        with self.assertRaises(exception.CTERAException):
+            shares.Shares(self._filer).set_screened_file_types(share_name, ['new'])
+
+    def test_add_screened_file_types(self):
+        share_name = 'share'
+        current_screened_file_type = ['old']
+        current_share = self._test_screened_file_types_get_current_share(share_name, Acl.WindowsNT, current_screened_file_type)
+        self._init_filer(get_response=current_share)
+        new_screened_file_type = ['new']
+        shares.Shares(self._filer).add_screened_file_types(share_name, new_screened_file_type)
+        self._filer.get.assert_called_once_with('/config/fileservices/share/' + share_name)
+        self._filer.put.assert_called_once_with('/config/fileservices/share/' + share_name + '/screenedFileTypes', mock.ANY)
+        self.assertListEqual(sorted(current_screened_file_type + new_screened_file_type), sorted(self._filer.put.call_args[0][1]))
+
+    def test_add_screened_file_types_invalid_access(self):
+        share_name = 'share'
+        current_share = self._test_screened_file_types_get_current_share(
+            share_name,
+            Acl.OnlyAuthenticatedUsers,
+            ['old']
+        )
+        self._init_filer(get_response=current_share)
+        with self.assertRaises(exception.CTERAException):
+            shares.Shares(self._filer).add_screened_file_types(share_name, ['new'])
+
+    def test_remove_screened_file_types(self):
+        share_name = 'share'
+        current_screened_file_types = ['old', 'new']
+        current_share = self._test_screened_file_types_get_current_share(share_name, Acl.WindowsNT, current_screened_file_types)
+        self._init_filer(get_response=current_share)
+        removed_screened_file_type = 'new'
+        shares.Shares(self._filer).remove_screened_file_types(share_name, [removed_screened_file_type])
+        self._filer.get.assert_called_once_with('/config/fileservices/share/' + share_name)
+        self._filer.put.assert_called_once_with('/config/fileservices/share/' + share_name + '/screenedFileTypes', mock.ANY)
+        self.assertListEqual(['old'], sorted(self._filer.put.call_args[0][1]))
+
+    def test_remove_screened_file_types_invalid_access(self):
+        share_name = 'share'
+        current_share = self._test_screened_file_types_get_current_share(
+            share_name,
+            Acl.OnlyAuthenticatedUsers,
+            ['old']
+        )
+        self._init_filer(get_response=current_share)
+        with self.assertRaises(exception.CTERAException):
+            shares.Shares(self._filer).remove_screened_file_types(share_name, ['new'])
+
+    @staticmethod
+    def _test_screened_file_types_get_current_share(name, access, current_list):
+        share = Object()
+        share.name = name
+        share.access = access
+        share.screenedFileTypes = current_list
+        return share