« back to all changes in this revision
Viewing changes to bzrlib/selftest/testtransport.py
- browse files at revision 1185.11.22
- compare with another revision
- compare with revision 1185.46.2
- download diff
- download tarball
- view history from revision 1185.11.22
- Committer: John Arbash Meinel
- Date: 2005-09-19 03:26:14 UTC
- mto: (1393.2.1)
- mto: This revision was merged to the branch mainline in revision 1396.
- Revision ID: john@arbash-meinel.com-20050919032614-79751b0510f24929
Major refactoring of testtransport.
Rewrote test_transport() as a class which can be subclassed which provides
many test_* functions, rather than one large function.
Much easier to debug.
Rewrote test_transport() as a class which can be subclassed which provides
many test_* functions, rather than one large function.
Much easier to debug.
- files modified:
- bzrlib/selftest/testtransport.py
added
removed
bzrlib/selftest/testtransport.py
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
16
16
17
17
18
import os
19
from cStringIO import StringIO
18
20
from bzrlib.selftest import TestCaseInTempDir
19
21
from bzrlib.selftest.HTTPTestUtil import TestCaseWithWebserver
20
22
from bzrlib.errors import NoSuchFile, FileExists, TransportNotPossible
21
from cStringIO import StringIO
22
23
23
24
def _append(fn, txt):
24
25
"""Append the given text (file-like object) to the supplied filename."""
28
29
f.close()
29
30
del f
30
31
31
def test_transport(tester, t, readonly=False):
32
"""Test a transport object. Basically, it assumes that the
33
Transport object is connected to the current working directory.
34
So that whatever is done through the transport, should show
35
up in the working directory, and vice-versa.
32
class TestTransportMixIn(object):
33
"""Subclass this, and it will provide a series of tests for a Transport.
34
It assumes that the Transport object is connected to the
35
current working directory. So that whatever is done
36
through the transport, should show up in the working
37
directory, and vice-versa.
36
38
37
39
This also tests to make sure that the functions work with both
38
40
generators and lists (assuming iter(list) is effectively a generator)
39
41
"""
40
import tempfile, os
41
from bzrlib.transport.local import LocalTransport
42
43
# Test has
44
files = ['a', 'b', 'e', 'g']
45
tester.build_tree(files)
46
tester.assertEqual(t.has('a'), True)
47
tester.assertEqual(t.has('c'), False)
48
tester.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
49
[True, True, False, False, True, False, True, False])
50
tester.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
51
[True, True, False, False, True, False, True, False])
52
53
# Test get
54
tester.assertEqual(t.get('a').read(), open('a').read())
55
content_f = t.get_multi(files)
56
for path,f in zip(files, content_f):
57
tester.assertEqual(open(path).read(), f.read())
58
59
content_f = t.get_multi(iter(files))
60
for path,f in zip(files, content_f):
61
tester.assertEqual(open(path).read(), f.read())
62
63
tester.assertRaises(NoSuchFile, t.get, 'c')
64
try:
65
files = list(t.get_multi(['a', 'b', 'c']))
66
except NoSuchFile:
67
pass
68
else:
69
tester.fail('Failed to raise NoSuchFile for missing file in get_multi')
70
try:
71
files = list(t.get_multi(iter(['a', 'b', 'c', 'e'])))
72
except NoSuchFile:
73
pass
74
else:
75
tester.fail('Failed to raise NoSuchFile for missing file in get_multi')
76
77
# Test put
78
if readonly:
79
tester.assertRaises(TransportNotPossible,
80
t.put, 'c', 'some text for c\n')
81
open('c', 'wb').write('some text for c\n')
82
else:
83
t.put('c', 'some text for c\n')
84
tester.assert_(os.path.exists('c'))
85
tester.check_file_contents('c', 'some text for c\n')
86
tester.assertEqual(t.get('c').read(), 'some text for c\n')
87
# Make sure 'has' is updated
88
tester.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
89
[True, True, True, False, True, False, True, False])
90
if readonly:
91
tester.assertRaises(TransportNotPossible,
92
t.put_multi,
93
[('a', 'new\ncontents for\na\n'),
94
('d', 'contents\nfor d\n')])
95
open('a', 'wb').write('new\ncontents for\na\n')
96
open('d', 'wb').write('contents\nfor d\n')
97
else:
98
# Put also replaces contents
99
tester.assertEqual(t.put_multi([('a', 'new\ncontents for\na\n'),
100
('d', 'contents\nfor d\n')]),
101
2)
102
tester.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
103
[True, True, True, True, True, False, True, False])
104
tester.check_file_contents('a', 'new\ncontents for\na\n')
105
tester.check_file_contents('d', 'contents\nfor d\n')
106
107
if readonly:
108
tester.assertRaises(TransportNotPossible,
109
t.put_multi, iter([('a', 'diff\ncontents for\na\n'),
110
('d', 'another contents\nfor d\n')]))
111
open('a', 'wb').write('diff\ncontents for\na\n')
112
open('d', 'wb').write('another contents\nfor d\n')
113
else:
114
tester.assertEqual(
115
t.put_multi(iter([('a', 'diff\ncontents for\na\n'),
116
('d', 'another contents\nfor d\n')]))
117
, 2)
118
tester.check_file_contents('a', 'diff\ncontents for\na\n')
119
tester.check_file_contents('d', 'another contents\nfor d\n')
120
121
if readonly:
122
tester.assertRaises(TransportNotPossible,
123
t.put, 'path/doesnt/exist/c', 'contents')
124
else:
125
tester.assertRaises(NoSuchFile,
126
t.put, 'path/doesnt/exist/c', 'contents')
127
128
# Test mkdir
129
os.mkdir('dir_a')
130
tester.assertEqual(t.has('dir_a'), True)
131
tester.assertEqual(t.has('dir_b'), False)
132
133
if readonly:
134
tester.assertRaises(TransportNotPossible,
135
t.mkdir, 'dir_b')
136
os.mkdir('dir_b')
137
else:
138
t.mkdir('dir_b')
139
tester.assertEqual(t.has('dir_b'), True)
140
tester.assert_(os.path.isdir('dir_b'))
141
142
if readonly:
143
tester.assertRaises(TransportNotPossible,
144
t.mkdir_multi, ['dir_c', 'dir_d'])
145
os.mkdir('dir_c')
146
os.mkdir('dir_d')
147
else:
148
t.mkdir_multi(['dir_c', 'dir_d'])
149
150
if readonly:
151
tester.assertRaises(TransportNotPossible,
152
t.mkdir_multi, iter(['dir_e', 'dir_f']))
153
os.mkdir('dir_e')
154
os.mkdir('dir_f')
155
else:
156
t.mkdir_multi(iter(['dir_e', 'dir_f']))
157
tester.assertEqual(list(t.has_multi(
158
['dir_a', 'dir_b', 'dir_c', 'dir_q',
159
'dir_d', 'dir_e', 'dir_f', 'dir_b'])),
160
[True, True, True, False,
161
True, True, True, True])
162
for d in ['dir_a', 'dir_b', 'dir_c', 'dir_d', 'dir_e', 'dir_f']:
163
tester.assert_(os.path.isdir(d))
164
165
if not readonly:
166
tester.assertRaises(NoSuchFile, t.mkdir, 'path/doesnt/exist')
167
tester.assertRaises(FileExists, t.mkdir, 'dir_a') # Creating a directory again should fail
168
169
# Make sure the transport recognizes when a
170
# directory is created by other means
171
# Caching Transports will fail, because dir_e was already seen not
172
# to exist. So instead, we will search for a new directory
173
#os.mkdir('dir_e')
174
#if not readonly:
175
# tester.assertRaises(FileExists, t.mkdir, 'dir_e')
176
177
os.mkdir('dir_g')
178
if not readonly:
179
tester.assertRaises(FileExists, t.mkdir, 'dir_g')
180
181
# Test get/put in sub-directories
182
if readonly:
183
open('dir_a/a', 'wb').write('contents of dir_a/a')
184
open('dir_b/b', 'wb').write('contents of dir_b/b')
185
else:
186
tester.assertEqual(
187
t.put_multi([('dir_a/a', 'contents of dir_a/a'),
188
('dir_b/b', 'contents of dir_b/b')])
189
, 2)
190
for f in ('dir_a/a', 'dir_b/b'):
191
tester.assertEqual(t.get(f).read(), open(f).read())
192
193
# Test copy_to
194
dtmp = tempfile.mkdtemp(dir='.', prefix='test-transport-')
195
dtmp_base = os.path.basename(dtmp)
196
local_t = LocalTransport(dtmp)
197
198
files = ['a', 'b', 'c', 'd']
199
t.copy_to(files, local_t)
200
for f in files:
201
tester.assertEquals(open(f).read(),
202
open(os.path.join(dtmp_base, f)).read())
203
204
del dtmp, dtmp_base, local_t
205
206
dtmp = tempfile.mkdtemp(dir='.', prefix='test-transport-')
207
dtmp_base = os.path.basename(dtmp)
208
local_t = LocalTransport(dtmp)
209
210
files = ['a', 'b', 'c', 'd']
211
t.copy_to(iter(files), local_t)
212
for f in files:
213
tester.assertEquals(open(f).read(),
214
open(os.path.join(dtmp_base, f)).read())
215
216
del dtmp, dtmp_base, local_t
217
218
# Test append, and append_multi
219
if readonly:
220
tester.assertRaises(TransportNotPossible,
221
t.append, 'a', 'add\nsome\nmore\ncontents\n')
222
_append('a', 'add\nsome\nmore\ncontents\n')
223
else:
224
t.append('a', 'add\nsome\nmore\ncontents\n')
225
226
tester.check_file_contents('a',
227
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n')
228
229
if readonly:
230
tester.assertRaises(TransportNotPossible,
231
t.append_multi,
232
[('a', 'and\nthen\nsome\nmore\n'),
233
('d', 'some\nmore\nfor\nd\n')])
234
_append('a', 'and\nthen\nsome\nmore\n')
235
_append('d', 'some\nmore\nfor\nd\n')
236
else:
237
t.append_multi([('a', 'and\nthen\nsome\nmore\n'),
238
('d', 'some\nmore\nfor\nd\n')])
239
tester.check_file_contents('a',
240
'diff\ncontents for\na\n'
241
'add\nsome\nmore\ncontents\n'
242
'and\nthen\nsome\nmore\n')
243
tester.check_file_contents('d',
244
'another contents\nfor d\n'
245
'some\nmore\nfor\nd\n')
246
247
if readonly:
248
_append('a', 'a little bit more\n')
249
_append('d', 'from an iterator\n')
250
else:
251
t.append_multi(iter([('a', 'a little bit more\n'),
252
('d', 'from an iterator\n')]))
253
tester.check_file_contents('a',
254
'diff\ncontents for\na\n'
255
'add\nsome\nmore\ncontents\n'
256
'and\nthen\nsome\nmore\n'
257
'a little bit more\n')
258
tester.check_file_contents('d',
259
'another contents\nfor d\n'
260
'some\nmore\nfor\nd\n'
261
'from an iterator\n')
262
263
# Test that StringIO can be used as a file-like object with put
264
f1 = StringIO('this is a string\nand some more stuff\n')
265
if readonly:
266
open('f1', 'wb').write(f1.read())
267
else:
268
t.put('f1', f1)
269
270
del f1
271
272
tester.check_file_contents('f1',
273
'this is a string\nand some more stuff\n')
274
275
f2 = StringIO('here is some text\nand a bit more\n')
276
f3 = StringIO('some text for the\nthird file created\n')
277
278
if readonly:
279
open('f2', 'wb').write(f2.read())
280
open('f3', 'wb').write(f3.read())
281
else:
282
t.put_multi([('f2', f2), ('f3', f3)])
283
284
del f2, f3
285
286
tester.check_file_contents('f2', 'here is some text\nand a bit more\n')
287
tester.check_file_contents('f3', 'some text for the\nthird file created\n')
288
289
# Test that an actual file object can be used with put
290
f4 = open('f1', 'rb')
291
if readonly:
292
open('f4', 'wb').write(f4.read())
293
else:
294
t.put('f4', f4)
295
296
del f4
297
298
tester.check_file_contents('f4',
299
'this is a string\nand some more stuff\n')
300
301
f5 = open('f2', 'rb')
302
f6 = open('f3', 'rb')
303
if readonly:
304
open('f5', 'wb').write(f5.read())
305
open('f6', 'wb').write(f6.read())
306
else:
307
t.put_multi([('f5', f5), ('f6', f6)])
308
309
del f5, f6
310
311
tester.check_file_contents('f5', 'here is some text\nand a bit more\n')
312
tester.check_file_contents('f6', 'some text for the\nthird file created\n')
313
314
# Test that StringIO can be used as a file-like object with append
315
a1 = StringIO('appending to\none\n')
316
if readonly:
317
_append('f1', a1.read())
318
else:
319
t.append('f1', a1)
320
321
del a1
322
323
tester.check_file_contents('f1',
324
'this is a string\nand some more stuff\n'
325
'appending to\none\n')
326
327
a2 = StringIO('adding more\ntext to two\n')
328
a3 = StringIO('some garbage\nto put in three\n')
329
330
if readonly:
331
_append('f2', a2.read())
332
_append('f3', a3.read())
333
else:
334
t.append_multi([('f2', a2), ('f3', a3)])
335
336
del a2, a3
337
338
tester.check_file_contents('f2',
339
'here is some text\nand a bit more\n'
340
'adding more\ntext to two\n')
341
tester.check_file_contents('f3',
342
'some text for the\nthird file created\n'
343
'some garbage\nto put in three\n')
344
345
# Test that an actual file object can be used with put
346
a4 = open('f1', 'rb')
347
if readonly:
348
_append('f4', a4.read())
349
else:
350
t.append('f4', a4)
351
352
del a4
353
354
tester.check_file_contents('f4',
355
'this is a string\nand some more stuff\n'
356
'this is a string\nand some more stuff\n'
357
'appending to\none\n')
358
359
a5 = open('f2', 'rb')
360
a6 = open('f3', 'rb')
361
if readonly:
362
_append('f5', a5.read())
363
_append('f6', a6.read())
364
else:
365
t.append_multi([('f5', a5), ('f6', a6)])
366
367
del a5, a6
368
369
tester.check_file_contents('f5',
370
'here is some text\nand a bit more\n'
371
'here is some text\nand a bit more\n'
372
'adding more\ntext to two\n')
373
tester.check_file_contents('f6',
374
'some text for the\nthird file created\n'
375
'some text for the\nthird file created\n'
376
'some garbage\nto put in three\n')
377
378
# Test get_partial
379
tester.assertRaises(NoSuchFile,
380
t.get_partial, 'a-missing-file', 20)
381
tester.assertRaises(NoSuchFile,
382
t.get_partial, 'another-missing-file', 20, 30)
383
f = t.get_partial('f5', 33)
384
tester.assertEqual(f.read(),
385
'here is some text\nand a bit more\n'
386
'adding more\ntext to two\n')
387
f = t.get_partial('f5', 66)
388
tester.assertEqual(f.read(),
389
'adding more\ntext to two\n')
390
f = t.get_partial('f5', 66, 10)
391
tester.assertEqual(f.read(10),
392
'adding mor')
393
394
del f
395
396
offsets = [('f5', 33), ('f6', 20, 10), ('f4', 10, 20)]
397
values = ['here is some text\nand a bit more\nadding more\ntext to two\n',
398
'ird file c',
399
'string\nand some more'
400
]
401
contents_f = t.get_partial_multi(offsets)
402
count = 0
403
for f, val in zip(contents_f, values):
404
count += 1
405
tester.assertEqual(val, f.read(len(val)))
406
# Make sure we saw all values, and no extra
407
tester.assertEqual(len(values), count)
408
tester.assertEqual(list(contents_f), [])
409
410
# Do the same thing with an iterator
411
offsets = iter([('f5', 34), ('f6', 18, 10), ('f4', 15, 15)])
412
values = ['ere is some text\nand a bit more\nadding more\ntext to two\n',
413
'third file',
414
'g\nand some more'
415
]
416
contents_f = t.get_partial_multi(offsets)
417
count = 0
418
for f, val in zip(contents_f, values):
419
count += 1
420
tester.assertEqual(val, f.read(len(val)))
421
tester.assertEqual(len(values), count)
422
tester.assertEqual(list(contents_f), [])
423
424
# TODO: Test delete, move, etc.
425
# TODO: Test locking
426
427
class LocalTransportTest(TestCaseInTempDir):
428
def test_local_transport(self):
429
from bzrlib.transport.local import LocalTransport
430
431
t = LocalTransport('.')
432
test_transport(self, t)
433
434
class HttpTransportTest(TestCaseWithWebserver):
435
def test_http_transport(self):
42
readonly = False
43
def get_transport(self):
44
"""Children should override this to return the Transport object.
45
"""
46
raise NotImplementedError
47
48
def test_has(self):
49
t = self.get_transport()
50
51
files = ['a', 'b', 'e', 'g']
52
self.build_tree(files)
53
self.assertEqual(t.has('a'), True)
54
self.assertEqual(t.has('c'), False)
55
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
56
[True, True, False, False, True, False, True, False])
57
self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
58
[True, True, False, False, True, False, True, False])
59
60
def test_get(self):
61
t = self.get_transport()
62
63
files = ['a', 'b', 'e', 'g']
64
self.build_tree(files)
65
self.assertEqual(t.get('a').read(), open('a').read())
66
content_f = t.get_multi(files)
67
for path,f in zip(files, content_f):
68
self.assertEqual(open(path).read(), f.read())
69
70
content_f = t.get_multi(iter(files))
71
for path,f in zip(files, content_f):
72
self.assertEqual(open(path).read(), f.read())
73
74
self.assertRaises(NoSuchFile, t.get, 'c')
75
try:
76
files = list(t.get_multi(['a', 'b', 'c']))
77
except NoSuchFile:
78
pass
79
else:
80
self.fail('Failed to raise NoSuchFile for missing file in get_multi')
81
try:
82
files = list(t.get_multi(iter(['a', 'b', 'c', 'e'])))
83
except NoSuchFile:
84
pass
85
else:
86
self.fail('Failed to raise NoSuchFile for missing file in get_multi')
87
88
def test_put(self):
89
t = self.get_transport()
90
91
if self.readonly:
92
self.assertRaises(TransportNotPossible,
93
t.put, 'a', 'some text for a\n')
94
open('a', 'wb').write('some text for a\n')
95
else:
96
t.put('a', 'some text for a\n')
97
self.assert_(os.path.exists('a'))
98
self.check_file_contents('a', 'some text for a\n')
99
self.assertEqual(t.get('a').read(), 'some text for a\n')
100
# Make sure 'has' is updated
101
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
102
[True, False, False, False, False])
103
if self.readonly:
104
self.assertRaises(TransportNotPossible,
105
t.put_multi,
106
[('a', 'new\ncontents for\na\n'),
107
('d', 'contents\nfor d\n')])
108
open('a', 'wb').write('new\ncontents for\na\n')
109
open('d', 'wb').write('contents\nfor d\n')
110
else:
111
# Put also replaces contents
112
self.assertEqual(t.put_multi([('a', 'new\ncontents for\na\n'),
113
('d', 'contents\nfor d\n')]),
114
2)
115
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
116
[True, False, False, True, False])
117
self.check_file_contents('a', 'new\ncontents for\na\n')
118
self.check_file_contents('d', 'contents\nfor d\n')
119
120
if self.readonly:
121
self.assertRaises(TransportNotPossible,
122
t.put_multi, iter([('a', 'diff\ncontents for\na\n'),
123
('d', 'another contents\nfor d\n')]))
124
open('a', 'wb').write('diff\ncontents for\na\n')
125
open('d', 'wb').write('another contents\nfor d\n')
126
else:
127
self.assertEqual(
128
t.put_multi(iter([('a', 'diff\ncontents for\na\n'),
129
('d', 'another contents\nfor d\n')]))
130
, 2)
131
self.check_file_contents('a', 'diff\ncontents for\na\n')
132
self.check_file_contents('d', 'another contents\nfor d\n')
133
134
if self.readonly:
135
self.assertRaises(TransportNotPossible,
136
t.put, 'path/doesnt/exist/c', 'contents')
137
else:
138
self.assertRaises(NoSuchFile,
139
t.put, 'path/doesnt/exist/c', 'contents')
140
141
def test_put_file(self):
142
t = self.get_transport()
143
144
# Test that StringIO can be used as a file-like object with put
145
f1 = StringIO('this is a string\nand some more stuff\n')
146
if self.readonly:
147
open('f1', 'wb').write(f1.read())
148
else:
149
t.put('f1', f1)
150
151
del f1
152
153
self.check_file_contents('f1',
154
'this is a string\nand some more stuff\n')
155
156
f2 = StringIO('here is some text\nand a bit more\n')
157
f3 = StringIO('some text for the\nthird file created\n')
158
159
if self.readonly:
160
open('f2', 'wb').write(f2.read())
161
open('f3', 'wb').write(f3.read())
162
else:
163
t.put_multi([('f2', f2), ('f3', f3)])
164
165
del f2, f3
166
167
self.check_file_contents('f2', 'here is some text\nand a bit more\n')
168
self.check_file_contents('f3', 'some text for the\nthird file created\n')
169
170
# Test that an actual file object can be used with put
171
f4 = open('f1', 'rb')
172
if self.readonly:
173
open('f4', 'wb').write(f4.read())
174
else:
175
t.put('f4', f4)
176
177
del f4
178
179
self.check_file_contents('f4',
180
'this is a string\nand some more stuff\n')
181
182
f5 = open('f2', 'rb')
183
f6 = open('f3', 'rb')
184
if self.readonly:
185
open('f5', 'wb').write(f5.read())
186
open('f6', 'wb').write(f6.read())
187
else:
188
t.put_multi([('f5', f5), ('f6', f6)])
189
190
del f5, f6
191
192
self.check_file_contents('f5', 'here is some text\nand a bit more\n')
193
self.check_file_contents('f6', 'some text for the\nthird file created\n')
194
195
196
197
def test_mkdir(self):
198
t = self.get_transport()
199
200
# Test mkdir
201
os.mkdir('dir_a')
202
self.assertEqual(t.has('dir_a'), True)
203
self.assertEqual(t.has('dir_b'), False)
204
205
if self.readonly:
206
self.assertRaises(TransportNotPossible,
207
t.mkdir, 'dir_b')
208
os.mkdir('dir_b')
209
else:
210
t.mkdir('dir_b')
211
self.assertEqual(t.has('dir_b'), True)
212
self.assert_(os.path.isdir('dir_b'))
213
214
if self.readonly:
215
self.assertRaises(TransportNotPossible,
216
t.mkdir_multi, ['dir_c', 'dir_d'])
217
os.mkdir('dir_c')
218
os.mkdir('dir_d')
219
else:
220
t.mkdir_multi(['dir_c', 'dir_d'])
221
222
if self.readonly:
223
self.assertRaises(TransportNotPossible,
224
t.mkdir_multi, iter(['dir_e', 'dir_f']))
225
os.mkdir('dir_e')
226
os.mkdir('dir_f')
227
else:
228
t.mkdir_multi(iter(['dir_e', 'dir_f']))
229
self.assertEqual(list(t.has_multi(
230
['dir_a', 'dir_b', 'dir_c', 'dir_q',
231
'dir_d', 'dir_e', 'dir_f', 'dir_b'])),
232
[True, True, True, False,
233
True, True, True, True])
234
for d in ['dir_a', 'dir_b', 'dir_c', 'dir_d', 'dir_e', 'dir_f']:
235
self.assert_(os.path.isdir(d))
236
237
if not self.readonly:
238
self.assertRaises(NoSuchFile, t.mkdir, 'path/doesnt/exist')
239
self.assertRaises(FileExists, t.mkdir, 'dir_a') # Creating a directory again should fail
240
241
# Make sure the transport recognizes when a
242
# directory is created by other means
243
# Caching Transports will fail, because dir_e was already seen not
244
# to exist. So instead, we will search for a new directory
245
#os.mkdir('dir_e')
246
#if not self.readonly:
247
# self.assertRaises(FileExists, t.mkdir, 'dir_e')
248
249
os.mkdir('dir_g')
250
if not self.readonly:
251
self.assertRaises(FileExists, t.mkdir, 'dir_g')
252
253
# Test get/put in sub-directories
254
if self.readonly:
255
open('dir_a/a', 'wb').write('contents of dir_a/a')
256
open('dir_b/b', 'wb').write('contents of dir_b/b')
257
else:
258
self.assertEqual(
259
t.put_multi([('dir_a/a', 'contents of dir_a/a'),
260
('dir_b/b', 'contents of dir_b/b')])
261
, 2)
262
for f in ('dir_a/a', 'dir_b/b'):
263
self.assertEqual(t.get(f).read(), open(f).read())
264
265
def test_copy_to(self):
266
import tempfile
267
from bzrlib.transport.local import LocalTransport
268
269
t = self.get_transport()
270
271
files = ['a', 'b', 'c', 'd']
272
self.build_tree(files)
273
274
dtmp = tempfile.mkdtemp(dir='.', prefix='test-transport-')
275
dtmp_base = os.path.basename(dtmp)
276
local_t = LocalTransport(dtmp)
277
278
t.copy_to(files, local_t)
279
for f in files:
280
self.assertEquals(open(f).read(),
281
open(os.path.join(dtmp_base, f)).read())
282
283
del dtmp, dtmp_base, local_t
284
285
dtmp = tempfile.mkdtemp(dir='.', prefix='test-transport-')
286
dtmp_base = os.path.basename(dtmp)
287
local_t = LocalTransport(dtmp)
288
289
files = ['a', 'b', 'c', 'd']
290
t.copy_to(iter(files), local_t)
291
for f in files:
292
self.assertEquals(open(f).read(),
293
open(os.path.join(dtmp_base, f)).read())
294
295
del dtmp, dtmp_base, local_t
296
297
def test_append(self):
298
t = self.get_transport()
299
300
if self.readonly:
301
open('a', 'wb').write('diff\ncontents for\na\n')
302
open('b', 'wb').write('contents\nfor b\n')
303
else:
304
t.put_multi([
305
('a', 'diff\ncontents for\na\n'),
306
('b', 'contents\nfor b\n')
307
])
308
309
if self.readonly:
310
self.assertRaises(TransportNotPossible,
311
t.append, 'a', 'add\nsome\nmore\ncontents\n')
312
_append('a', 'add\nsome\nmore\ncontents\n')
313
else:
314
t.append('a', 'add\nsome\nmore\ncontents\n')
315
316
self.check_file_contents('a',
317
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n')
318
319
if self.readonly:
320
self.assertRaises(TransportNotPossible,
321
t.append_multi,
322
[('a', 'and\nthen\nsome\nmore\n'),
323
('b', 'some\nmore\nfor\nb\n')])
324
_append('a', 'and\nthen\nsome\nmore\n')
325
_append('b', 'some\nmore\nfor\nb\n')
326
else:
327
t.append_multi([('a', 'and\nthen\nsome\nmore\n'),
328
('b', 'some\nmore\nfor\nb\n')])
329
self.check_file_contents('a',
330
'diff\ncontents for\na\n'
331
'add\nsome\nmore\ncontents\n'
332
'and\nthen\nsome\nmore\n')
333
self.check_file_contents('b',
334
'contents\nfor b\n'
335
'some\nmore\nfor\nb\n')
336
337
if self.readonly:
338
_append('a', 'a little bit more\n')
339
_append('b', 'from an iterator\n')
340
else:
341
t.append_multi(iter([('a', 'a little bit more\n'),
342
('b', 'from an iterator\n')]))
343
self.check_file_contents('a',
344
'diff\ncontents for\na\n'
345
'add\nsome\nmore\ncontents\n'
346
'and\nthen\nsome\nmore\n'
347
'a little bit more\n')
348
self.check_file_contents('b',
349
'contents\nfor b\n'
350
'some\nmore\nfor\nb\n'
351
'from an iterator\n')
352
353
def test_append_file(self):
354
t = self.get_transport()
355
356
contents = [
357
('f1', 'this is a string\nand some more stuff\n'),
358
('f2', 'here is some text\nand a bit more\n'),
359
('f3', 'some text for the\nthird file created\n'),
360
('f4', 'this is a string\nand some more stuff\n'),
361
('f5', 'here is some text\nand a bit more\n'),
362
('f6', 'some text for the\nthird file created\n')
363
]
364
365
if self.readonly:
366
for f, val in contents:
367
open(f, 'wb').write(val)
368
else:
369
t.put_multi(contents)
370
371
a1 = StringIO('appending to\none\n')
372
if self.readonly:
373
_append('f1', a1.read())
374
else:
375
t.append('f1', a1)
376
377
del a1
378
379
self.check_file_contents('f1',
380
'this is a string\nand some more stuff\n'
381
'appending to\none\n')
382
383
a2 = StringIO('adding more\ntext to two\n')
384
a3 = StringIO('some garbage\nto put in three\n')
385
386
if self.readonly:
387
_append('f2', a2.read())
388
_append('f3', a3.read())
389
else:
390
t.append_multi([('f2', a2), ('f3', a3)])
391
392
del a2, a3
393
394
self.check_file_contents('f2',
395
'here is some text\nand a bit more\n'
396
'adding more\ntext to two\n')
397
self.check_file_contents('f3',
398
'some text for the\nthird file created\n'
399
'some garbage\nto put in three\n')
400
401
# Test that an actual file object can be used with put
402
a4 = open('f1', 'rb')
403
if self.readonly:
404
_append('f4', a4.read())
405
else:
406
t.append('f4', a4)
407
408
del a4
409
410
self.check_file_contents('f4',
411
'this is a string\nand some more stuff\n'
412
'this is a string\nand some more stuff\n'
413
'appending to\none\n')
414
415
a5 = open('f2', 'rb')
416
a6 = open('f3', 'rb')
417
if self.readonly:
418
_append('f5', a5.read())
419
_append('f6', a6.read())
420
else:
421
t.append_multi([('f5', a5), ('f6', a6)])
422
423
del a5, a6
424
425
self.check_file_contents('f5',
426
'here is some text\nand a bit more\n'
427
'here is some text\nand a bit more\n'
428
'adding more\ntext to two\n')
429
self.check_file_contents('f6',
430
'some text for the\nthird file created\n'
431
'some text for the\nthird file created\n'
432
'some garbage\nto put in three\n')
433
434
def test_get_partial(self):
435
t = self.get_transport()
436
437
contents = [
438
('f1',
439
'here is some text\nand a bit more\n'
440
'adding more\ntext to two\n'),
441
('f2',
442
'this is a string\nand some more stuff\n'
443
'appending to\none\n'),
444
('f3',
445
'some text for the\nthird file created\n'
446
'some garbage\nto put in three\n')
447
]
448
if self.readonly:
449
for f, val in contents:
450
open(f, 'wb').write(val)
451
else:
452
t.put_multi(contents)
453
454
self.assertRaises(NoSuchFile,
455
t.get_partial, 'a-missing-file', 20)
456
self.assertRaises(NoSuchFile,
457
t.get_partial, 'another-missing-file', 20, 30)
458
f = t.get_partial('f1', 33)
459
self.assertEqual(f.read(),
460
'adding more\ntext to two\n')
461
f = t.get_partial('f1', 33, 10)
462
self.assertEqual(f.read(10),
463
'adding mor')
464
465
del f
466
467
offsets = [('f2', 37), ('f3', 20, 10), ('f1', 10, 20)]
468
values = ['appending to\none\n',
469
'ird file c',
470
'me text\nand a bit mo'
471
]
472
contents_f = t.get_partial_multi(offsets)
473
count = 0
474
for f, val in zip(contents_f, values):
475
count += 1
476
self.assertEqual(val, f.read(len(val)))
477
# Make sure we saw all values, and no extra
478
self.assertEqual(len(values), count)
479
self.assertEqual(list(contents_f), [])
480
481
# Do the same thing with an iterator
482
offsets = iter([('f2', 34), ('f3', 18, 10), ('f1', 15, 15)])
483
values = ['ff\nappending to\none\n',
484
'third file',
485
'xt\nand a bit mo'
486
]
487
contents_f = t.get_partial_multi(offsets)
488
count = 0
489
for f, val in zip(contents_f, values):
490
count += 1
491
self.assertEqual(val, f.read(len(val)))
492
self.assertEqual(len(values), count)
493
self.assertEqual(list(contents_f), [])
494
495
496
def test_delete(self):
497
# TODO: Test Transport.delete
498
pass
499
500
def test_move(self):
501
# TODO: Test Transport.move
502
pass
503
504
class LocalTransportTest(TestCaseInTempDir, TestTransportMixIn):
505
def get_transport(self):
506
from bzrlib.transport.local import LocalTransport
507
return LocalTransport('.')
508
509
class HttpTransportTest(TestCaseWithWebserver, TestTransportMixIn):
510
readonly = True
511
def get_transport(self):
436
512
from bzrlib.transport.http import HttpTransport
437
438
513
url = self.get_remote_url('.')
439
t = HttpTransport(url)
440
test_transport(self, t, readonly=True)
514
return HttpTransport(url)
441
515
Loggerhead is a web-based interface for Breezy
Version: 2.0.1