« back to all changes in this revision
Viewing changes to tests/test_rw_handles.py
- browse files at revision 21
- compare with another revision
- download diff
- download tarball
- view history from revision 21
- Committer: Package Import Robot
- Author(s): James Page
- Date: 2015-08-11 09:25:22 UTC
- mfrom: (1.1.12) (2.1.3 experimental)
- Revision ID: package-import@ubuntu.com-20150811092522-6epbeuzn6a0jt750
Tags: 1.18.0-2ubuntu1
* Resync with Debian experimental.
* d/pydist-overrides: Map suds-jurko -> suds to ease backporting.
* d/pydist-overrides: Map suds-jurko -> suds to ease backporting.
- files removed:
- oslo
- oslo/vmware
- oslo/vmware/objects
- tests
- tests/objects
- files modified:
- bandit.yaml
- setup.py *
added
removed
tests/test_rw_handles.py
1
# Copyright (c) 2014 VMware, 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
16
"""
17
Unit tests for read and write handles for image transfer.
18
"""
19
20
import mock
21
import six
22
23
from oslo.vmware import exceptions
24
from oslo.vmware import rw_handles
25
from oslo_vmware import vim_util as new_vim_util
26
from tests import base
27
28
29
class FileHandleTest(base.TestCase):
30
"""Tests for FileHandle."""
31
32
def test_close(self):
33
file_handle = mock.Mock()
34
vmw_http_file = rw_handles.FileHandle(file_handle)
35
vmw_http_file.close()
36
file_handle.close.assert_called_once_with()
37
38
def test_find_vmdk_url(self):
39
device_url_0 = mock.Mock()
40
device_url_0.disk = False
41
device_url_1 = mock.Mock()
42
device_url_1.disk = True
43
device_url_1.url = 'https://*/ds1/vm1.vmdk'
44
lease_info = mock.Mock()
45
lease_info.deviceUrl = [device_url_0, device_url_1]
46
host = '10.1.2.3'
47
port = 443
48
exp_url = 'https://%s:%d/ds1/vm1.vmdk' % (host, port)
49
vmw_http_file = rw_handles.FileHandle(None)
50
self.assertEqual(exp_url, vmw_http_file._find_vmdk_url(lease_info,
51
host,
52
port))
53
54
55
class FileWriteHandleTest(base.TestCase):
56
"""Tests for FileWriteHandle."""
57
58
def setUp(self):
59
super(FileWriteHandleTest, self).setUp()
60
61
vim_cookie = mock.Mock()
62
vim_cookie.name = 'name'
63
vim_cookie.value = 'value'
64
65
self._conn = mock.Mock()
66
patcher = mock.patch(
67
'urllib3.connection.HTTPConnection')
68
self.addCleanup(patcher.stop)
69
HTTPConnectionMock = patcher.start()
70
HTTPConnectionMock.return_value = self._conn
71
72
self.vmw_http_write_file = rw_handles.FileWriteHandle(
73
'10.1.2.3', 443, 'dc-0', 'ds-0', [vim_cookie], '1.vmdk', 100,
74
'http')
75
76
def test_write(self):
77
self.vmw_http_write_file.write(None)
78
self._conn.send.assert_called_once_with(None)
79
80
def test_close(self):
81
self.vmw_http_write_file.close()
82
self._conn.getresponse.assert_called_once_with()
83
self._conn.close.assert_called_once_with()
84
85
86
class VmdkWriteHandleTest(base.TestCase):
87
"""Tests for VmdkWriteHandle."""
88
89
def setUp(self):
90
super(VmdkWriteHandleTest, self).setUp()
91
self._conn = mock.Mock()
92
patcher = mock.patch(
93
'urllib3.connection.HTTPConnection')
94
self.addCleanup(patcher.stop)
95
HTTPConnectionMock = patcher.start()
96
HTTPConnectionMock.return_value = self._conn
97
98
def _create_mock_session(self, disk=True, progress=-1):
99
device_url = mock.Mock()
100
device_url.disk = disk
101
device_url.url = 'http://*/ds/disk1.vmdk'
102
lease_info = mock.Mock()
103
lease_info.deviceUrl = [device_url]
104
session = mock.Mock()
105
106
def session_invoke_api_side_effect(module, method, *args, **kwargs):
107
if module == session.vim:
108
if method == 'ImportVApp':
109
return mock.Mock()
110
elif method == 'HttpNfcLeaseProgress':
111
self.assertEqual(progress, kwargs['percent'])
112
return
113
return lease_info
114
115
session.invoke_api.side_effect = session_invoke_api_side_effect
116
vim_cookie = mock.Mock()
117
vim_cookie.name = 'name'
118
vim_cookie.value = 'value'
119
session.vim.client.options.transport.cookiejar = [vim_cookie]
120
return session
121
122
def test_init_failure(self):
123
session = self._create_mock_session(False)
124
self.assertRaises(exceptions.VimException,
125
rw_handles.VmdkWriteHandle,
126
session,
127
'10.1.2.3',
128
443,
129
'rp-1',
130
'folder-1',
131
None,
132
100)
133
134
def test_write(self):
135
session = self._create_mock_session()
136
handle = rw_handles.VmdkWriteHandle(session, '10.1.2.3', 443,
137
'rp-1', 'folder-1', None,
138
100)
139
data = [1] * 10
140
handle.write(data)
141
self.assertEqual(len(data), handle._bytes_written)
142
self._conn.send.assert_called_once_with(data)
143
144
def test_update_progress(self):
145
vmdk_size = 100
146
data_size = 10
147
session = self._create_mock_session(True, 10)
148
handle = rw_handles.VmdkWriteHandle(session, '10.1.2.3', 443,
149
'rp-1', 'folder-1', None,
150
vmdk_size)
151
handle.write([1] * data_size)
152
handle.update_progress()
153
154
def test_update_progress_with_error(self):
155
session = self._create_mock_session(True, 10)
156
handle = rw_handles.VmdkWriteHandle(session, '10.1.2.3', 443,
157
'rp-1', 'folder-1', None,
158
100)
159
session.invoke_api.side_effect = exceptions.VimException(None)
160
self.assertRaises(exceptions.VimException, handle.update_progress)
161
162
def test_close(self):
163
session = self._create_mock_session()
164
handle = rw_handles.VmdkWriteHandle(session, '10.1.2.3', 443,
165
'rp-1', 'folder-1', None,
166
100)
167
168
def session_invoke_api_side_effect(module, method, *args, **kwargs):
169
if module == new_vim_util and method == 'get_object_property':
170
return 'ready'
171
self.assertEqual(session.vim, module)
172
self.assertEqual('HttpNfcLeaseComplete', method)
173
174
session.invoke_api = mock.Mock(
175
side_effect=session_invoke_api_side_effect)
176
handle.close()
177
self.assertEqual(2, session.invoke_api.call_count)
178
179
180
class VmdkReadHandleTest(base.TestCase):
181
"""Tests for VmdkReadHandle."""
182
183
def setUp(self):
184
super(VmdkReadHandleTest, self).setUp()
185
186
send_patcher = mock.patch('requests.sessions.Session.send')
187
self.addCleanup(send_patcher.stop)
188
send_mock = send_patcher.start()
189
self._response = mock.Mock()
190
send_mock.return_value = self._response
191
192
def _create_mock_session(self, disk=True, progress=-1):
193
device_url = mock.Mock()
194
device_url.disk = disk
195
device_url.url = 'http://*/ds/disk1.vmdk'
196
lease_info = mock.Mock()
197
lease_info.deviceUrl = [device_url]
198
session = mock.Mock()
199
200
def session_invoke_api_side_effect(module, method, *args, **kwargs):
201
if module == session.vim:
202
if method == 'ExportVm':
203
return mock.Mock()
204
elif method == 'HttpNfcLeaseProgress':
205
self.assertEqual(progress, kwargs['percent'])
206
return
207
return lease_info
208
209
session.invoke_api.side_effect = session_invoke_api_side_effect
210
vim_cookie = mock.Mock()
211
vim_cookie.name = 'name'
212
vim_cookie.value = 'value'
213
session.vim.client.options.transport.cookiejar = [vim_cookie]
214
return session
215
216
def test_init_failure(self):
217
session = self._create_mock_session(False)
218
self.assertRaises(exceptions.VimException,
219
rw_handles.VmdkReadHandle,
220
session,
221
'10.1.2.3',
222
443,
223
'vm-1',
224
'[ds] disk1.vmdk',
225
100)
226
227
def test_read(self):
228
chunk_size = rw_handles.READ_CHUNKSIZE
229
session = self._create_mock_session()
230
self._response.raw.read.return_value = [1] * chunk_size
231
handle = rw_handles.VmdkReadHandle(session, '10.1.2.3', 443,
232
'vm-1', '[ds] disk1.vmdk',
233
chunk_size * 10)
234
handle.read(chunk_size)
235
self.assertEqual(chunk_size, handle._bytes_read)
236
self._response.raw.read.assert_called_once_with(chunk_size)
237
238
def test_update_progress(self):
239
chunk_size = rw_handles.READ_CHUNKSIZE
240
vmdk_size = chunk_size * 10
241
session = self._create_mock_session(True, 10)
242
self._response.raw.read.return_value = [1] * chunk_size
243
handle = rw_handles.VmdkReadHandle(session, '10.1.2.3', 443,
244
'vm-1', '[ds] disk1.vmdk',
245
vmdk_size)
246
handle.read(chunk_size)
247
handle.update_progress()
248
self._response.raw.read.assert_called_once_with(chunk_size)
249
250
def test_update_progress_with_error(self):
251
session = self._create_mock_session(True, 10)
252
handle = rw_handles.VmdkReadHandle(session, '10.1.2.3', 443,
253
'vm-1', '[ds] disk1.vmdk',
254
100)
255
session.invoke_api.side_effect = exceptions.VimException(None)
256
self.assertRaises(exceptions.VimException, handle.update_progress)
257
258
def test_close(self):
259
session = self._create_mock_session()
260
handle = rw_handles.VmdkReadHandle(session, '10.1.2.3', 443,
261
'vm-1', '[ds] disk1.vmdk',
262
100)
263
264
def session_invoke_api_side_effect(module, method, *args, **kwargs):
265
if module == new_vim_util and method == 'get_object_property':
266
return 'ready'
267
self.assertEqual(session.vim, module)
268
self.assertEqual('HttpNfcLeaseComplete', method)
269
270
session.invoke_api = mock.Mock(
271
side_effect=session_invoke_api_side_effect)
272
handle.close()
273
self.assertEqual(2, session.invoke_api.call_count)
274
275
276
class ImageReadHandleTest(base.TestCase):
277
"""Tests for ImageReadHandle."""
278
279
def test_read(self):
280
max_items = 10
281
item = [1] * 10
282
283
class ImageReadIterator(six.Iterator):
284
285
def __init__(self):
286
self.num_items = 0
287
288
def __iter__(self):
289
return self
290
291
def __next__(self):
292
if (self.num_items < max_items):
293
self.num_items += 1
294
return item
295
raise StopIteration
296
297
next = __next__
298
299
handle = rw_handles.ImageReadHandle(ImageReadIterator())
300
for _ in range(0, max_items):
301
self.assertEqual(item, handle.read(10))
302
self.assertFalse(handle.read(10))
Loggerhead is a web-based interface for Breezy
Version: 2.0.1