Coverage for manila/tests/share/drivers/windows/test_windows_smb_driver.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.share import configuration 

22from manila.share.drivers import generic 

23from manila.share.drivers.windows import service_instance 

24from manila.share.drivers.windows import windows_smb_driver as windows_drv 

25from manila.share.drivers.windows import windows_smb_helper 

26from manila.share.drivers.windows import windows_utils 

27from manila.share.drivers.windows import winrm_helper 

28from manila import test 

29from manila.tests import fake_share 

30 

31 

32@ddt.ddt 

33class WindowsSMBDriverTestCase(test.TestCase): 

34 @mock.patch.object(winrm_helper, 'WinRMHelper') 

35 @mock.patch.object(windows_utils, 'WindowsUtils') 

36 @mock.patch.object(windows_smb_helper, 'WindowsSMBHelper') 

37 @mock.patch.object(service_instance, 

38 'WindowsServiceInstanceManager') 

39 def setUp(self, mock_sv_instance_mgr, mock_smb_helper_cls, 

40 mock_utils_cls, mock_winrm_helper_cls): 

41 self.flags(driver_handles_share_servers=True) 

42 self._fake_conf = configuration.Configuration(None) 

43 

44 self._share = fake_share.fake_share(share_proto='SMB') 

45 self._share_server = dict( 

46 backend_details=mock.sentinel.backend_details) 

47 

48 self._drv = windows_drv.WindowsSMBDriver( 

49 configuration=self._fake_conf) 

50 self._drv._setup_helpers() 

51 

52 self._remote_execute = mock_winrm_helper_cls.return_value 

53 self._windows_utils = mock_utils_cls.return_value 

54 self._smb_helper = mock_smb_helper_cls.return_value 

55 super(WindowsSMBDriverTestCase, self).setUp() 

56 

57 @mock.patch('manila.share.driver.ShareDriver') 

58 def test_update_share_stats(self, mock_base_driver): 

59 self._drv._update_share_stats() 

60 mock_base_driver._update_share_stats.assert_called_once_with( 

61 self._drv, 

62 data=dict(storage_protocol="CIFS")) 

63 

64 @mock.patch.object(service_instance, 'WindowsServiceInstanceManager') 

65 def test_setup_service_instance_manager(self, mock_sv_instance_mgr): 

66 self._drv._setup_service_instance_manager() 

67 mock_sv_instance_mgr.assert_called_once_with( 

68 driver_config=self._fake_conf) 

69 

70 def test_setup_helpers(self): 

71 expected_helpers = {"SMB": self._smb_helper, 

72 "CIFS": self._smb_helper} 

73 self._drv._setup_helpers() 

74 self.assertEqual(expected_helpers, self._drv._helpers) 

75 

76 @mock.patch.object(generic.GenericShareDriver, '_teardown_server') 

77 def test_teardown_server(self, mock_super_teardown): 

78 mock_server = {'joined_domain': True, 

79 'instance_id': mock.sentinel.instance_id} 

80 mock_sec_service = {'user': mock.sentinel.user, 

81 'password': mock.sentinel.password, 

82 'domain': mock.sentinel.domain} 

83 

84 sv_mgr = self._drv.service_instance_manager 

85 sv_mgr.get_valid_security_service.return_value = mock_sec_service 

86 # We ensure that domain unjoin exceptions do not prevent the 

87 # service instance from being teared down. 

88 self._windows_utils.unjoin_domain.side_effect = Exception 

89 

90 self._drv._teardown_server(mock_server, 

91 mock_sec_service) 

92 

93 sv_mgr.get_valid_security_service.assert_called_once_with( 

94 mock_sec_service) 

95 self._windows_utils.unjoin_domain.assert_called_once_with( 

96 mock_server, 

97 mock_sec_service['user'], 

98 mock_sec_service['password']) 

99 mock_super_teardown.assert_called_once_with(mock_server, 

100 mock_sec_service) 

101 

102 @mock.patch.object(windows_drv.WindowsSMBDriver, '_get_disk_number') 

103 def test_format_device(self, mock_get_disk_number): 

104 mock_get_disk_number.return_value = mock.sentinel.disk_number 

105 

106 self._drv._format_device(mock.sentinel.server, mock.sentinel.vol) 

107 

108 self._drv._get_disk_number.assert_called_once_with( 

109 mock.sentinel.server, mock.sentinel.vol) 

110 self._windows_utils.initialize_disk.assert_called_once_with( 

111 mock.sentinel.server, mock.sentinel.disk_number) 

112 self._windows_utils.create_partition.assert_called_once_with( 

113 mock.sentinel.server, mock.sentinel.disk_number) 

114 self._windows_utils.format_partition.assert_called_once_with( 

115 mock.sentinel.server, mock.sentinel.disk_number, 

116 self._drv._DEFAULT_SHARE_PARTITION) 

117 

118 @mock.patch.object(windows_drv.WindowsSMBDriver, 

119 '_ensure_disk_online_and_writable') 

120 @mock.patch.object(windows_drv.WindowsSMBDriver, '_get_disk_number') 

121 @mock.patch.object(windows_drv.WindowsSMBDriver, '_get_mount_path') 

122 @mock.patch.object(windows_drv.WindowsSMBDriver, '_is_device_mounted') 

123 def test_mount_device(self, mock_device_mounted, mock_get_mount_path, 

124 mock_get_disk_number, mock_ensure_disk): 

125 mock_get_mount_path.return_value = mock.sentinel.mount_path 

126 mock_get_disk_number.return_value = mock.sentinel.disk_number 

127 mock_device_mounted.return_value = False 

128 

129 self._drv._mount_device(share=mock.sentinel.share, 

130 server_details=mock.sentinel.server, 

131 volume=mock.sentinel.vol) 

132 

133 mock_device_mounted.assert_called_once_with( 

134 mock.sentinel.mount_path, mock.sentinel.server, mock.sentinel.vol) 

135 mock_get_disk_number.assert_called_once_with( 

136 mock.sentinel.server, mock.sentinel.vol) 

137 self._windows_utils.ensure_directory_exists.assert_called_once_with( 

138 mock.sentinel.server, mock.sentinel.mount_path) 

139 self._windows_utils.add_access_path( 

140 mock.sentinel.server, 

141 mock.sentinel.mount_path, 

142 mock.sentinel.disk_number, 

143 self._drv._DEFAULT_SHARE_PARTITION) 

144 mock_ensure_disk.assert_called_once_with( 

145 mock.sentinel.server, mock.sentinel.disk_number) 

146 

147 @mock.patch.object(windows_drv.WindowsSMBDriver, '_get_mount_path') 

148 def test_unmount_device(self, mock_get_mount_path): 

149 mock_get_mount_path.return_value = mock.sentinel.mount_path 

150 mock_get_disk_number_by_path = ( 

151 self._windows_utils.get_disk_number_by_mount_path) 

152 

153 self._drv._unmount_device(mock.sentinel.share, 

154 mock.sentinel.server) 

155 

156 mock_get_mount_path.assert_called_once_with(mock.sentinel.share) 

157 

158 mock_get_disk_number_by_path.assert_called_once_with( 

159 mock.sentinel.server, mock.sentinel.mount_path) 

160 self._windows_utils.set_disk_online_status.assert_called_once_with( 

161 mock.sentinel.server, 

162 mock_get_disk_number_by_path.return_value, 

163 online=False) 

164 

165 @ddt.data(None, 1) 

166 @mock.patch.object(windows_drv.WindowsSMBDriver, '_get_disk_number') 

167 @mock.patch.object(windows_drv.WindowsSMBDriver, 

168 '_ensure_disk_online_and_writable') 

169 def test_resize_filesystem(self, new_size, mock_ensure_disk, 

170 mock_get_disk_number): 

171 mock_get_disk_number.return_value = mock.sentinel.disk_number 

172 mock_get_max_size = self._windows_utils.get_partition_maximum_size 

173 mock_get_max_size.return_value = mock.sentinel.max_size 

174 

175 self._drv._resize_filesystem(mock.sentinel.server, 

176 mock.sentinel.vol, 

177 new_size=new_size) 

178 

179 mock_get_disk_number.assert_called_once_with(mock.sentinel.server, 

180 mock.sentinel.vol) 

181 self._drv._ensure_disk_online_and_writable.assert_called_once_with( 

182 mock.sentinel.server, mock.sentinel.disk_number) 

183 

184 if not new_size: 

185 mock_get_max_size.assert_called_once_with( 

186 mock.sentinel.server, 

187 mock.sentinel.disk_number, 

188 self._drv._DEFAULT_SHARE_PARTITION) 

189 expected_new_size = mock.sentinel.max_size 

190 else: 

191 expected_new_size = new_size << 30 

192 

193 self._windows_utils.resize_partition.assert_called_once_with( 

194 mock.sentinel.server, 

195 expected_new_size, 

196 mock.sentinel.disk_number, 

197 self._drv._DEFAULT_SHARE_PARTITION) 

198 

199 def test_ensure_disk_online_and_writable(self): 

200 self._drv._ensure_disk_online_and_writable( 

201 mock.sentinel.server, mock.sentinel.disk_number) 

202 

203 self._windows_utils.update_disk.assert_called_once_with( 

204 mock.sentinel.server, mock.sentinel.disk_number) 

205 self._windows_utils.set_disk_online_status.assert_called_once_with( 

206 mock.sentinel.server, mock.sentinel.disk_number, online=True) 

207 self._windows_utils.set_disk_readonly_status.assert_called_once_with( 

208 mock.sentinel.server, mock.sentinel.disk_number, readonly=False) 

209 

210 def test_get_mounted_share_size(self): 

211 fake_size_gb = 10 

212 self._windows_utils.get_disk_space_by_path.return_value = ( 

213 fake_size_gb << 30, mock.sentinel.free_bytes) 

214 

215 share_size = self._drv._get_mounted_share_size( 

216 mock.sentinel.mount_path, 

217 mock.sentinel.server) 

218 

219 self.assertEqual(fake_size_gb, share_size) 

220 

221 def test_get_consumed_space(self): 

222 fake_size_gb = 2 

223 fake_free_space_gb = 1 

224 self._windows_utils.get_disk_space_by_path.return_value = ( 

225 fake_size_gb << 30, fake_free_space_gb << 30) 

226 

227 consumed_space = self._drv._get_consumed_space( 

228 mock.sentinel.mount_path, 

229 mock.sentinel.server) 

230 

231 self.assertEqual(fake_size_gb - fake_free_space_gb, consumed_space) 

232 

233 def test_get_mount_path(self): 

234 fake_mount_path = 'fake_mount_path' 

235 fake_share_name = 'fake_share_name' 

236 mock_share = {'name': fake_share_name} 

237 self.flags(share_mount_path=fake_mount_path) 

238 

239 mount_path = self._drv._get_mount_path(mock_share) 

240 

241 self._windows_utils.normalize_path.assert_called_once_with( 

242 os.path.join(fake_mount_path, fake_share_name)) 

243 self.assertEqual(self._windows_utils.normalize_path.return_value, 

244 mount_path) 

245 

246 @ddt.data(None, 2) 

247 def test_get_disk_number(self, disk_number_by_serial=None): 

248 mock_get_disk_number_by_serial = ( 

249 self._windows_utils.get_disk_number_by_serial_number) 

250 

251 mock_get_disk_number_by_serial.return_value = disk_number_by_serial 

252 mock_volume = {'id': mock.sentinel.vol_id, 

253 'mountpoint': "/dev/sdb"} 

254 # If the disk number cannot be identified using the disk serial 

255 # number, we expect it to be retrieved based on the volume mountpoint, 

256 # having disk number 1 in this case. 

257 expected_disk_number = (disk_number_by_serial 

258 if disk_number_by_serial else 1) 

259 

260 disk_number = self._drv._get_disk_number(mock.sentinel.server, 

261 mock_volume) 

262 

263 mock_get_disk_number_by_serial.assert_called_once_with( 

264 mock.sentinel.server, mock.sentinel.vol_id) 

265 self.assertEqual(expected_disk_number, disk_number) 

266 

267 @ddt.data(None, 2) 

268 def test_is_device_mounted(self, disk_number_by_path): 

269 mock_get_disk_number_by_path = ( 

270 self._windows_utils.get_disk_number_by_mount_path) 

271 mock_get_disk_number_by_path.return_value = disk_number_by_path 

272 

273 expected_result = disk_number_by_path is not None 

274 is_mounted = self._drv._is_device_mounted( 

275 mount_path=mock.sentinel.mount_path, 

276 server_details=mock.sentinel.server) 

277 

278 mock_get_disk_number_by_path.assert_called_once_with( 

279 mock.sentinel.server, 

280 mock.sentinel.mount_path) 

281 self.assertEqual(expected_result, is_mounted)