Coverage for manila/tests/share/drivers/container/test_protocol_helper.py: 98%

1# Copyright 2016 Mirantis, 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"""Unit tests for the Protocol helper module.""" 

16 

17import functools 

18from unittest import mock 

19 

20import ddt 

21 

22from manila.common import constants as const 

23from manila import exception 

24from manila.share.drivers.container import protocol_helper 

25from manila import test 

26from manila.tests.share.drivers.container.fakes import fake_share 

27 

28 

29@ddt.ddt 

30class DockerCIFSHelperTestCase(test.TestCase): 

31 """Tests ContainerShareDriver""" 

32 

33 def setUp(self): 

34 super(DockerCIFSHelperTestCase, self).setUp() 

35 self._helper = mock.Mock() 

36 self.fake_conf = mock.Mock() 

37 self.fake_conf.container_cifs_guest_ok = "yes" 

38 self.DockerCIFSHelper = protocol_helper.DockerCIFSHelper( 

39 self._helper, share=fake_share(), config=self.fake_conf) 

40 

41 def fake_exec_sync(self, *args, **kwargs): 

42 kwargs["execute_arguments"].append(args) 

43 try: 

44 ret_val = kwargs["ret_val"] 

45 except KeyError: 

46 ret_val = None 

47 return [ret_val] 

48 

49 def test_create_share_guest_ok(self): 

50 expected_arguments = [ 

51 ("fakeserver", ["net", "conf", "addshare", "fakeshareid", 

52 "/shares/fakeshareid", "writeable=y", "guest_ok=y"]), 

53 ("fakeserver", ["net", "conf", "setparm", "fakeshareid", 

54 "browseable", "yes"]), 

55 ("fakeserver", ["net", "conf", "setparm", "fakeshareid", 

56 "hosts allow", "127.0.0.1"]), 

57 ("fakeserver", ["net", "conf", "setparm", "fakeshareid", 

58 "read only", "no"]), 

59 ("fakeserver", ["net", "conf", "setparm", "fakeshareid", 

60 "hosts deny", "0.0.0.0/0"]), 

61 ("fakeserver", ["net", "conf", "setparm", "fakeshareid", 

62 "create mask", "0755"])] 

63 actual_arguments = [] 

64 self._helper.execute = functools.partial( 

65 self.fake_exec_sync, execute_arguments=actual_arguments, 

66 ret_val=" fake 192.0.2.2/24 more fake \n" * 20) 

67 self.DockerCIFSHelper.share = fake_share() 

68 self.mock_object(self.DockerCIFSHelper.container, 

69 'fetch_container_addresses', 

70 mock.Mock(return_value=['192.0.2.2'])) 

71 

72 self.DockerCIFSHelper.create_share("fakeserver") 

73 

74 self.assertEqual(expected_arguments.sort(), actual_arguments.sort()) 

75 

76 def test_create_share_guest_not_ok(self): 

77 self.DockerCIFSHelper.conf = mock.Mock() 

78 self.DockerCIFSHelper.conf.container_cifs_guest_ok = False 

79 expected_arguments = [ 

80 ("fakeserver", ["net", "conf", "addshare", "fakeshareid", 

81 "/shares/fakeshareid", "writeable=y", "guest_ok=n"]), 

82 ("fakeserver", ["net", "conf", "setparm", "fakeshareid", 

83 "browseable", "yes"]), 

84 ("fakeserver", ["net", "conf", "setparm", "fakeshareid", 

85 "hosts allow", "192.0.2.2"]), 

86 ("fakeserver", ["net", "conf", "setparm", "fakeshareid", 

87 "read only", "no"]), 

88 ("fakeserver", ["net", "conf", "setparm", "fakeshareid", 

89 "hosts deny", "0.0.0.0/0"]), 

90 ("fakeserver", ["net", "conf", "setparm", "fakeshareid", 

91 "create mask", "0755"])] 

92 actual_arguments = [] 

93 self._helper.execute = functools.partial( 

94 self.fake_exec_sync, execute_arguments=actual_arguments, 

95 ret_val=" fake 192.0.2.2/24 more fake \n" * 20) 

96 self.DockerCIFSHelper.share = fake_share() 

97 self.mock_object(self.DockerCIFSHelper.container, 

98 'fetch_container_addresses', 

99 mock.Mock(return_value=['192.0.2.2'])) 

100 

101 self.DockerCIFSHelper.create_share("fakeserver") 

102 

103 self.assertEqual(expected_arguments.sort(), actual_arguments.sort()) 

104 

105 def test_delete_share(self): 

106 self.DockerCIFSHelper.share = fake_share() 

107 

108 self.DockerCIFSHelper.delete_share("fakeserver", "fakeshareid") 

109 

110 self.DockerCIFSHelper.container.execute.assert_called_with( 

111 "fakeserver", 

112 ["net", "conf", "delshare", "fakeshareid"], ignore_errors=False) 

113 

114 def test__get_access_group_ro(self): 

115 result = self.DockerCIFSHelper._get_access_group(const.ACCESS_LEVEL_RO) 

116 

117 self.assertEqual("read list", result) 

118 

119 def test__get_access_group_rw(self): 

120 result = self.DockerCIFSHelper._get_access_group(const.ACCESS_LEVEL_RW) 

121 

122 self.assertEqual("valid users", result) 

123 

124 def test__get_access_group_other(self): 

125 self.assertRaises(exception.InvalidShareAccessLevel, 

126 self.DockerCIFSHelper._get_access_group, 

127 "fake_level") 

128 

129 def test__get_existing_users(self): 

130 self.DockerCIFSHelper.container.execute = mock.Mock( 

131 return_value=("fake_user", "")) 

132 

133 result = self.DockerCIFSHelper._get_existing_users("fake_server_id", 

134 "fake_share", 

135 "fake_access") 

136 

137 self.assertEqual("fake_user", result) 

138 self.DockerCIFSHelper.container.execute.assert_called_once_with( 

139 "fake_server_id", 

140 ["net", "conf", "getparm", "fake_share", "fake_access"], 

141 ignore_errors=True) 

142 

143 def test__set_users(self): 

144 self.DockerCIFSHelper.container.execute = mock.Mock() 

145 

146 self.DockerCIFSHelper._set_users("fake_server_id", "fake_share", 

147 "fake_access", "fake_user") 

148 

149 self.DockerCIFSHelper.container.execute.assert_called_once_with( 

150 "fake_server_id", 

151 ["net", "conf", "setparm", "fake_share", "fake_access", 

152 "fake_user"]) 

153 

154 def test__allow_access_ok(self): 

155 self.DockerCIFSHelper._get_access_group = mock.Mock( 

156 return_value="valid users") 

157 self.DockerCIFSHelper._get_existing_users = mock.Mock( 

158 return_value="fake_user") 

159 self.DockerCIFSHelper._set_users = mock.Mock() 

160 

161 self.DockerCIFSHelper._allow_access("fake_share", "fake_server_id", 

162 "fake_user2", "rw") 

163 

164 self.DockerCIFSHelper._get_access_group.assert_called_once_with("rw") 

165 self.DockerCIFSHelper._get_existing_users.assert_called_once_with( 

166 "fake_server_id", "fake_share", "valid users") 

167 self.DockerCIFSHelper._set_users.assert_called_once_with( 

168 "fake_server_id", "fake_share", "valid users", 

169 "fake_user fake_user2") 

170 

171 def test__allow_access_not_ok(self): 

172 self.DockerCIFSHelper._get_access_group = mock.Mock( 

173 return_value="valid users") 

174 self.DockerCIFSHelper._get_existing_users = mock.Mock() 

175 self.DockerCIFSHelper._get_existing_users.side_effect = TypeError 

176 self.DockerCIFSHelper._set_users = mock.Mock() 

177 

178 self.DockerCIFSHelper._allow_access("fake_share", "fake_server_id", 

179 "fake_user2", "rw") 

180 

181 self.DockerCIFSHelper._get_access_group.assert_called_once_with("rw") 

182 self.DockerCIFSHelper._get_existing_users.assert_called_once_with( 

183 "fake_server_id", "fake_share", "valid users") 

184 self.DockerCIFSHelper._set_users.assert_called_once_with( 

185 "fake_server_id", "fake_share", "valid users", 

186 "fake_user2") 

187 

188 def test__deny_access_ok(self): 

189 self.DockerCIFSHelper._get_access_group = mock.Mock( 

190 return_value="valid users") 

191 self.DockerCIFSHelper._get_existing_users = mock.Mock( 

192 return_value="fake_user fake_user2") 

193 self.DockerCIFSHelper._set_users = mock.Mock() 

194 

195 self.DockerCIFSHelper._deny_access("fake_share", "fake_server_id", 

196 "fake_user2", "rw") 

197 

198 self.DockerCIFSHelper._get_access_group.assert_called_once_with("rw") 

199 self.DockerCIFSHelper._get_existing_users.assert_called_once_with( 

200 "fake_server_id", "fake_share", "valid users") 

201 self.DockerCIFSHelper._set_users.assert_called_once_with( 

202 "fake_server_id", "fake_share", "valid users", 

203 "fake_user") 

204 

205 def test__deny_access_ok_so_many_users(self): 

206 self.DockerCIFSHelper._get_access_group = mock.Mock( 

207 return_value="valid users") 

208 self.DockerCIFSHelper._get_existing_users = mock.Mock( 

209 return_value="joost jaap huub dirk") 

210 self.DockerCIFSHelper._set_users = mock.Mock() 

211 

212 # Sorry, Jaap. 

213 self.DockerCIFSHelper._deny_access("fake_share", "fake_server_id", 

214 "jaap", "rw") 

215 

216 self.DockerCIFSHelper._get_access_group.assert_called_once_with("rw") 

217 self.DockerCIFSHelper._get_existing_users.assert_called_once_with( 

218 "fake_server_id", "fake_share", "valid users") 

219 self.DockerCIFSHelper._set_users.assert_called_once_with( 

220 "fake_server_id", "fake_share", "valid users", 

221 "dirk huub joost") 

222 

223 def test__deny_access_not_ok(self): 

224 self.DockerCIFSHelper._get_access_group = mock.Mock( 

225 return_value="valid users") 

226 self.DockerCIFSHelper._get_existing_users = mock.Mock() 

227 self.DockerCIFSHelper._get_existing_users.side_effect = TypeError 

228 self.DockerCIFSHelper._set_users = mock.Mock() 

229 self.mock_object(protocol_helper.LOG, "warning") 

230 

231 self.DockerCIFSHelper._deny_access("fake_share", "fake_server_id", 

232 "fake_user2", "rw") 

233 

234 self.DockerCIFSHelper._get_access_group.assert_called_once_with("rw") 

235 self.DockerCIFSHelper._get_existing_users.assert_called_once_with( 

236 "fake_server_id", "fake_share", "valid users") 

237 self.assertFalse(self.DockerCIFSHelper._set_users.called) 

238 self.assertTrue(protocol_helper.LOG.warning.called) 

239 

240 def test_update_access_access_rules_wrong_type(self): 

241 allow_rules = [{ 

242 "access_to": "192.0.2.2", 

243 "access_level": "ro", 

244 "access_type": "fake" 

245 }] 

246 self.mock_object(self.DockerCIFSHelper, "_allow_access") 

247 

248 self.assertRaises(exception.InvalidShareAccess, 

249 self.DockerCIFSHelper.update_access, 

250 "fakeserver", "fakeshareid", allow_rules, [], []) 

251 

252 def test_update_access_access_rules_ok(self): 

253 access_rules = [{ 

254 "access_to": "fakeuser", 

255 "access_level": "ro", 

256 "access_type": "user" 

257 }] 

258 self.mock_object(self.DockerCIFSHelper, "_allow_access") 

259 self.DockerCIFSHelper.container.execute = mock.Mock() 

260 

261 self.DockerCIFSHelper.update_access("fakeserver", "fakeshareid", 

262 access_rules, [], []) 

263 

264 self.DockerCIFSHelper._allow_access.assert_called_once_with( 

265 "fakeshareid", 

266 "fakeserver", 

267 "fakeuser", 

268 "ro") 

269 self.DockerCIFSHelper.container.execute.assert_called_once_with( 

270 "fakeserver", 

271 ["net", "conf", "setparm", "fakeshareid", "valid users", ""]) 

272 

273 def test_update_access_add_rules(self): 

274 add_rules = [{ 

275 "access_to": "fakeuser", 

276 "access_level": "ro", 

277 "access_type": "user" 

278 }] 

279 self.mock_object(self.DockerCIFSHelper, "_allow_access") 

280 

281 self.DockerCIFSHelper.update_access("fakeserver", "fakeshareid", [], 

282 add_rules, []) 

283 

284 self.DockerCIFSHelper._allow_access.assert_called_once_with( 

285 "fakeshareid", 

286 "fakeserver", 

287 "fakeuser", 

288 "ro") 

289 

290 def test_update_access_delete_rules(self): 

291 delete_rules = [{ 

292 "access_to": "fakeuser", 

293 "access_level": "ro", 

294 "access_type": "user" 

295 }] 

296 self.mock_object(self.DockerCIFSHelper, "_deny_access") 

297 

298 self.DockerCIFSHelper.update_access("fakeserver", "fakeshareid", 

299 [], [], delete_rules) 

300 

301 self.DockerCIFSHelper._deny_access.assert_called_once_with( 

302 "fakeshareid", 

303 "fakeserver", 

304 "fakeuser", 

305 "ro")