"Fossies" - the Fresh Open Source Software Archive

Member "manila-8.1.4/manila/tests/share/drivers/netapp/dataontap/protocols/test_cifs_cmode.py" (19 Nov 2020, 8414 Bytes) of package /linux/misc/openstack/manila-8.1.4.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_cifs_cmode.py": 8.1.3_vs_8.1.4.

    1 # Copyright (c) 2015 Clinton Knight.  All rights reserved.
    2 #
    3 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    4 #    not use this file except in compliance with the License. You may obtain
    5 #    a copy of the License at
    6 #
    7 #         http://www.apache.org/licenses/LICENSE-2.0
    8 #
    9 #    Unless required by applicable law or agreed to in writing, software
   10 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   11 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   12 #    License for the specific language governing permissions and limitations
   13 #    under the License.
   14 """
   15 Mock unit tests for the NetApp driver protocols CIFS class module.
   16 """
   17 
   18 import copy
   19 
   20 import ddt
   21 import mock
   22 
   23 from manila.common import constants
   24 from manila import exception
   25 from manila.share.drivers.netapp.dataontap.protocols import cifs_cmode
   26 from manila import test
   27 from manila.tests.share.drivers.netapp.dataontap.protocols \
   28     import fakes as fake
   29 
   30 
   31 @ddt.ddt
   32 class NetAppClusteredCIFSHelperTestCase(test.TestCase):
   33 
   34     def setUp(self):
   35         super(NetAppClusteredCIFSHelperTestCase, self).setUp()
   36 
   37         self.mock_context = mock.Mock()
   38 
   39         self.mock_client = mock.Mock()
   40         self.helper = cifs_cmode.NetAppCmodeCIFSHelper()
   41         self.helper.set_client(self.mock_client)
   42 
   43     def test_create_share(self):
   44 
   45         result = self.helper.create_share(fake.CIFS_SHARE, fake.SHARE_NAME)
   46 
   47         export_addresses = [fake.SHARE_ADDRESS_1, fake.SHARE_ADDRESS_2]
   48         export_paths = [result(address) for address in export_addresses]
   49         expected_paths = [
   50             r'\\%s\%s' % (fake.SHARE_ADDRESS_1, fake.SHARE_NAME),
   51             r'\\%s\%s' % (fake.SHARE_ADDRESS_2, fake.SHARE_NAME),
   52         ]
   53         self.assertEqual(expected_paths, export_paths)
   54         self.mock_client.create_cifs_share.assert_called_once_with(
   55             fake.SHARE_NAME)
   56         self.mock_client.remove_cifs_share_access.assert_called_once_with(
   57             fake.SHARE_NAME, 'Everyone')
   58         self.mock_client.set_volume_security_style.assert_called_once_with(
   59             fake.SHARE_NAME, security_style='ntfs')
   60 
   61     def test_delete_share(self):
   62 
   63         self.helper.delete_share(fake.CIFS_SHARE, fake.SHARE_NAME)
   64 
   65         self.mock_client.remove_cifs_share.assert_called_once_with(
   66             fake.SHARE_NAME)
   67 
   68     def test_update_access(self):
   69 
   70         mock_validate_access_rule = self.mock_object(self.helper,
   71                                                      '_validate_access_rule')
   72         mock_get_access_rules = self.mock_object(
   73             self.helper, '_get_access_rules',
   74             mock.Mock(return_value=fake.EXISTING_CIFS_RULES))
   75         mock_handle_added_rules = self.mock_object(self.helper,
   76                                                    '_handle_added_rules')
   77         mock_handle_ro_to_rw_rules = self.mock_object(self.helper,
   78                                                       '_handle_ro_to_rw_rules')
   79         mock_handle_rw_to_ro_rules = self.mock_object(self.helper,
   80                                                       '_handle_rw_to_ro_rules')
   81         mock_handle_deleted_rules = self.mock_object(self.helper,
   82                                                      '_handle_deleted_rules')
   83 
   84         self.helper.update_access(fake.CIFS_SHARE,
   85                                   fake.SHARE_NAME,
   86                                   [fake.USER_ACCESS])
   87 
   88         new_rules = {'fake_user': constants.ACCESS_LEVEL_RW}
   89         mock_validate_access_rule.assert_called_once_with(fake.USER_ACCESS)
   90         mock_get_access_rules.assert_called_once_with(fake.CIFS_SHARE,
   91                                                       fake.SHARE_NAME)
   92         mock_handle_added_rules.assert_called_once_with(
   93             fake.SHARE_NAME, fake.EXISTING_CIFS_RULES, new_rules)
   94         mock_handle_ro_to_rw_rules.assert_called_once_with(
   95             fake.SHARE_NAME, fake.EXISTING_CIFS_RULES, new_rules)
   96         mock_handle_rw_to_ro_rules.assert_called_once_with(
   97             fake.SHARE_NAME, fake.EXISTING_CIFS_RULES, new_rules)
   98         mock_handle_deleted_rules.assert_called_once_with(
   99             fake.SHARE_NAME, fake.EXISTING_CIFS_RULES, new_rules)
  100 
  101     def test_validate_access_rule(self):
  102 
  103         result = self.helper._validate_access_rule(fake.USER_ACCESS)
  104 
  105         self.assertIsNone(result)
  106 
  107     def test_validate_access_rule_invalid_type(self):
  108 
  109         rule = copy.copy(fake.USER_ACCESS)
  110         rule['access_type'] = 'ip'
  111 
  112         self.assertRaises(exception.InvalidShareAccess,
  113                           self.helper._validate_access_rule,
  114                           rule)
  115 
  116     def test_validate_access_rule_invalid_level(self):
  117 
  118         rule = copy.copy(fake.USER_ACCESS)
  119         rule['access_level'] = 'none'
  120 
  121         self.assertRaises(exception.InvalidShareAccessLevel,
  122                           self.helper._validate_access_rule,
  123                           rule)
  124 
  125     def test_handle_added_rules(self):
  126 
  127         self.helper._handle_added_rules(fake.SHARE_NAME,
  128                                         fake.EXISTING_CIFS_RULES,
  129                                         fake.NEW_CIFS_RULES)
  130 
  131         self.mock_client.add_cifs_share_access.assert_has_calls([
  132             mock.call(fake.SHARE_NAME, 'user5', False),
  133             mock.call(fake.SHARE_NAME, 'user6', True),
  134         ], any_order=True)
  135 
  136     def test_handle_ro_to_rw_rules(self):
  137 
  138         self.helper._handle_ro_to_rw_rules(fake.SHARE_NAME,
  139                                            fake.EXISTING_CIFS_RULES,
  140                                            fake.NEW_CIFS_RULES)
  141 
  142         self.mock_client.modify_cifs_share_access.assert_has_calls([
  143             mock.call(fake.SHARE_NAME, 'user2', False)
  144         ])
  145 
  146     def test_handle_rw_to_ro_rules(self):
  147 
  148         self.helper._handle_rw_to_ro_rules(fake.SHARE_NAME,
  149                                            fake.EXISTING_CIFS_RULES,
  150                                            fake.NEW_CIFS_RULES)
  151 
  152         self.mock_client.modify_cifs_share_access.assert_has_calls([
  153             mock.call(fake.SHARE_NAME, 'user3', True)
  154         ])
  155 
  156     def test_handle_deleted_rules(self):
  157 
  158         self.helper._handle_deleted_rules(fake.SHARE_NAME,
  159                                           fake.EXISTING_CIFS_RULES,
  160                                           fake.NEW_CIFS_RULES)
  161 
  162         self.mock_client.remove_cifs_share_access.assert_has_calls([
  163             mock.call(fake.SHARE_NAME, 'user4')
  164         ])
  165 
  166     def test_get_access_rules(self):
  167 
  168         self.mock_client.get_cifs_share_access = (
  169             mock.Mock(return_value='fake_rules'))
  170 
  171         result = self.helper._get_access_rules(fake.CIFS_SHARE,
  172                                                fake.SHARE_NAME)
  173 
  174         self.assertEqual('fake_rules', result)
  175         self.mock_client.get_cifs_share_access.assert_called_once_with(
  176             fake.SHARE_NAME)
  177 
  178     def test_get_target(self):
  179 
  180         target = self.helper.get_target(fake.CIFS_SHARE)
  181         self.assertEqual(fake.SHARE_ADDRESS_1, target)
  182 
  183     def test_get_target_missing_location(self):
  184 
  185         target = self.helper.get_target({'export_location': ''})
  186         self.assertEqual('', target)
  187 
  188     def test_get_share_name_for_share(self):
  189 
  190         share_name = self.helper.get_share_name_for_share(fake.CIFS_SHARE)
  191 
  192         self.assertEqual(fake.SHARE_NAME, share_name)
  193 
  194     @ddt.data(
  195         {
  196             'location': r'\\%s\%s' % (fake.SHARE_ADDRESS_1, fake.SHARE_NAME),
  197             'ip': fake.SHARE_ADDRESS_1,
  198             'share_name': fake.SHARE_NAME,
  199         }, {
  200             'location': r'//%s/%s' % (fake.SHARE_ADDRESS_1, fake.SHARE_NAME),
  201             'ip': fake.SHARE_ADDRESS_1,
  202             'share_name': fake.SHARE_NAME,
  203         },
  204         {'location': '', 'ip': '', 'share_name': ''},
  205         {'location': 'invalid', 'ip': '', 'share_name': ''},
  206     )
  207     @ddt.unpack
  208     def test_get_export_location(self, location, ip, share_name):
  209 
  210         share = fake.CIFS_SHARE.copy()
  211         share['export_location'] = location
  212 
  213         result_ip, result_share_name = self.helper._get_export_location(share)
  214 
  215         self.assertEqual(ip, result_ip)
  216         self.assertEqual(share_name, result_share_name)
  217 
  218     def test_cleanup_demoted_replica(self):
  219         self.helper.cleanup_demoted_replica(fake.CIFS_SHARE, fake.SHARE_NAME)
  220 
  221         self.mock_client.remove_cifs_share.assert_called_once_with(
  222             fake.SHARE_NAME)