"Fossies" - the Fresh Open Source Software Archive

Member "manila-8.1.4/manila/tests/share/drivers/dell_emc/plugins/unity/test_connection.py" (19 Nov 2020, 27827 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_connection.py": 8.1.3_vs_8.1.4.

    1 # Copyright (c) 2016 EMC Corporation.
    2 # All Rights Reserved.
    3 #
    4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    5 #    not use this file except in compliance with the License. You may obtain
    6 #    a copy of the License at
    7 #
    8 #         http://www.apache.org/licenses/LICENSE-2.0
    9 #
   10 #    Unless required by applicable law or agreed to in writing, software
   11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   13 #    License for the specific language governing permissions and limitations
   14 #    under the License.
   15 
   16 import copy
   17 import ddt
   18 import mock
   19 import six
   20 
   21 from manila import exception
   22 from manila.share.drivers.dell_emc.common.enas import utils as enas_utils
   23 from manila import test
   24 from manila.tests.share.drivers.dell_emc.plugins.unity import fake_exceptions
   25 from manila.tests.share.drivers.dell_emc.plugins.unity import res_mock
   26 from manila.tests.share.drivers.dell_emc.plugins.unity import utils
   27 from oslo_utils import units
   28 
   29 
   30 @ddt.ddt
   31 class TestConnection(test.TestCase):
   32     client = None
   33 
   34     @classmethod
   35     def setUpClass(cls):
   36         cls.emc_share_driver = res_mock.FakeEMCShareDriver()
   37 
   38     @res_mock.patch_connection_init
   39     def test_connect(self, connection):
   40         connection.connect(res_mock.FakeEMCShareDriver(), None)
   41 
   42     @res_mock.patch_connection_init
   43     def test_connect_with_ipv6(self, connection):
   44         connection.connect(res_mock.FakeEMCShareDriverIPv6(),
   45                            None)
   46 
   47     @res_mock.patch_connection
   48     def test_connect__invalid_pool_configuration(self, connection):
   49         f = connection.client.system.get_pool
   50         f.side_effect = fake_exceptions.UnityResourceNotFoundError()
   51 
   52         self.assertRaises(exception.BadConfigurationException,
   53                           connection._config_pool,
   54                           'faked_pool_name')
   55 
   56     @res_mock.mock_manila_input
   57     @res_mock.patch_connection
   58     def test_create_nfs_share(self, connection, mocked_input):
   59         share = mocked_input['nfs_share']
   60         share_server = mocked_input['share_server']
   61 
   62         location = connection.create_share(None, share, share_server)
   63 
   64         exp_location = [
   65             {'path': 'fake_ip_addr_1:/cb532599-8dc6-4c3e-bb21-74ea54be566c'},
   66             {'path': 'fake_ip_addr_2:/cb532599-8dc6-4c3e-bb21-74ea54be566c'},
   67         ]
   68         exp_location = sorted(exp_location, key=lambda x: sorted(x['path']))
   69         location = sorted(location, key=lambda x: sorted(x['path']))
   70         self.assertEqual(exp_location, location)
   71 
   72     @res_mock.mock_manila_input
   73     @res_mock.patch_connection
   74     def test_create_cifs_share(self, connection, mocked_input):
   75         share = mocked_input['cifs_share']
   76         share_server = mocked_input['share_server']
   77 
   78         location = connection.create_share(None, share, share_server)
   79 
   80         exp_location = [
   81             {'path': r'\\fake_ip_addr_1\716100cc-e0b4-416b-ac27-d38dd019330d'},
   82             {'path': r'\\fake_ip_addr_2\716100cc-e0b4-416b-ac27-d38dd019330d'},
   83         ]
   84         exp_location = sorted(exp_location, key=lambda x: sorted(x['path']))
   85         location = sorted(location, key=lambda x: sorted(x['path']))
   86         self.assertEqual(exp_location, location)
   87 
   88     @res_mock.mock_manila_input
   89     @res_mock.patch_connection
   90     def test_create_share_with_invalid_proto(self, connection, mocked_input):
   91         share = mocked_input['invalid_share']
   92         share_server = mocked_input['share_server']
   93 
   94         self.assertRaises(exception.InvalidShare,
   95                           connection.create_share,
   96                           None,
   97                           share,
   98                           share_server)
   99 
  100     @res_mock.mock_manila_input
  101     @res_mock.patch_connection
  102     def test_create_share_without_share_server(self, connection,
  103                                                mocked_input):
  104         share = mocked_input['cifs_share']
  105 
  106         self.assertRaises(exception.InvalidInput,
  107                           connection.create_share,
  108                           None,
  109                           share,
  110                           None)
  111 
  112     @res_mock.mock_manila_input
  113     @res_mock.patch_connection
  114     def test_create_share__no_server_name_in_backend_details(self, connection,
  115                                                              mocked_input):
  116         share = mocked_input['cifs_share']
  117         share_server = {
  118             'backend_details': {'share_server_name': None},
  119             'id': 'test',
  120         }
  121 
  122         self.assertRaises(exception.InvalidInput,
  123                           connection.create_share,
  124                           None,
  125                           share,
  126                           share_server)
  127 
  128     @res_mock.mock_manila_input
  129     @res_mock.patch_connection
  130     def test_create_share_with_invalid_share_server(self, connection,
  131                                                     mocked_input):
  132         share = mocked_input['cifs_share']
  133         share_server = mocked_input['share_server']
  134 
  135         self.assertRaises(exception.EMCUnityError,
  136                           connection.create_share,
  137                           None,
  138                           share,
  139                           share_server)
  140 
  141     @res_mock.mock_manila_input
  142     @res_mock.patch_connection
  143     def test_delete_share(self, connection, mocked_input):
  144         share = mocked_input['cifs_share']
  145         share_server = mocked_input['share_server']
  146 
  147         connection.delete_share(None, share, share_server)
  148 
  149     @res_mock.mock_manila_input
  150     @res_mock.patch_connection
  151     def test_delete_share__with_invalid_share(self, connection, mocked_input):
  152         share = mocked_input['cifs_share']
  153 
  154         connection.delete_share(None, share, None)
  155 
  156     @res_mock.mock_manila_input
  157     @res_mock.patch_connection
  158     def test_delete_share__create_from_snap(self, connection,
  159                                             mocked_input):
  160         share = mocked_input['cifs_share']
  161         share_server = mocked_input['share_server']
  162 
  163         connection.delete_share(None, share, share_server)
  164 
  165     @res_mock.mock_manila_input
  166     @res_mock.patch_connection
  167     def test_delete_share__create_from_snap_but_not_isolated(self,
  168                                                              connection,
  169                                                              mocked_input):
  170         share = mocked_input['cifs_share']
  171         share_server = mocked_input['share_server']
  172 
  173         connection.delete_share(None, share, share_server)
  174 
  175     @res_mock.mock_manila_input
  176     @res_mock.patch_connection
  177     def test_delete_share__but_not_isolated(self, connection,
  178                                             mocked_input):
  179         share = mocked_input['cifs_share']
  180         share_server = mocked_input['share_server']
  181 
  182         connection.delete_share(None, share, share_server)
  183 
  184     @res_mock.mock_manila_input
  185     @res_mock.patch_connection
  186     def test_shrink_cifs_share(self, connection, mocked_input):
  187         share = mocked_input['shrink_cifs_share']
  188         new_size = 4 * units.Gi
  189 
  190         connection.shrink_share(share, new_size)
  191 
  192     @res_mock.mock_manila_input
  193     @res_mock.patch_connection
  194     def test_shrink_nfs_share(self, connection, mocked_input):
  195         share = mocked_input['shrink_nfs_share']
  196         new_size = 4 * units.Gi
  197 
  198         connection.shrink_share(share, new_size)
  199 
  200     @res_mock.mock_manila_input
  201     @res_mock.patch_connection
  202     def test_extend_cifs_share(self, connection, mocked_input):
  203         share = mocked_input['cifs_share']
  204         share_server = mocked_input['share_server']
  205         new_size = 50 * units.Gi
  206 
  207         connection.extend_share(share, new_size, share_server)
  208 
  209     @res_mock.mock_manila_input
  210     @res_mock.patch_connection
  211     def test_extend_nfs_share(self, connection, mocked_input):
  212         share = mocked_input['nfs_share']
  213         share_server = mocked_input['share_server']
  214         new_size = 50 * units.Gi
  215 
  216         connection.extend_share(share, new_size, share_server)
  217 
  218     @res_mock.mock_manila_input
  219     @res_mock.patch_connection
  220     def test_extend_share__create_from_snap(self, connection, mocked_input):
  221         share = mocked_input['cifs_share']
  222         share_server = mocked_input['share_server']
  223         new_size = 50 * units.Gi
  224 
  225         self.assertRaises(exception.ShareExtendingError,
  226                           connection.extend_share,
  227                           share,
  228                           new_size,
  229                           share_server)
  230 
  231     @res_mock.mock_manila_input
  232     @res_mock.patch_connection
  233     def test_shrink_share_create_from_snap(self, connection, mocked_input):
  234         share = mocked_input['shrink_cifs_share']
  235         share_server = mocked_input['share_server']
  236         new_size = 4 * units.Gi
  237 
  238         self.assertRaises(exception.ShareShrinkingError,
  239                           connection.shrink_share,
  240                           share,
  241                           new_size,
  242                           share_server)
  243 
  244     @res_mock.mock_manila_input
  245     @res_mock.patch_connection
  246     def test_create_snapshot_from_filesystem(self, connection, mocked_input):
  247         snapshot = mocked_input['snapshot']
  248         share_server = mocked_input['share_server']
  249 
  250         connection.create_snapshot(None, snapshot, share_server)
  251 
  252     @res_mock.mock_manila_input
  253     @res_mock.patch_connection
  254     def test_create_snapshot_from_snapshot(self, connection, mocked_input):
  255         snapshot = mocked_input['snapshot']
  256         share_server = mocked_input['share_server']
  257 
  258         connection.create_snapshot(None, snapshot, share_server)
  259 
  260     @res_mock.mock_manila_input
  261     @res_mock.patch_connection
  262     def test_delete_snapshot(self, connection, mocked_input):
  263         snapshot = mocked_input['snapshot']
  264         share_server = mocked_input['share_server']
  265 
  266         connection.delete_snapshot(None, snapshot, share_server)
  267 
  268     @res_mock.mock_manila_input
  269     @res_mock.patch_connection
  270     def test_ensure_share_exists(self, connection, mocked_input):
  271         share = mocked_input['cifs_share']
  272 
  273         connection.ensure_share(None, share, None)
  274 
  275     @res_mock.mock_manila_input
  276     @res_mock.patch_connection
  277     def test_ensure_share_not_exists(self, connection, mocked_input):
  278         share = mocked_input['cifs_share']
  279 
  280         self.assertRaises(exception.ShareNotFound,
  281                           connection.ensure_share,
  282                           None,
  283                           share,
  284                           None)
  285 
  286     @res_mock.patch_connection
  287     def test_update_share_stats(self, connection):
  288         stat_dict = copy.deepcopy(res_mock.STATS)
  289         connection.update_share_stats(stat_dict)
  290         self.assertEqual(5, len(stat_dict))
  291         pool = stat_dict['pools'][0]
  292         self.assertEqual('pool_1', pool['pool_name'])
  293         self.assertEqual(
  294             enas_utils.bytes_to_gb(500000.0), pool['total_capacity_gb'])
  295         self.assertEqual(False, pool['qos'])
  296         self.assertEqual(
  297             enas_utils.bytes_to_gb(30000.0), pool['provisioned_capacity_gb'])
  298         self.assertEqual(20, pool['max_over_subscription_ratio'])
  299         self.assertEqual(
  300             enas_utils.bytes_to_gb(10000.0), pool['allocated_capacity_gb'])
  301         self.assertEqual(0, pool['reserved_percentage'])
  302         self.assertTrue(pool['thin_provisioning'])
  303         self.assertEqual(
  304             enas_utils.bytes_to_gb(490000.0), pool['free_capacity_gb'])
  305 
  306     @res_mock.patch_connection
  307     def test_update_share_stats__nonexistent_pools(self, connection):
  308         stat_dict = copy.deepcopy(res_mock.STATS)
  309 
  310         self.assertRaises(exception.EMCUnityError,
  311                           connection.update_share_stats,
  312                           stat_dict)
  313 
  314     @res_mock.mock_manila_input
  315     @res_mock.patch_connection
  316     def test_get_pool(self, connection, mocked_input):
  317         share = mocked_input['cifs_share']
  318 
  319         connection.get_pool(share)
  320 
  321     @utils.patch_find_ports_by_mtu
  322     @res_mock.mock_manila_input
  323     @res_mock.patch_connection
  324     def test_setup_server(self, connection, mocked_input, find_ports):
  325         find_ports.return_value = {'SPA': {'spa_eth1'}}
  326         network_info = mocked_input['network_info__flat']
  327         server_info = connection.setup_server(network_info)
  328         self.assertEqual(
  329             {'share_server_name':
  330              '78fd845f-8e7d-487f-bfde-051d83e78103'},
  331             server_info)
  332         self.assertIsNone(connection.client.system.create_nas_server.
  333                           call_args[1]['tenant'])
  334 
  335     @utils.patch_find_ports_by_mtu
  336     @res_mock.mock_manila_input
  337     @res_mock.patch_connection
  338     def test_setup_server__vlan_network(self, connection, mocked_input,
  339                                         find_ports):
  340         find_ports.return_value = {'SPA': {'spa_eth1'}}
  341         network_info = mocked_input['network_info__vlan']
  342 
  343         connection.setup_server(network_info)
  344         self.assertEqual('tenant_1',
  345                          connection.client.system.create_nas_server
  346                          .call_args[1]['tenant'].id)
  347 
  348     @utils.patch_find_ports_by_mtu
  349     @res_mock.mock_manila_input
  350     @res_mock.patch_connection
  351     def test_setup_server__vxlan_network(self, connection, mocked_input,
  352                                          find_ports):
  353         find_ports.return_value = {'SPA': {'spa_eth1'}}
  354         network_info = mocked_input['network_info__vxlan']
  355 
  356         self.assertRaises(exception.NetworkBadConfigurationException,
  357                           connection.setup_server,
  358                           network_info)
  359 
  360     @utils.patch_find_ports_by_mtu
  361     @res_mock.mock_manila_input
  362     @res_mock.patch_connection
  363     def test_setup_server__active_directory(self, connection, mocked_input,
  364                                             find_ports):
  365         find_ports.return_value = {'SPA': {'spa_eth1'}}
  366         network_info = mocked_input['network_info__active_directory']
  367 
  368         connection.setup_server(network_info)
  369 
  370     @utils.patch_find_ports_by_mtu
  371     @res_mock.mock_manila_input
  372     @res_mock.patch_connection
  373     def test_setup_server__kerberos(self, connection, mocked_input,
  374                                     find_ports):
  375         find_ports.return_value = {'SPA': {'spa_eth1'}}
  376         network_info = mocked_input['network_info__kerberos']
  377 
  378         connection.setup_server(network_info)
  379 
  380     @utils.patch_find_ports_by_mtu
  381     @res_mock.mock_manila_input
  382     @res_mock.patch_connection
  383     def test_setup_server__throw_exception(self, connection, mocked_input,
  384                                            find_ports):
  385         find_ports.return_value = {'SPA': {'spa_eth1'}}
  386         network_info = mocked_input['network_info__flat']
  387 
  388         self.assertRaises(fake_exceptions.UnityException,
  389                           connection.setup_server,
  390                           network_info)
  391 
  392     @res_mock.mock_manila_input
  393     @res_mock.patch_connection
  394     def test_teardown_server(self, connection, mocked_input):
  395         server_detail = mocked_input['server_detail']
  396         security_services = mocked_input['security_services']
  397 
  398         connection.teardown_server(server_detail, security_services)
  399 
  400     @res_mock.mock_manila_input
  401     @res_mock.patch_connection
  402     def test_teardown_server__no_server_detail(self, connection, mocked_input):
  403         security_services = mocked_input['security_services']
  404 
  405         connection.teardown_server(None, security_services)
  406 
  407     @res_mock.mock_manila_input
  408     @res_mock.patch_connection
  409     def test_teardown_server__no_share_server_name(self, connection,
  410                                                    mocked_input):
  411         server_detail = {'share_server_name': None}
  412         security_services = mocked_input['security_services']
  413 
  414         connection.teardown_server(server_detail, security_services)
  415 
  416     @ddt.data({'configured_pools': None,
  417                'matched_pools': {'pool_1', 'pool_2', 'nas_server_pool'}},
  418               {'configured_pools': ['*'],
  419                'matched_pools': {'pool_1', 'pool_2', 'nas_server_pool'}},
  420               {'configured_pools': ['pool_*'],
  421                'matched_pools': {'pool_1', 'pool_2'}},
  422               {'configured_pools': ['*pool'],
  423                'matched_pools': {'nas_server_pool'}},
  424               {'configured_pools': ['nas_server_pool'],
  425                'matched_pools': {'nas_server_pool'}},
  426               {'configured_pools': ['nas_*', 'pool_*'],
  427                'matched_pools': {'pool_1', 'pool_2', 'nas_server_pool'}})
  428     @res_mock.patch_connection
  429     @ddt.unpack
  430     def test__get_managed_pools(self, connection, mocked_input):
  431         configured_pools = mocked_input['configured_pools']
  432         matched_pool = mocked_input['matched_pools']
  433 
  434         pools = connection._get_managed_pools(configured_pools)
  435 
  436         self.assertEqual(matched_pool, pools)
  437 
  438     @res_mock.patch_connection
  439     def test__get_managed_pools__invalid_pool_configuration(self, connection):
  440         configured_pools = 'fake_pool'
  441 
  442         self.assertRaises(exception.BadConfigurationException,
  443                           connection._get_managed_pools,
  444                           configured_pools)
  445 
  446     @res_mock.patch_connection
  447     def test_validate_port_configuration(self, connection):
  448         sp_ports_map = connection.validate_port_configuration(['sp*'])
  449 
  450         self.assertEqual({'spa_eth1', 'spa_eth2', 'spa_la_4'},
  451                          sp_ports_map['SPA'])
  452         self.assertEqual({'spb_eth1'}, sp_ports_map['SPB'])
  453 
  454     @res_mock.patch_connection
  455     def test_validate_port_configuration_exception(self, connection):
  456         self.assertRaises(exception.BadConfigurationException,
  457                           connection.validate_port_configuration,
  458                           ['xxxx*'])
  459 
  460     @res_mock.patch_connection
  461     def test__get_pool_name_from_host__no_pool_name(self, connection):
  462         host = 'openstack@Unity'
  463 
  464         self.assertRaises(exception.InvalidHost,
  465                           connection._get_pool_name_from_host,
  466                           host)
  467 
  468     @res_mock.mock_manila_input
  469     @res_mock.patch_connection
  470     def test_create_cifs_share_from_snapshot(self, connection, mocked_input):
  471         share = mocked_input['cifs_share']
  472         snapshot = mocked_input['snapshot']
  473         share_server = mocked_input['share_server']
  474 
  475         connection.create_share_from_snapshot(None, share, snapshot,
  476                                               share_server)
  477 
  478     @res_mock.mock_manila_input
  479     @res_mock.patch_connection
  480     def test_create_nfs_share_from_snapshot(self, connection, mocked_input):
  481         share = mocked_input['nfs_share']
  482         snapshot = mocked_input['snapshot']
  483         share_server = mocked_input['share_server']
  484 
  485         connection.create_share_from_snapshot(None, share, snapshot,
  486                                               share_server)
  487 
  488     @res_mock.mock_manila_input
  489     @res_mock.patch_connection
  490     def test_create_share_from_snapshot_no_server_name(self,
  491                                                        connection,
  492                                                        mocked_input):
  493         share = mocked_input['nfs_share']
  494         snapshot = mocked_input['snapshot']
  495         share_server = mocked_input['share_server__no_share_server_name']
  496 
  497         self.assertRaises(exception.EMCUnityError,
  498                           connection.create_share_from_snapshot,
  499                           None,
  500                           share,
  501                           snapshot,
  502                           share_server)
  503 
  504     @res_mock.mock_manila_input
  505     @res_mock.patch_connection
  506     def test_clear_share_access_cifs(self, connection, mocked_input):
  507         share = mocked_input['cifs_share']
  508 
  509         self.assertRaises(fake_exceptions.UnityException,
  510                           connection.clear_access,
  511                           share)
  512 
  513     @res_mock.mock_manila_input
  514     @res_mock.patch_connection
  515     def test_clear_share_access_nfs(self, connection, mocked_input):
  516         share = mocked_input['nfs_share']
  517 
  518         self.assertRaises(fake_exceptions.UnityException,
  519                           connection.clear_access,
  520                           share)
  521 
  522     @res_mock.mock_manila_input
  523     @res_mock.patch_connection
  524     def test_allow_rw_cifs_share_access(self, connection, mocked_input):
  525         share = mocked_input['cifs_share']
  526         rw_access = mocked_input['cifs_rw_access']
  527         share_server = mocked_input['share_server']
  528 
  529         connection.allow_access(None, share, rw_access, share_server)
  530 
  531     @res_mock.mock_manila_input
  532     @res_mock.patch_connection
  533     def test_update_access_allow_rw(self, connection, mocked_input):
  534         share = mocked_input['cifs_share']
  535         rw_access = mocked_input['cifs_rw_access']
  536         share_server = mocked_input['share_server']
  537 
  538         connection.update_access(None, share, None, [rw_access], None,
  539                                  share_server)
  540 
  541     @res_mock.mock_manila_input
  542     @res_mock.patch_connection
  543     def test_update_access_recovery(self, connection, mocked_input):
  544         share = mocked_input['cifs_share']
  545         rw_access = mocked_input['cifs_rw_access']
  546         share_server = mocked_input['share_server']
  547 
  548         connection.update_access(None, share, [rw_access], None, None,
  549                                  share_server)
  550 
  551     @res_mock.mock_manila_input
  552     @res_mock.patch_connection
  553     def test_allow_ro_cifs_share_access(self, connection, mocked_input):
  554         share = mocked_input['cifs_share']
  555         rw_access = mocked_input['cifs_ro_access']
  556         share_server = mocked_input['share_server']
  557 
  558         connection.allow_access(None, share, rw_access, share_server)
  559 
  560     @res_mock.mock_manila_input
  561     @res_mock.patch_connection
  562     def test_allow_rw_nfs_share_access(self, connection, mocked_input):
  563         share = mocked_input['nfs_share']
  564         rw_access = mocked_input['nfs_rw_access']
  565         share_server = mocked_input['share_server']
  566 
  567         connection.allow_access(None, share, rw_access, share_server)
  568 
  569     @res_mock.mock_manila_input
  570     @res_mock.patch_connection
  571     def test_allow_rw_nfs_share_access_cidr(self, connection, mocked_input):
  572         share = mocked_input['nfs_share']
  573         rw_access = mocked_input['nfs_rw_access_cidr']
  574         share_server = mocked_input['share_server']
  575 
  576         connection.allow_access(None, share, rw_access, share_server)
  577 
  578     @res_mock.mock_manila_input
  579     @res_mock.patch_connection
  580     def test_allow_ro_nfs_share_access(self, connection, mocked_input):
  581         share = mocked_input['nfs_share']
  582         ro_access = mocked_input['nfs_ro_access']
  583         share_server = mocked_input['share_server']
  584 
  585         connection.allow_access(None, share, ro_access, share_server)
  586 
  587     @res_mock.mock_manila_input
  588     @res_mock.patch_connection
  589     def test_deny_cifs_share_access(self, connection, mocked_input):
  590         share = mocked_input['cifs_share']
  591         rw_access = mocked_input['cifs_rw_access']
  592         share_server = mocked_input['share_server']
  593 
  594         connection.deny_access(None, share, rw_access, share_server)
  595 
  596     @res_mock.mock_manila_input
  597     @res_mock.patch_connection
  598     def test_deny_nfs_share_access(self, connection, mocked_input):
  599         share = mocked_input['nfs_share']
  600         rw_access = mocked_input['nfs_rw_access']
  601         share_server = mocked_input['share_server']
  602 
  603         connection.deny_access(None, share, rw_access, share_server)
  604 
  605     @res_mock.mock_manila_input
  606     @res_mock.patch_connection
  607     def test_update_access_deny_nfs(self, connection, mocked_input):
  608         share = mocked_input['nfs_share']
  609         rw_access = mocked_input['nfs_rw_access']
  610 
  611         connection.update_access(None, share, None, None, [rw_access], None)
  612 
  613     @res_mock.mock_manila_input
  614     @res_mock.patch_connection
  615     def test__validate_cifs_share_access_type(self, connection, mocked_input):
  616         share = mocked_input['cifs_share']
  617         rw_access = mocked_input['invalid_access']
  618 
  619         self.assertRaises(exception.InvalidShareAccess,
  620                           connection._validate_share_access_type,
  621                           share,
  622                           rw_access)
  623 
  624     @res_mock.mock_manila_input
  625     @res_mock.patch_connection
  626     def test__validate_nfs_share_access_type(self, connection, mocked_input):
  627         share = mocked_input['nfs_share']
  628         rw_access = mocked_input['invalid_access']
  629 
  630         self.assertRaises(exception.InvalidShareAccess,
  631                           connection._validate_share_access_type,
  632                           share,
  633                           rw_access)
  634 
  635     @res_mock.patch_connection
  636     def test_get_network_allocations_number(self, connection):
  637         self.assertEqual(1, connection.get_network_allocations_number())
  638 
  639     @res_mock.patch_connection
  640     def test_get_proto_enum(self, connection):
  641         self.assertIn('FSSupportedProtocolEnum.CIFS',
  642                       six.text_type(connection._get_proto_enum('CIFS')))
  643         self.assertIn('FSSupportedProtocolEnum.NFS',
  644                       six.text_type(connection._get_proto_enum('nfs')))
  645 
  646     @res_mock.mock_manila_input
  647     @res_mock.patch_connection
  648     def test_allow_access_error_access_level(self, connection, mocked_input):
  649         share = mocked_input['nfs_share']
  650         rw_access = mocked_input['invalid_access']
  651 
  652         self.assertRaises(exception.InvalidShareAccessLevel,
  653                           connection.allow_access,
  654                           None, share, rw_access)
  655 
  656     @res_mock.patch_connection
  657     def test__create_network_interface_ipv6(self, connection):
  658         connection.client.create_interface = mock.Mock(return_value=None)
  659         nas_server = mock.Mock()
  660         network = {'ip_address': '2001:db8:0:1:f816:3eff:fe76:35c4',
  661                    'cidr': '2001:db8:0:1:f816:3eff:fe76:35c4/64',
  662                    'gateway': '2001:db8:0:1::1',
  663                    'segmentation_id': '201'}
  664         port_id = mock.Mock()
  665         connection._create_network_interface(nas_server, network, port_id)
  666 
  667         expected = {'ip_addr': '2001:db8:0:1:f816:3eff:fe76:35c4',
  668                     'netmask': None,
  669                     'gateway': '2001:db8:0:1::1',
  670                     'port_id': port_id,
  671                     'vlan_id': '201',
  672                     'prefix_length': '64'}
  673         connection.client.create_interface.assert_called_once_with(nas_server,
  674                                                                    **expected)
  675 
  676     @res_mock.patch_connection
  677     def test__create_network_interface_ipv4(self, connection):
  678         connection.client.create_interface = mock.Mock(return_value=None)
  679         nas_server = mock.Mock()
  680         network = {'ip_address': '192.168.1.10',
  681                    'cidr': '192.168.1.10/24',
  682                    'gateway': '192.168.1.1',
  683                    'segmentation_id': '201'}
  684         port_id = mock.Mock()
  685         connection._create_network_interface(nas_server, network, port_id)
  686 
  687         expected = {'ip_addr': '192.168.1.10',
  688                     'netmask': '255.255.255.0',
  689                     'gateway': '192.168.1.1',
  690                     'port_id': port_id,
  691                     'vlan_id': '201'}
  692         connection.client.create_interface.assert_called_once_with(nas_server,
  693                                                                    **expected)
  694 
  695     @res_mock.mock_manila_input
  696     @res_mock.patch_connection
  697     def test_revert_to_snapshot(self, connection, mocked_input):
  698         context = mock.Mock()
  699         snapshot = mocked_input['snapshot']
  700         share_access_rules = [mocked_input['nfs_rw_access'], ]
  701         snapshot_access_rules = [mocked_input['nfs_rw_access'], ]
  702 
  703         connection.revert_to_snapshot(context, snapshot, share_access_rules,
  704                                       snapshot_access_rules)