300 lines
12 KiB
Python
300 lines
12 KiB
Python
# Copyright 2016 Mirantis, Inc.
|
|
# All Rights Reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
"""Unit tests for the Protocol helper module."""
|
|
|
|
import functools
|
|
from unittest import mock
|
|
|
|
import ddt
|
|
|
|
from manila.common import constants as const
|
|
from manila import exception
|
|
from manila.share.drivers.container import protocol_helper
|
|
from manila import test
|
|
from manila.tests.share.drivers.container.fakes import fake_share
|
|
|
|
|
|
@ddt.ddt
|
|
class DockerCIFSHelperTestCase(test.TestCase):
|
|
"""Tests ContainerShareDriver"""
|
|
|
|
def setUp(self):
|
|
super(DockerCIFSHelperTestCase, self).setUp()
|
|
self._helper = mock.Mock()
|
|
self.fake_conf = mock.Mock()
|
|
self.fake_conf.container_cifs_guest_ok = "yes"
|
|
self.DockerCIFSHelper = protocol_helper.DockerCIFSHelper(
|
|
self._helper, share=fake_share(), config=self.fake_conf)
|
|
|
|
def fake_exec_sync(self, *args, **kwargs):
|
|
kwargs["execute_arguments"].append(args)
|
|
try:
|
|
ret_val = kwargs["ret_val"]
|
|
except KeyError:
|
|
ret_val = None
|
|
return [ret_val]
|
|
|
|
def test_create_share_guest_ok(self):
|
|
expected_arguments = [
|
|
("fakeserver", ["net", "conf", "addshare", "fakeshareid",
|
|
"/shares/fakeshareid", "writeable=y", "guest_ok=y"]),
|
|
("fakeserver", ["net", "conf", "setparm", "fakeshareid",
|
|
"browseable", "yes"]),
|
|
("fakeserver", ["net", "conf", "setparm", "fakeshareid",
|
|
"hosts allow", "127.0.0.1"]),
|
|
("fakeserver", ["net", "conf", "setparm", "fakeshareid",
|
|
"read only", "no"]),
|
|
("fakeserver", ["net", "conf", "setparm", "fakeshareid",
|
|
"hosts deny", "0.0.0.0/0"]),
|
|
("fakeserver", ["net", "conf", "setparm", "fakeshareid",
|
|
"create mask", "0755"])]
|
|
actual_arguments = []
|
|
self._helper.execute = functools.partial(
|
|
self.fake_exec_sync, execute_arguments=actual_arguments,
|
|
ret_val=" fake 192.0.2.2/24 more fake \n" * 20)
|
|
self.DockerCIFSHelper.share = fake_share()
|
|
|
|
self.DockerCIFSHelper.create_share("fakeserver")
|
|
|
|
self.assertEqual(expected_arguments.sort(), actual_arguments.sort())
|
|
|
|
def test_create_share_guest_not_ok(self):
|
|
self.DockerCIFSHelper.conf = mock.Mock()
|
|
self.DockerCIFSHelper.conf.container_cifs_guest_ok = False
|
|
expected_arguments = [
|
|
("fakeserver", ["net", "conf", "addshare", "fakeshareid",
|
|
"/shares/fakeshareid", "writeable=y", "guest_ok=n"]),
|
|
("fakeserver", ["net", "conf", "setparm", "fakeshareid",
|
|
"browseable", "yes"]),
|
|
("fakeserver", ["net", "conf", "setparm", "fakeshareid",
|
|
"hosts allow", "192.0.2.2"]),
|
|
("fakeserver", ["net", "conf", "setparm", "fakeshareid",
|
|
"read only", "no"]),
|
|
("fakeserver", ["net", "conf", "setparm", "fakeshareid",
|
|
"hosts deny", "0.0.0.0/0"]),
|
|
("fakeserver", ["net", "conf", "setparm", "fakeshareid",
|
|
"create mask", "0755"])]
|
|
actual_arguments = []
|
|
self._helper.execute = functools.partial(
|
|
self.fake_exec_sync, execute_arguments=actual_arguments,
|
|
ret_val=" fake 192.0.2.2/24 more fake \n" * 20)
|
|
self.DockerCIFSHelper.share = fake_share()
|
|
|
|
self.DockerCIFSHelper.create_share("fakeserver")
|
|
|
|
self.assertEqual(expected_arguments.sort(), actual_arguments.sort())
|
|
|
|
def test_delete_share(self):
|
|
self.DockerCIFSHelper.share = fake_share()
|
|
|
|
self.DockerCIFSHelper.delete_share("fakeserver", "fakeshareid")
|
|
|
|
self.DockerCIFSHelper.container.execute.assert_called_with(
|
|
"fakeserver",
|
|
["net", "conf", "delshare", "fakeshareid"], ignore_errors=False)
|
|
|
|
def test__get_access_group_ro(self):
|
|
result = self.DockerCIFSHelper._get_access_group(const.ACCESS_LEVEL_RO)
|
|
|
|
self.assertEqual("read list", result)
|
|
|
|
def test__get_access_group_rw(self):
|
|
result = self.DockerCIFSHelper._get_access_group(const.ACCESS_LEVEL_RW)
|
|
|
|
self.assertEqual("valid users", result)
|
|
|
|
def test__get_access_group_other(self):
|
|
self.assertRaises(exception.InvalidShareAccessLevel,
|
|
self.DockerCIFSHelper._get_access_group,
|
|
"fake_level")
|
|
|
|
def test__get_existing_users(self):
|
|
self.DockerCIFSHelper.container.execute = mock.Mock(
|
|
return_value=("fake_user", ""))
|
|
|
|
result = self.DockerCIFSHelper._get_existing_users("fake_server_id",
|
|
"fake_share",
|
|
"fake_access")
|
|
|
|
self.assertEqual("fake_user", result)
|
|
self.DockerCIFSHelper.container.execute.assert_called_once_with(
|
|
"fake_server_id",
|
|
["net", "conf", "getparm", "fake_share", "fake_access"],
|
|
ignore_errors=True)
|
|
|
|
def test__set_users(self):
|
|
self.DockerCIFSHelper.container.execute = mock.Mock()
|
|
|
|
self.DockerCIFSHelper._set_users("fake_server_id", "fake_share",
|
|
"fake_access", "fake_user")
|
|
|
|
self.DockerCIFSHelper.container.execute.assert_called_once_with(
|
|
"fake_server_id",
|
|
["net", "conf", "setparm", "fake_share", "fake_access",
|
|
"fake_user"])
|
|
|
|
def test__allow_access_ok(self):
|
|
self.DockerCIFSHelper._get_access_group = mock.Mock(
|
|
return_value="valid users")
|
|
self.DockerCIFSHelper._get_existing_users = mock.Mock(
|
|
return_value="fake_user")
|
|
self.DockerCIFSHelper._set_users = mock.Mock()
|
|
|
|
self.DockerCIFSHelper._allow_access("fake_share", "fake_server_id",
|
|
"fake_user2", "rw")
|
|
|
|
self.DockerCIFSHelper._get_access_group.assert_called_once_with("rw")
|
|
self.DockerCIFSHelper._get_existing_users.assert_called_once_with(
|
|
"fake_server_id", "fake_share", "valid users")
|
|
self.DockerCIFSHelper._set_users.assert_called_once_with(
|
|
"fake_server_id", "fake_share", "valid users",
|
|
"fake_user fake_user2")
|
|
|
|
def test__allow_access_not_ok(self):
|
|
self.DockerCIFSHelper._get_access_group = mock.Mock(
|
|
return_value="valid users")
|
|
self.DockerCIFSHelper._get_existing_users = mock.Mock()
|
|
self.DockerCIFSHelper._get_existing_users.side_effect = TypeError
|
|
self.DockerCIFSHelper._set_users = mock.Mock()
|
|
|
|
self.DockerCIFSHelper._allow_access("fake_share", "fake_server_id",
|
|
"fake_user2", "rw")
|
|
|
|
self.DockerCIFSHelper._get_access_group.assert_called_once_with("rw")
|
|
self.DockerCIFSHelper._get_existing_users.assert_called_once_with(
|
|
"fake_server_id", "fake_share", "valid users")
|
|
self.DockerCIFSHelper._set_users.assert_called_once_with(
|
|
"fake_server_id", "fake_share", "valid users",
|
|
"fake_user2")
|
|
|
|
def test__deny_access_ok(self):
|
|
self.DockerCIFSHelper._get_access_group = mock.Mock(
|
|
return_value="valid users")
|
|
self.DockerCIFSHelper._get_existing_users = mock.Mock(
|
|
return_value="fake_user fake_user2")
|
|
self.DockerCIFSHelper._set_users = mock.Mock()
|
|
|
|
self.DockerCIFSHelper._deny_access("fake_share", "fake_server_id",
|
|
"fake_user2", "rw")
|
|
|
|
self.DockerCIFSHelper._get_access_group.assert_called_once_with("rw")
|
|
self.DockerCIFSHelper._get_existing_users.assert_called_once_with(
|
|
"fake_server_id", "fake_share", "valid users")
|
|
self.DockerCIFSHelper._set_users.assert_called_once_with(
|
|
"fake_server_id", "fake_share", "valid users",
|
|
"fake_user")
|
|
|
|
def test__deny_access_ok_so_many_users(self):
|
|
self.DockerCIFSHelper._get_access_group = mock.Mock(
|
|
return_value="valid users")
|
|
self.DockerCIFSHelper._get_existing_users = mock.Mock(
|
|
return_value="joost jaap huub dirk")
|
|
self.DockerCIFSHelper._set_users = mock.Mock()
|
|
|
|
# Sorry, Jaap.
|
|
self.DockerCIFSHelper._deny_access("fake_share", "fake_server_id",
|
|
"jaap", "rw")
|
|
|
|
self.DockerCIFSHelper._get_access_group.assert_called_once_with("rw")
|
|
self.DockerCIFSHelper._get_existing_users.assert_called_once_with(
|
|
"fake_server_id", "fake_share", "valid users")
|
|
self.DockerCIFSHelper._set_users.assert_called_once_with(
|
|
"fake_server_id", "fake_share", "valid users",
|
|
"dirk huub joost")
|
|
|
|
def test__deny_access_not_ok(self):
|
|
self.DockerCIFSHelper._get_access_group = mock.Mock(
|
|
return_value="valid users")
|
|
self.DockerCIFSHelper._get_existing_users = mock.Mock()
|
|
self.DockerCIFSHelper._get_existing_users.side_effect = TypeError
|
|
self.DockerCIFSHelper._set_users = mock.Mock()
|
|
self.mock_object(protocol_helper.LOG, "warning")
|
|
|
|
self.DockerCIFSHelper._deny_access("fake_share", "fake_server_id",
|
|
"fake_user2", "rw")
|
|
|
|
self.DockerCIFSHelper._get_access_group.assert_called_once_with("rw")
|
|
self.DockerCIFSHelper._get_existing_users.assert_called_once_with(
|
|
"fake_server_id", "fake_share", "valid users")
|
|
self.assertFalse(self.DockerCIFSHelper._set_users.called)
|
|
self.assertTrue(protocol_helper.LOG.warning.called)
|
|
|
|
def test_update_access_access_rules_wrong_type(self):
|
|
allow_rules = [{
|
|
"access_to": "192.0.2.2",
|
|
"access_level": "ro",
|
|
"access_type": "fake"
|
|
}]
|
|
self.mock_object(self.DockerCIFSHelper, "_allow_access")
|
|
|
|
self.assertRaises(exception.InvalidShareAccess,
|
|
self.DockerCIFSHelper.update_access,
|
|
"fakeserver", "fakeshareid", allow_rules, [], [])
|
|
|
|
def test_update_access_access_rules_ok(self):
|
|
access_rules = [{
|
|
"access_to": "fakeuser",
|
|
"access_level": "ro",
|
|
"access_type": "user"
|
|
}]
|
|
self.mock_object(self.DockerCIFSHelper, "_allow_access")
|
|
self.DockerCIFSHelper.container.execute = mock.Mock()
|
|
|
|
self.DockerCIFSHelper.update_access("fakeserver", "fakeshareid",
|
|
access_rules, [], [])
|
|
|
|
self.DockerCIFSHelper._allow_access.assert_called_once_with(
|
|
"fakeshareid",
|
|
"fakeserver",
|
|
"fakeuser",
|
|
"ro")
|
|
self.DockerCIFSHelper.container.execute.assert_called_once_with(
|
|
"fakeserver",
|
|
["net", "conf", "setparm", "fakeshareid", "valid users", ""])
|
|
|
|
def test_update_access_add_rules(self):
|
|
add_rules = [{
|
|
"access_to": "fakeuser",
|
|
"access_level": "ro",
|
|
"access_type": "user"
|
|
}]
|
|
self.mock_object(self.DockerCIFSHelper, "_allow_access")
|
|
|
|
self.DockerCIFSHelper.update_access("fakeserver", "fakeshareid", [],
|
|
add_rules, [])
|
|
|
|
self.DockerCIFSHelper._allow_access.assert_called_once_with(
|
|
"fakeshareid",
|
|
"fakeserver",
|
|
"fakeuser",
|
|
"ro")
|
|
|
|
def test_update_access_delete_rules(self):
|
|
delete_rules = [{
|
|
"access_to": "fakeuser",
|
|
"access_level": "ro",
|
|
"access_type": "user"
|
|
}]
|
|
self.mock_object(self.DockerCIFSHelper, "_deny_access")
|
|
|
|
self.DockerCIFSHelper.update_access("fakeserver", "fakeshareid",
|
|
[], [], delete_rules)
|
|
|
|
self.DockerCIFSHelper._deny_access.assert_called_once_with(
|
|
"fakeshareid",
|
|
"fakeserver",
|
|
"fakeuser",
|
|
"ro")
|