Coverage for manila/tests/share/drivers/netapp/dataontap/protocols/test_cifs_cmode.py: 100%

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""" 

15Mock unit tests for the NetApp driver protocols CIFS class module. 

16""" 

17 

18import copy 

19from unittest import mock 

20 

21import ddt 

22 

23from manila.common import constants 

24from manila import exception 

25from manila.share.drivers.netapp.dataontap.protocols import cifs_cmode 

26from manila import test 

27from manila.tests.share.drivers.netapp.dataontap.protocols \ 

28 import fakes as fake 

29 

30 

31@ddt.ddt 

32class 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 @ddt.data({'replica': True, 'cifs_exist': False}, 

44 {'replica': False, 'cifs_exist': True}) 

45 @ddt.unpack 

46 def test_create_share(self, replica, cifs_exist): 

47 

48 self.mock_client.cifs_share_exists.return_value = cifs_exist 

49 self.mock_client.get_volume_junction_path.return_value = ( 

50 fake.CIFS_SHARE_PATH) 

51 

52 result = self.helper.create_share( 

53 fake.CIFS_SHARE, fake.SHARE_NAME, 

54 replica=replica) 

55 

56 export_addresses = [fake.SHARE_ADDRESS_1, fake.SHARE_ADDRESS_2] 

57 export_paths = [result(address) for address in export_addresses] 

58 expected_paths = [ 

59 r'\\%s%s' % (fake.SHARE_ADDRESS_1, fake.CIFS_SHARE_PATH_PARSED), 

60 r'\\%s%s' % (fake.SHARE_ADDRESS_2, fake.CIFS_SHARE_PATH_PARSED), 

61 ] 

62 self.assertEqual(expected_paths, export_paths) 

63 

64 self.mock_client.cifs_share_exists.assert_called_once_with( 

65 fake.SHARE_NAME) 

66 if cifs_exist: 

67 self.mock_client.create_cifs_share.assert_not_called() 

68 self.mock_client.remove_cifs_share.assert_not_called() 

69 else: 

70 self.mock_client.create_cifs_share.assert_called_once_with( 

71 fake.SHARE_NAME, fake.CIFS_SHARE_PATH) 

72 self.mock_client.remove_cifs_share_access.assert_called_once_with( 

73 fake.SHARE_NAME, 'Everyone') 

74 

75 if replica: 

76 self.mock_client.set_volume_security_style.assert_not_called() 

77 else: 

78 self.mock_client.set_volume_security_style.assert_called_once_with( 

79 fake.SHARE_NAME, security_style='ntfs') 

80 

81 def test_create_share_ensure_not_exist_error(self): 

82 

83 self.mock_client.cifs_share_exists.return_value = False 

84 

85 self.assertRaises(exception.NetAppException, 

86 self.helper.create_share, 

87 fake.CIFS_SHARE, fake.SHARE_NAME, 

88 ensure_share_already_exists=True) 

89 

90 def test_delete_share(self): 

91 

92 self.helper.delete_share(fake.CIFS_SHARE, fake.SHARE_NAME) 

93 

94 self.mock_client.remove_cifs_share.assert_called_once_with( 

95 fake.SHARE_NAME) 

96 

97 def test_update_access(self): 

98 

99 mock_validate_access_rule = self.mock_object(self.helper, 

100 '_validate_access_rule') 

101 mock_get_access_rules = self.mock_object( 

102 self.helper, '_get_access_rules', 

103 mock.Mock(return_value=fake.EXISTING_CIFS_RULES)) 

104 mock_handle_added_rules = self.mock_object(self.helper, 

105 '_handle_added_rules') 

106 mock_handle_ro_to_rw_rules = self.mock_object(self.helper, 

107 '_handle_ro_to_rw_rules') 

108 mock_handle_rw_to_ro_rules = self.mock_object(self.helper, 

109 '_handle_rw_to_ro_rules') 

110 mock_handle_deleted_rules = self.mock_object(self.helper, 

111 '_handle_deleted_rules') 

112 

113 self.helper.update_access(fake.CIFS_SHARE, 

114 fake.SHARE_NAME, 

115 [fake.USER_ACCESS]) 

116 

117 new_rules = {'fake_user': constants.ACCESS_LEVEL_RW} 

118 mock_validate_access_rule.assert_called_once_with(fake.USER_ACCESS) 

119 mock_get_access_rules.assert_called_once_with(fake.CIFS_SHARE, 

120 fake.SHARE_NAME) 

121 mock_handle_added_rules.assert_called_once_with( 

122 fake.SHARE_NAME, fake.EXISTING_CIFS_RULES, new_rules) 

123 mock_handle_ro_to_rw_rules.assert_called_once_with( 

124 fake.SHARE_NAME, fake.EXISTING_CIFS_RULES, new_rules) 

125 mock_handle_rw_to_ro_rules.assert_called_once_with( 

126 fake.SHARE_NAME, fake.EXISTING_CIFS_RULES, new_rules) 

127 mock_handle_deleted_rules.assert_called_once_with( 

128 fake.SHARE_NAME, fake.EXISTING_CIFS_RULES, new_rules) 

129 

130 def test_validate_access_rule(self): 

131 

132 result = self.helper._validate_access_rule(fake.USER_ACCESS) 

133 

134 self.assertIsNone(result) 

135 

136 def test_validate_access_rule_invalid_type(self): 

137 

138 rule = copy.copy(fake.USER_ACCESS) 

139 rule['access_type'] = 'ip' 

140 

141 self.assertRaises(exception.InvalidShareAccess, 

142 self.helper._validate_access_rule, 

143 rule) 

144 

145 def test_validate_access_rule_invalid_level(self): 

146 

147 rule = copy.copy(fake.USER_ACCESS) 

148 rule['access_level'] = 'none' 

149 

150 self.assertRaises(exception.InvalidShareAccessLevel, 

151 self.helper._validate_access_rule, 

152 rule) 

153 

154 def test_handle_added_rules(self): 

155 

156 self.helper._handle_added_rules(fake.SHARE_NAME, 

157 fake.EXISTING_CIFS_RULES, 

158 fake.NEW_CIFS_RULES) 

159 

160 self.mock_client.add_cifs_share_access.assert_has_calls([ 

161 mock.call(fake.SHARE_NAME, 'user5', False), 

162 mock.call(fake.SHARE_NAME, 'user6', True), 

163 ], any_order=True) 

164 

165 def test_handle_ro_to_rw_rules(self): 

166 

167 self.helper._handle_ro_to_rw_rules(fake.SHARE_NAME, 

168 fake.EXISTING_CIFS_RULES, 

169 fake.NEW_CIFS_RULES) 

170 

171 self.mock_client.modify_cifs_share_access.assert_has_calls([ 

172 mock.call(fake.SHARE_NAME, 'user2', False) 

173 ]) 

174 

175 def test_handle_rw_to_ro_rules(self): 

176 

177 self.helper._handle_rw_to_ro_rules(fake.SHARE_NAME, 

178 fake.EXISTING_CIFS_RULES, 

179 fake.NEW_CIFS_RULES) 

180 

181 self.mock_client.modify_cifs_share_access.assert_has_calls([ 

182 mock.call(fake.SHARE_NAME, 'user3', True) 

183 ]) 

184 

185 def test_handle_deleted_rules(self): 

186 

187 self.helper._handle_deleted_rules(fake.SHARE_NAME, 

188 fake.EXISTING_CIFS_RULES, 

189 fake.NEW_CIFS_RULES) 

190 

191 self.mock_client.remove_cifs_share_access.assert_has_calls([ 

192 mock.call(fake.SHARE_NAME, 'user4') 

193 ]) 

194 

195 def test_get_access_rules(self): 

196 

197 self.mock_client.get_cifs_share_access = ( 

198 mock.Mock(return_value='fake_rules')) 

199 

200 result = self.helper._get_access_rules(fake.CIFS_SHARE, 

201 fake.SHARE_NAME) 

202 

203 self.assertEqual('fake_rules', result) 

204 self.mock_client.get_cifs_share_access.assert_called_once_with( 

205 fake.SHARE_NAME) 

206 

207 def test_get_target(self): 

208 

209 target = self.helper.get_target(fake.CIFS_SHARE) 

210 self.assertEqual(fake.SHARE_ADDRESS_1, target) 

211 

212 def test_get_target_missing_location(self): 

213 

214 target = self.helper.get_target({'export_location': ''}) 

215 self.assertEqual('', target) 

216 

217 def test_get_share_name_for_share(self): 

218 

219 self.mock_client.get_volume_at_junction_path.return_value = ( 

220 fake.VOLUME) 

221 

222 share_name = self.helper.get_share_name_for_share(fake.CIFS_SHARE) 

223 

224 self.assertEqual(fake.SHARE_NAME, share_name) 

225 self.mock_client.get_volume_at_junction_path.assert_called_once_with( 

226 fake.CIFS_SHARE_PATH) 

227 

228 def test_get_share_name_for_share_not_found(self): 

229 

230 self.mock_client.get_volume_at_junction_path.return_value = None 

231 

232 share_name = self.helper.get_share_name_for_share(fake.CIFS_SHARE) 

233 

234 self.assertIsNone(share_name) 

235 self.mock_client.get_volume_at_junction_path.assert_called_once_with( 

236 fake.CIFS_SHARE_PATH) 

237 

238 @ddt.data( 

239 { 

240 'location': r'\\%s\%s' % (fake.SHARE_ADDRESS_1, fake.SHARE_NAME), 

241 'ip': fake.SHARE_ADDRESS_1, 

242 'share_name': fake.SHARE_NAME, 

243 }, { 

244 'location': r'//%s/%s' % (fake.SHARE_ADDRESS_1, fake.SHARE_NAME), 

245 'ip': fake.SHARE_ADDRESS_1, 

246 'share_name': fake.SHARE_NAME, 

247 }, 

248 {'location': '', 'ip': '', 'share_name': ''}, 

249 {'location': 'invalid', 'ip': '', 'share_name': ''}, 

250 ) 

251 @ddt.unpack 

252 def test_get_export_location(self, location, ip, share_name): 

253 

254 share = fake.CIFS_SHARE.copy() 

255 share['export_location'] = location 

256 self.mock_object(self.helper, '_get_share_export_location', 

257 mock.Mock(return_value=location)) 

258 

259 result_ip, result_share_name = self.helper._get_export_location(share) 

260 

261 self.assertEqual(ip, result_ip) 

262 self.assertEqual(share_name, result_share_name) 

263 self.helper._get_share_export_location.assert_called_once_with(share) 

264 

265 def test_cleanup_demoted_replica(self): 

266 self.helper.cleanup_demoted_replica(fake.CIFS_SHARE, fake.SHARE_NAME) 

267 

268 self.mock_client.remove_cifs_share.assert_called_once_with( 

269 fake.SHARE_NAME)