Coverage for manila/tests/share/drivers/windows/test_windows_smb_helper.py: 100%

1# Copyright (c) 2015 Cloudbase Solutions SRL 

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.common import constants 

22from manila import exception 

23from manila.share import configuration 

24from manila.share.drivers.windows import windows_smb_helper 

25from manila.share.drivers.windows import windows_utils 

26from manila import test 

27 

28from oslo_config import cfg 

29 

30CONF = cfg.CONF 

31CONF.import_opt('share_mount_path', 

32 'manila.share.drivers.generic') 

33 

34 

35@ddt.ddt 

36class WindowsSMBHelperTestCase(test.TestCase): 

37 _FAKE_SERVER = {'public_address': mock.sentinel.public_address} 

38 _FAKE_SHARE_NAME = "fake_share_name" 

39 _FAKE_SHARE = "\\\\%s\\%s" % (_FAKE_SERVER['public_address'], 

40 _FAKE_SHARE_NAME) 

41 _FAKE_SHARE_LOCATION = os.path.join( 

42 configuration.Configuration(None).share_mount_path, 

43 _FAKE_SHARE_NAME) 

44 _FAKE_ACCOUNT_NAME = 'FakeDomain\\FakeUser' 

45 _FAKE_RW_ACC_RULE = { 

46 'access_to': _FAKE_ACCOUNT_NAME, 

47 'access_level': constants.ACCESS_LEVEL_RW, 

48 'access_type': 'user', 

49 } 

50 

51 def setUp(self): 

52 self._remote_exec = mock.Mock() 

53 fake_conf = configuration.Configuration(None) 

54 

55 self._win_smb_helper = windows_smb_helper.WindowsSMBHelper( 

56 self._remote_exec, fake_conf) 

57 

58 super(WindowsSMBHelperTestCase, self).setUp() 

59 

60 def test_init_helper(self): 

61 self._win_smb_helper.init_helper(mock.sentinel.server) 

62 self._remote_exec.assert_called_once_with(mock.sentinel.server, 

63 "Get-SmbShare") 

64 

65 @ddt.data(True, False) 

66 @mock.patch.object(windows_smb_helper.WindowsSMBHelper, '_share_exists') 

67 def test_create_exports(self, share_exists, mock_share_exists): 

68 mock_share_exists.return_value = share_exists 

69 

70 result = self._win_smb_helper.create_exports(self._FAKE_SERVER, 

71 self._FAKE_SHARE_NAME) 

72 

73 if not share_exists: 

74 cmd = ['New-SmbShare', '-Name', self._FAKE_SHARE_NAME, '-Path', 

75 self._win_smb_helper._windows_utils.normalize_path( 

76 self._FAKE_SHARE_LOCATION), 

77 '-ReadAccess', "*%s" % self._win_smb_helper._NULL_SID] 

78 self._remote_exec.assert_called_once_with(self._FAKE_SERVER, cmd) 

79 else: 

80 self.assertFalse(self._remote_exec.called) 

81 

82 expected_exports = [ 

83 { 

84 'is_admin_only': False, 

85 'metadata': {'export_location_metadata_example': 'example'}, 

86 'path': self._FAKE_SHARE 

87 }, 

88 ] 

89 

90 self.assertEqual(expected_exports, result) 

91 

92 @mock.patch.object(windows_smb_helper.WindowsSMBHelper, '_share_exists') 

93 def test_remove_exports(self, mock_share_exists): 

94 mock_share_exists.return_value = True 

95 

96 self._win_smb_helper.remove_exports(mock.sentinel.server, 

97 mock.sentinel.share_name) 

98 

99 cmd = ['Remove-SmbShare', '-Name', mock.sentinel.share_name, "-Force"] 

100 self._remote_exec.assert_called_once_with(mock.sentinel.server, cmd) 

101 

102 @mock.patch.object(windows_utils.WindowsUtils, 

103 'get_volume_path_by_mount_path') 

104 @mock.patch.object(windows_smb_helper.WindowsSMBHelper, 

105 '_get_share_path_by_name') 

106 def test_get_volume_path_by_share_name(self, mock_get_share_path, 

107 mock_get_vol_path): 

108 mock_get_share_path.return_value = self._FAKE_SHARE_LOCATION 

109 

110 volume_path = self._win_smb_helper._get_volume_path_by_share_name( 

111 mock.sentinel.server, self._FAKE_SHARE_NAME) 

112 

113 mock_get_share_path.assert_called_once_with(mock.sentinel.server, 

114 self._FAKE_SHARE_NAME) 

115 mock_get_vol_path.assert_called_once_with(mock.sentinel.server, 

116 self._FAKE_SHARE_LOCATION) 

117 

118 self.assertEqual(mock_get_vol_path.return_value, volume_path) 

119 

120 @ddt.data({'raw_out': '', 'expected': []}, 

121 {'raw_out': '{"key": "val"}', 

122 'expected': [{"key": "val"}]}, 

123 {'raw_out': '[{"key": "val"}, {"key2": "val2"}]', 

124 'expected': [{"key": "val"}, {"key2": "val2"}]}) 

125 @ddt.unpack 

126 def test_get_acls_helper(self, raw_out, expected): 

127 self._remote_exec.return_value = (raw_out, mock.sentinel.err) 

128 

129 rules = self._win_smb_helper._get_acls(mock.sentinel.server, 

130 self._FAKE_SHARE_NAME) 

131 

132 self.assertEqual(expected, rules) 

133 expected_cmd = ( 

134 'Get-SmbShareAccess -Name %s | ' 

135 'Select-Object @("Name", "AccountName", ' 

136 '"AccessControlType", "AccessRight") | ' 

137 'ConvertTo-JSON -Compress') % self._FAKE_SHARE_NAME 

138 self._remote_exec.assert_called_once_with(mock.sentinel.server, 

139 expected_cmd) 

140 

141 @mock.patch.object(windows_smb_helper.WindowsSMBHelper, 

142 '_get_acls') 

143 def test_get_access_rules(self, mock_get_acls): 

144 helper = self._win_smb_helper 

145 valid_acl = { 

146 'AccountName': self._FAKE_ACCOUNT_NAME, 

147 'AccessRight': helper._WIN_ACCESS_RIGHT_FULL, 

148 'AccessControlType': helper._WIN_ACL_ALLOW, 

149 } 

150 

151 valid_acls = [valid_acl, 

152 dict(valid_acl, 

153 AccessRight=helper._WIN_ACCESS_RIGHT_CHANGE), 

154 dict(valid_acl, 

155 AccessRight=helper._WIN_ACCESS_RIGHT_READ)] 

156 # Those are rules that were not added by us and are expected to 

157 # be ignored. When encountering such a rule, a warning message 

158 # will be logged. 

159 ignored_acls = [ 

160 dict(valid_acl, AccessRight=helper._WIN_ACCESS_RIGHT_CUSTOM), 

161 dict(valid_acl, AccessControlType=helper._WIN_ACL_DENY)] 

162 

163 mock_get_acls.return_value = valid_acls + ignored_acls 

164 # There won't be multiple access rules for the same account, 

165 # but we'll ignore this fact for the sake of this test. 

166 expected_rules = [self._FAKE_RW_ACC_RULE, self._FAKE_RW_ACC_RULE, 

167 dict(self._FAKE_RW_ACC_RULE, 

168 access_level=constants.ACCESS_LEVEL_RO)] 

169 

170 rules = helper.get_access_rules(mock.sentinel.server, 

171 mock.sentinel.share_name) 

172 self.assertEqual(expected_rules, rules) 

173 

174 mock_get_acls.assert_called_once_with(mock.sentinel.server, 

175 mock.sentinel.share_name) 

176 

177 @mock.patch.object(windows_smb_helper.WindowsSMBHelper, '_refresh_acl') 

178 def test_grant_share_access(self, mock_refresh_acl): 

179 self._win_smb_helper._grant_share_access(mock.sentinel.server, 

180 mock.sentinel.share_name, 

181 constants.ACCESS_LEVEL_RW, 

182 mock.sentinel.username) 

183 

184 cmd = ["Grant-SmbShareAccess", "-Name", mock.sentinel.share_name, 

185 "-AccessRight", "Change", 

186 "-AccountName", "'%s'" % mock.sentinel.username, "-Force"] 

187 self._remote_exec.assert_called_once_with(mock.sentinel.server, cmd) 

188 mock_refresh_acl.assert_called_once_with(mock.sentinel.server, 

189 mock.sentinel.share_name) 

190 

191 def test_refresh_acl(self): 

192 self._win_smb_helper._refresh_acl(mock.sentinel.server, 

193 mock.sentinel.share_name) 

194 

195 cmd = ['Set-SmbPathAcl', '-ShareName', mock.sentinel.share_name] 

196 self._remote_exec.assert_called_once_with(mock.sentinel.server, cmd) 

197 

198 @mock.patch.object(windows_smb_helper.WindowsSMBHelper, '_refresh_acl') 

199 def test_revoke_share_access(self, mock_refresh_acl): 

200 self._win_smb_helper._revoke_share_access(mock.sentinel.server, 

201 mock.sentinel.share_name, 

202 mock.sentinel.username) 

203 

204 cmd = ["Revoke-SmbShareAccess", "-Name", mock.sentinel.share_name, 

205 "-AccountName", '"%s"' % mock.sentinel.username, "-Force"] 

206 self._remote_exec.assert_called_once_with(mock.sentinel.server, cmd) 

207 mock_refresh_acl.assert_called_once_with(mock.sentinel.server, 

208 mock.sentinel.share_name) 

209 

210 def test_update_access_invalid_type(self): 

211 invalid_access_rule = dict(self._FAKE_RW_ACC_RULE, 

212 access_type='ip') 

213 self.assertRaises( 

214 exception.InvalidShareAccess, 

215 self._win_smb_helper.update_access, 

216 mock.sentinel.server, mock.sentinel.share_name, 

217 [invalid_access_rule], [], [], []) 

218 

219 def test_update_access_invalid_level(self): 

220 invalid_access_rule = dict(self._FAKE_RW_ACC_RULE, 

221 access_level='fake_level') 

222 self.assertRaises( 

223 exception.InvalidShareAccessLevel, 

224 self._win_smb_helper.update_access, 

225 mock.sentinel.server, mock.sentinel.share_name, 

226 [], [invalid_access_rule], [], []) 

227 

228 @mock.patch.object(windows_smb_helper.WindowsSMBHelper, 

229 '_revoke_share_access') 

230 def test_update_access_deleting_invalid_rule(self, mock_revoke): 

231 # We want to make sure that we allow deleting invalid rules. 

232 invalid_access_rule = dict(self._FAKE_RW_ACC_RULE, 

233 access_level='fake_level') 

234 delete_rules = [invalid_access_rule, self._FAKE_RW_ACC_RULE] 

235 

236 self._win_smb_helper.update_access( 

237 mock.sentinel.server, mock.sentinel.share_name, 

238 [], [], delete_rules, []) 

239 

240 mock_revoke.assert_called_once_with( 

241 mock.sentinel.server, mock.sentinel.share_name, 

242 self._FAKE_RW_ACC_RULE['access_to']) 

243 

244 @mock.patch.object(windows_smb_helper.WindowsSMBHelper, 

245 'validate_access_rules') 

246 @mock.patch.object(windows_smb_helper.WindowsSMBHelper, 

247 'get_access_rules') 

248 @mock.patch.object(windows_smb_helper.WindowsSMBHelper, 

249 '_grant_share_access') 

250 @mock.patch.object(windows_smb_helper.WindowsSMBHelper, 

251 '_revoke_share_access') 

252 def test_update_access(self, mock_revoke, mock_grant, 

253 mock_get_access_rules, mock_validate): 

254 added_rules = [mock.MagicMock(), mock.MagicMock()] 

255 deleted_rules = [mock.MagicMock(), mock.MagicMock()] 

256 

257 self._win_smb_helper.update_access( 

258 mock.sentinel.server, mock.sentinel.share_name, 

259 [], added_rules, deleted_rules, []) 

260 

261 mock_revoke.assert_has_calls( 

262 [mock.call(mock.sentinel.server, mock.sentinel.share_name, 

263 deleted_rule['access_to']) 

264 for deleted_rule in deleted_rules]) 

265 

266 mock_grant.assert_has_calls( 

267 [mock.call(mock.sentinel.server, mock.sentinel.share_name, 

268 added_rule['access_level'], added_rule['access_to']) 

269 for added_rule in added_rules]) 

270 

271 @mock.patch.object(windows_smb_helper.WindowsSMBHelper, 

272 '_get_rule_updates') 

273 @mock.patch.object(windows_smb_helper.WindowsSMBHelper, 

274 'validate_access_rules') 

275 @mock.patch.object(windows_smb_helper.WindowsSMBHelper, 

276 'get_access_rules') 

277 @mock.patch.object(windows_smb_helper.WindowsSMBHelper, 

278 '_grant_share_access') 

279 @mock.patch.object(windows_smb_helper.WindowsSMBHelper, 

280 '_revoke_share_access') 

281 def test_update_access_maintenance( 

282 self, mock_revoke, mock_grant, 

283 mock_get_access_rules, mock_validate, 

284 mock_get_rule_updates): 

285 all_rules = mock.MagicMock() 

286 added_rules = [mock.MagicMock(), mock.MagicMock()] 

287 deleted_rules = [mock.MagicMock(), mock.MagicMock()] 

288 

289 mock_get_rule_updates.return_value = [ 

290 added_rules, deleted_rules] 

291 

292 self._win_smb_helper.update_access( 

293 mock.sentinel.server, mock.sentinel.share_name, 

294 all_rules, [], [], []) 

295 

296 mock_get_access_rules.assert_called_once_with( 

297 mock.sentinel.server, mock.sentinel.share_name) 

298 mock_get_rule_updates.assert_called_once_with( 

299 existing_rules=mock_get_access_rules.return_value, 

300 requested_rules=all_rules) 

301 mock_revoke.assert_has_calls( 

302 [mock.call(mock.sentinel.server, mock.sentinel.share_name, 

303 deleted_rule['access_to']) 

304 for deleted_rule in deleted_rules]) 

305 

306 mock_grant.assert_has_calls( 

307 [mock.call(mock.sentinel.server, mock.sentinel.share_name, 

308 added_rule['access_level'], added_rule['access_to']) 

309 for added_rule in added_rules]) 

310 

311 def test_get_rule_updates(self): 

312 req_rule_0 = self._FAKE_RW_ACC_RULE 

313 req_rule_1 = dict(self._FAKE_RW_ACC_RULE, 

314 access_to='fake_acc') 

315 

316 curr_rule_0 = dict(self._FAKE_RW_ACC_RULE, 

317 access_to=self._FAKE_RW_ACC_RULE[ 

318 'access_to'].upper()) 

319 curr_rule_1 = dict(self._FAKE_RW_ACC_RULE, 

320 access_to='fake_acc2') 

321 curr_rule_2 = dict(req_rule_1, 

322 access_level=constants.ACCESS_LEVEL_RO) 

323 

324 expected_added_rules = [req_rule_1] 

325 expected_deleted_rules = [curr_rule_1, curr_rule_2] 

326 

327 existing_rules = [curr_rule_0, curr_rule_1, curr_rule_2] 

328 requested_rules = [req_rule_0, req_rule_1] 

329 

330 (added_rules, 

331 deleted_rules) = self._win_smb_helper._get_rule_updates( 

332 existing_rules, requested_rules) 

333 

334 self.assertEqual(expected_added_rules, added_rules) 

335 self.assertEqual(expected_deleted_rules, deleted_rules) 

336 

337 def test_get_share_name(self): 

338 result = self._win_smb_helper._get_share_name(self._FAKE_SHARE) 

339 self.assertEqual(self._FAKE_SHARE_NAME, result) 

340 

341 def test_get_share_path_by_name(self): 

342 self._remote_exec.return_value = (self._FAKE_SHARE_LOCATION, 

343 mock.sentinel.std_err) 

344 

345 result = self._win_smb_helper._get_share_path_by_name( 

346 mock.sentinel.server, 

347 mock.sentinel.share_name) 

348 

349 cmd = ('Get-SmbShare -Name %s | ' 

350 'Select-Object -ExpandProperty Path' % mock.sentinel.share_name) 

351 self._remote_exec.assert_called_once_with(mock.sentinel.server, 

352 cmd, 

353 check_exit_code=True) 

354 self.assertEqual(self._FAKE_SHARE_LOCATION, result) 

355 

356 @mock.patch.object(windows_smb_helper.WindowsSMBHelper, 

357 '_get_share_path_by_name') 

358 def test_get_share_path_by_export_location(self, 

359 mock_get_share_path_by_name): 

360 mock_get_share_path_by_name.return_value = mock.sentinel.share_path 

361 

362 result = self._win_smb_helper.get_share_path_by_export_location( 

363 mock.sentinel.server, self._FAKE_SHARE) 

364 

365 mock_get_share_path_by_name.assert_called_once_with( 

366 mock.sentinel.server, self._FAKE_SHARE_NAME) 

367 self.assertEqual(mock.sentinel.share_path, result) 

368 

369 @mock.patch.object(windows_smb_helper.WindowsSMBHelper, 

370 '_get_share_path_by_name') 

371 def test_share_exists(self, mock_get_share_path_by_name): 

372 result = self._win_smb_helper._share_exists(mock.sentinel.server, 

373 mock.sentinel.share_name) 

374 

375 mock_get_share_path_by_name.assert_called_once_with( 

376 mock.sentinel.server, 

377 mock.sentinel.share_name, 

378 ignore_missing=True) 

379 self.assertTrue(result)