Coverage for manila/tests/share/drivers/ganesha/test_utils.py: 100%

1# Copyright (c) 2014 Red Hat, Inc. 

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 

16import os 

17from unittest import mock 

18 

19import ddt 

20 

21from manila import exception 

22from manila.share.drivers.ganesha import utils as ganesha_utils 

23from manila import test 

24from manila.tests import fake_share 

25 

26 

27patch_test_dict1 = {'a': 1, 'b': {'c': 2}, 'd': 3, 'e': 4} 

28patch_test_dict2 = {'a': 11, 'b': {'f': 5}, 'd': {'g': 6}} 

29patch_test_dict3 = {'b': {'c': 22, 'h': {'i': 7}}, 'e': None} 

30patch_test_dict_result = { 

31 'a': 11, 

32 'b': {'c': 22, 'f': 5, 'h': {'i': 7}}, 

33 'd': {'g': 6}, 

34 'e': None, 

35} 

36 

37walk_test_dict = {'a': {'b': {'c': {'d': {'e': 'f'}}}}} 

38walk_test_list = [('e', 'f')] 

39 

40 

41def fake_access(kwargs): 

42 fake_access_rule = fake_share.fake_access(**kwargs) 

43 fake_access_rule.to_dict = lambda: fake_access_rule.values 

44 return fake_access_rule 

45 

46 

47@ddt.ddt 

48class GaneshaUtilsTests(test.TestCase): 

49 """Tests Ganesha utility functions.""" 

50 

51 def test_patch(self): 

52 ret = ganesha_utils.patch(patch_test_dict1, patch_test_dict2, 

53 patch_test_dict3) 

54 self.assertEqual(patch_test_dict_result, ret) 

55 

56 def test_walk(self): 

57 ret = [elem for elem in ganesha_utils.walk(walk_test_dict)] 

58 self.assertEqual(walk_test_list, ret) 

59 

60 def test_path_from(self): 

61 self.mock_object(os.path, 'abspath', 

62 lambda path: os.path.join('/foo/bar', path)) 

63 ret = ganesha_utils.path_from('baz.py', '../quux', 'tic/tac/toe') 

64 self.assertEqual('/foo/quux/tic/tac/toe', os.path.normpath(ret)) 

65 

66 @ddt.data({'rule': {'access_type': 'ip', 

67 'access_level': 'ro', 

68 'access_to': '10.10.10.12'}, 

69 'kwargs': {'abort': True}}, 

70 {'rule': {'access_type': 'cert', 

71 'access_level': 'ro', 

72 'access_to': 'some-CN'}, 

73 'kwargs': {'abort': False}}, 

74 {'rule': {'access_type': 'ip', 

75 'access_level': 'rw', 

76 'access_to': '10.10.10.12'}, 

77 'kwargs': {}}) 

78 @ddt.unpack 

79 def test_get_valid_access_rules(self, rule, kwargs): 

80 supported = ['ip', 'ro'] 

81 

82 ret = ganesha_utils.validate_access_rule( 

83 *([[a] for a in supported] + [fake_access(rule)]), **kwargs) 

84 

85 self.assertEqual( 

86 [rule['access_' + k] for k in ['type', 'level']] == supported, ret) 

87 

88 @ddt.data({'rule': {'access_type': 'cert', 

89 'access_level': 'ro', 

90 'access_to': 'some-CN'}, 

91 'trouble': exception.InvalidShareAccess}, 

92 {'rule': {'access_type': 'ip', 

93 'access_level': 'rw', 

94 'access_to': '10.10.10.12'}, 

95 'trouble': exception.InvalidShareAccessLevel}) 

96 @ddt.unpack 

97 def test_get_valid_access_rules_fail(self, rule, trouble): 

98 self.assertRaises(trouble, ganesha_utils.validate_access_rule, 

99 ['ip'], ['ro'], fake_access(rule), abort=True) 

100 

101 @ddt.data({'rule': {'access_type': 'ip', 

102 'access_level': 'rw', 

103 'access_to': '10.10.10.12'}, 

104 'result': {'access_type': 'ip', 

105 'access_level': 'rw', 

106 'access_to': '10.10.10.12'}, 

107 }, 

108 {'rule': {'access_type': 'ip', 

109 'access_level': 'rw', 

110 'access_to': '0.0.0.0/0'}, 

111 'result': {'access_type': 'ip', 

112 'access_level': 'rw', 

113 'access_to': '0.0.0.0'}, 

114 }, 

115 ) 

116 @ddt.unpack 

117 def test_fixup_access_rules(self, rule, result): 

118 

119 self.assertEqual(result, ganesha_utils.fixup_access_rule(rule)) 

120 

121 

122@ddt.ddt 

123class SSHExecutorTestCase(test.TestCase): 

124 """Tests SSHExecutor.""" 

125 

126 @ddt.data({'run_as_root': True, 'expected_prefix': 'sudo '}, 

127 {'run_as_root': False, 'expected_prefix': ''}) 

128 @ddt.unpack 

129 def test_call_ssh_exec_object_with_run_as_root( 

130 self, run_as_root, expected_prefix): 

131 with mock.patch.object(ganesha_utils.ssh_utils, 'SSHPool'): 

132 self.execute = ganesha_utils.SSHExecutor() 

133 fake_ssh_object = mock.Mock() 

134 self.mock_object(self.execute.pool, 'get', 

135 mock.Mock(return_value=fake_ssh_object)) 

136 self.mock_object(ganesha_utils.processutils, 'ssh_execute', 

137 mock.Mock(return_value=('', ''))) 

138 ret = self.execute('ls', run_as_root=run_as_root) 

139 self.assertEqual(('', ''), ret) 

140 self.execute.pool.get.assert_called_once_with() 

141 ganesha_utils.processutils.ssh_execute.assert_called_once_with( 

142 fake_ssh_object, expected_prefix + 'ls')