« back to all changes in this revision
Viewing changes to bzrlib/tests/test_transport_implementations.py
- browse files at revision 1988
- compare with another revision
- compare with revision 81
- download diff
- download tarball
- view history from revision 1988
- Committer: Canonical.com Patch Queue Manager
- Date: 2006-09-06 16:21:16 UTC
- mfrom: (1955.3.30 transport_bytes)
- Revision ID: pqm@pqm.ubuntu.com-20060906162116-90b02cf97bcc11e8
(jam) create Transport.*_{file,bytes}
- files modified:
- NEWS
added
removed
bzrlib/tests/test_transport_implementations.py
34
34
TransportNotPossible, ConnectionError,
35
35
InvalidURL)
36
36
from bzrlib.osutils import getcwd
37
from bzrlib.symbol_versioning import zero_eleven
37
38
from bzrlib.tests import TestCaseInTempDir, TestSkipped
38
39
from bzrlib.tests.test_transport import TestTransportImplementation
39
40
from bzrlib.transport import memory
111
112
self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
112
113
self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
113
114
115
def test_get_bytes(self):
116
t = self.get_transport()
117
118
files = ['a', 'b', 'e', 'g']
119
contents = ['contents of a\n',
120
'contents of b\n',
121
'contents of e\n',
122
'contents of g\n',
123
]
124
self.build_tree(files, transport=t, line_endings='binary')
125
self.check_transport_contents('contents of a\n', t, 'a')
126
127
for content, fname in zip(contents, files):
128
self.assertEqual(content, t.get_bytes(fname))
129
130
self.assertRaises(NoSuchFile, t.get_bytes, 'c')
131
114
132
def test_put(self):
115
133
t = self.get_transport()
116
134
117
135
if t.is_readonly():
118
self.assertRaises(TransportNotPossible,
119
t.put, 'a', 'some text for a\n')
120
return
121
122
t.put('a', StringIO('some text for a\n'))
123
self.failUnless(t.has('a'))
124
self.check_transport_contents('some text for a\n', t, 'a')
125
# Make sure 'has' is updated
126
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
127
[True, False, False, False, False])
128
# Put also replaces contents
129
self.assertEqual(t.put_multi([('a', StringIO('new\ncontents for\na\n')),
130
('d', StringIO('contents\nfor d\n'))]),
131
2)
132
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
133
[True, False, False, True, False])
136
return
137
138
self.applyDeprecated(zero_eleven, t.put, 'a', 'string\ncontents\n')
139
self.check_transport_contents('string\ncontents\n', t, 'a')
140
141
self.applyDeprecated(zero_eleven,
142
t.put, 'b', StringIO('file-like\ncontents\n'))
143
self.check_transport_contents('file-like\ncontents\n', t, 'b')
144
145
def test_put_bytes(self):
146
t = self.get_transport()
147
148
if t.is_readonly():
149
self.assertRaises(TransportNotPossible,
150
t.put_bytes, 'a', 'some text for a\n')
151
return
152
153
t.put_bytes('a', 'some text for a\n')
154
self.failUnless(t.has('a'))
155
self.check_transport_contents('some text for a\n', t, 'a')
156
157
# The contents should be overwritten
158
t.put_bytes('a', 'new text for a\n')
159
self.check_transport_contents('new text for a\n', t, 'a')
160
161
self.assertRaises(NoSuchFile,
162
t.put_bytes, 'path/doesnt/exist/c', 'contents')
163
164
def test_put_bytes_non_atomic(self):
165
t = self.get_transport()
166
167
if t.is_readonly():
168
self.assertRaises(TransportNotPossible,
169
t.put_bytes_non_atomic, 'a', 'some text for a\n')
170
return
171
172
self.failIf(t.has('a'))
173
t.put_bytes_non_atomic('a', 'some text for a\n')
174
self.failUnless(t.has('a'))
175
self.check_transport_contents('some text for a\n', t, 'a')
176
# Put also replaces contents
177
t.put_bytes_non_atomic('a', 'new\ncontents for\na\n')
178
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
179
180
# Make sure we can create another file
181
t.put_bytes_non_atomic('d', 'contents for\nd\n')
182
# And overwrite 'a' with empty contents
183
t.put_bytes_non_atomic('a', '')
184
self.check_transport_contents('contents for\nd\n', t, 'd')
185
self.check_transport_contents('', t, 'a')
186
187
self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'no/such/path',
188
'contents\n')
189
# Now test the create_parent flag
190
self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'dir/a',
191
'contents\n')
192
self.failIf(t.has('dir/a'))
193
t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',
194
create_parent_dir=True)
195
self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
196
197
# But we still get NoSuchFile if we can't make the parent dir
198
self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'not/there/a',
199
'contents\n',
200
create_parent_dir=True)
201
202
def test_put_bytes_permissions(self):
203
t = self.get_transport()
204
205
if t.is_readonly():
206
return
207
if not t._can_roundtrip_unix_modebits():
208
# Can't roundtrip, so no need to run this test
209
return
210
t.put_bytes('mode644', 'test text\n', mode=0644)
211
self.assertTransportMode(t, 'mode644', 0644)
212
t.put_bytes('mode666', 'test text\n', mode=0666)
213
self.assertTransportMode(t, 'mode666', 0666)
214
t.put_bytes('mode600', 'test text\n', mode=0600)
215
self.assertTransportMode(t, 'mode600', 0600)
216
# Yes, you can put_bytes a file such that it becomes readonly
217
t.put_bytes('mode400', 'test text\n', mode=0400)
218
self.assertTransportMode(t, 'mode400', 0400)
219
220
# The default permissions should be based on the current umask
221
umask = osutils.get_umask()
222
t.put_bytes('nomode', 'test text\n', mode=None)
223
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
224
225
def test_put_bytes_non_atomic_permissions(self):
226
t = self.get_transport()
227
228
if t.is_readonly():
229
return
230
if not t._can_roundtrip_unix_modebits():
231
# Can't roundtrip, so no need to run this test
232
return
233
t.put_bytes_non_atomic('mode644', 'test text\n', mode=0644)
234
self.assertTransportMode(t, 'mode644', 0644)
235
t.put_bytes_non_atomic('mode666', 'test text\n', mode=0666)
236
self.assertTransportMode(t, 'mode666', 0666)
237
t.put_bytes_non_atomic('mode600', 'test text\n', mode=0600)
238
self.assertTransportMode(t, 'mode600', 0600)
239
t.put_bytes_non_atomic('mode400', 'test text\n', mode=0400)
240
self.assertTransportMode(t, 'mode400', 0400)
241
242
# The default permissions should be based on the current umask
243
umask = osutils.get_umask()
244
t.put_bytes_non_atomic('nomode', 'test text\n', mode=None)
245
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
246
247
def test_put_file(self):
248
t = self.get_transport()
249
250
if t.is_readonly():
251
self.assertRaises(TransportNotPossible,
252
t.put_file, 'a', StringIO('some text for a\n'))
253
return
254
255
t.put_file('a', StringIO('some text for a\n'))
256
self.failUnless(t.has('a'))
257
self.check_transport_contents('some text for a\n', t, 'a')
258
# Put also replaces contents
259
t.put_file('a', StringIO('new\ncontents for\na\n'))
260
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
261
self.assertRaises(NoSuchFile,
262
t.put_file, 'path/doesnt/exist/c',
263
StringIO('contents'))
264
265
def test_put_file_non_atomic(self):
266
t = self.get_transport()
267
268
if t.is_readonly():
269
self.assertRaises(TransportNotPossible,
270
t.put_file_non_atomic, 'a', StringIO('some text for a\n'))
271
return
272
273
self.failIf(t.has('a'))
274
t.put_file_non_atomic('a', StringIO('some text for a\n'))
275
self.failUnless(t.has('a'))
276
self.check_transport_contents('some text for a\n', t, 'a')
277
# Put also replaces contents
278
t.put_file_non_atomic('a', StringIO('new\ncontents for\na\n'))
279
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
280
281
# Make sure we can create another file
282
t.put_file_non_atomic('d', StringIO('contents for\nd\n'))
283
# And overwrite 'a' with empty contents
284
t.put_file_non_atomic('a', StringIO(''))
285
self.check_transport_contents('contents for\nd\n', t, 'd')
286
self.check_transport_contents('', t, 'a')
287
288
self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'no/such/path',
289
StringIO('contents\n'))
290
# Now test the create_parent flag
291
self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'dir/a',
292
StringIO('contents\n'))
293
self.failIf(t.has('dir/a'))
294
t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
295
create_parent_dir=True)
296
self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
297
298
# But we still get NoSuchFile if we can't make the parent dir
299
self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'not/there/a',
300
StringIO('contents\n'),
301
create_parent_dir=True)
302
303
def test_put_file_permissions(self):
304
305
t = self.get_transport()
306
307
if t.is_readonly():
308
return
309
if not t._can_roundtrip_unix_modebits():
310
# Can't roundtrip, so no need to run this test
311
return
312
t.put_file('mode644', StringIO('test text\n'), mode=0644)
313
self.assertTransportMode(t, 'mode644', 0644)
314
t.put_file('mode666', StringIO('test text\n'), mode=0666)
315
self.assertTransportMode(t, 'mode666', 0666)
316
t.put_file('mode600', StringIO('test text\n'), mode=0600)
317
self.assertTransportMode(t, 'mode600', 0600)
318
# Yes, you can put a file such that it becomes readonly
319
t.put_file('mode400', StringIO('test text\n'), mode=0400)
320
self.assertTransportMode(t, 'mode400', 0400)
321
322
# XXX: put_multi is deprecated, so do we really care anymore?
323
self.applyDeprecated(zero_eleven, t.put_multi,
324
[('mmode644', StringIO('text\n'))], mode=0644)
325
self.assertTransportMode(t, 'mmode644', 0644)
326
327
# The default permissions should be based on the current umask
328
umask = osutils.get_umask()
329
t.put_file('nomode', StringIO('test text\n'), mode=None)
330
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
331
332
def test_put_file_non_atomic_permissions(self):
333
t = self.get_transport()
334
335
if t.is_readonly():
336
return
337
if not t._can_roundtrip_unix_modebits():
338
# Can't roundtrip, so no need to run this test
339
return
340
t.put_file_non_atomic('mode644', StringIO('test text\n'), mode=0644)
341
self.assertTransportMode(t, 'mode644', 0644)
342
t.put_file_non_atomic('mode666', StringIO('test text\n'), mode=0666)
343
self.assertTransportMode(t, 'mode666', 0666)
344
t.put_file_non_atomic('mode600', StringIO('test text\n'), mode=0600)
345
self.assertTransportMode(t, 'mode600', 0600)
346
# Yes, you can put_file_non_atomic a file such that it becomes readonly
347
t.put_file_non_atomic('mode400', StringIO('test text\n'), mode=0400)
348
self.assertTransportMode(t, 'mode400', 0400)
349
350
# The default permissions should be based on the current umask
351
umask = osutils.get_umask()
352
t.put_file_non_atomic('nomode', StringIO('test text\n'), mode=None)
353
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
354
355
def test_put_multi(self):
356
t = self.get_transport()
357
358
if t.is_readonly():
359
return
360
self.assertEqual(2, self.applyDeprecated(zero_eleven,
361
t.put_multi, [('a', StringIO('new\ncontents for\na\n')),
362
('d', StringIO('contents\nfor d\n'))]
363
))
364
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd'])),
365
[True, False, False, True])
134
366
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
135
367
self.check_transport_contents('contents\nfor d\n', t, 'd')
136
368
137
self.assertEqual(
138
t.put_multi(iter([('a', StringIO('diff\ncontents for\na\n')),
139
('d', StringIO('another contents\nfor d\n'))])),
140
2)
369
self.assertEqual(2, self.applyDeprecated(zero_eleven,
370
t.put_multi, iter([('a', StringIO('diff\ncontents for\na\n')),
371
('d', StringIO('another contents\nfor d\n'))])
372
))
141
373
self.check_transport_contents('diff\ncontents for\na\n', t, 'a')
142
374
self.check_transport_contents('another contents\nfor d\n', t, 'd')
143
375
144
self.assertRaises(NoSuchFile,
145
t.put, 'path/doesnt/exist/c', StringIO('contents'))
146
147
def test_put_permissions(self):
148
t = self.get_transport()
149
150
if t.is_readonly():
151
return
152
if not t._can_roundtrip_unix_modebits():
153
# Can't roundtrip, so no need to run this test
154
return
155
t.put('mode644', StringIO('test text\n'), mode=0644)
156
self.assertTransportMode(t, 'mode644', 0644)
157
t.put('mode666', StringIO('test text\n'), mode=0666)
158
self.assertTransportMode(t, 'mode666', 0666)
159
t.put('mode600', StringIO('test text\n'), mode=0600)
160
self.assertTransportMode(t, 'mode600', 0600)
161
# Yes, you can put a file such that it becomes readonly
162
t.put('mode400', StringIO('test text\n'), mode=0400)
163
self.assertTransportMode(t, 'mode400', 0400)
164
t.put_multi([('mmode644', StringIO('text\n'))], mode=0644)
165
self.assertTransportMode(t, 'mmode644', 0644)
166
167
# The default permissions should be based on the current umask
168
umask = osutils.get_umask()
169
t.put('nomode', StringIO('test text\n'), mode=None)
170
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
171
172
376
def test_mkdir(self):
173
377
t = self.get_transport()
174
378
207
411
self.assertRaises(FileExists, t.mkdir, 'dir_g')
208
412
209
413
# Test get/put in sub-directories
210
self.assertEqual(2,
211
t.put_multi([('dir_a/a', StringIO('contents of dir_a/a')),
212
('dir_b/b', StringIO('contents of dir_b/b'))]))
414
t.put_bytes('dir_a/a', 'contents of dir_a/a')
415
t.put_file('dir_b/b', StringIO('contents of dir_b/b'))
213
416
self.check_transport_contents('contents of dir_a/a', t, 'dir_a/a')
214
417
self.check_transport_contents('contents of dir_b/b', t, 'dir_b/b')
215
418
270
473
self.build_tree(['e/', 'e/f'])
271
474
else:
272
475
t.mkdir('e')
273
t.put('e/f', StringIO('contents of e'))
476
t.put_bytes('e/f', 'contents of e')
274
477
self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], temp_transport)
275
478
temp_transport.mkdir('e')
276
479
t.copy_to(['e/f'], temp_transport)
295
498
t = self.get_transport()
296
499
297
500
if t.is_readonly():
298
open('a', 'wb').write('diff\ncontents for\na\n')
299
open('b', 'wb').write('contents\nfor b\n')
300
else:
301
t.put_multi([
302
('a', StringIO('diff\ncontents for\na\n')),
303
('b', StringIO('contents\nfor b\n'))
304
])
305
306
if t.is_readonly():
307
self.assertRaises(TransportNotPossible,
308
t.append, 'a', 'add\nsome\nmore\ncontents\n')
309
_append('a', StringIO('add\nsome\nmore\ncontents\n'))
310
else:
311
self.assertEqual(20,
312
t.append('a', StringIO('add\nsome\nmore\ncontents\n')))
313
314
self.check_transport_contents(
315
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
316
t, 'a')
317
318
if t.is_readonly():
319
self.assertRaises(TransportNotPossible,
320
t.append_multi,
321
[('a', 'and\nthen\nsome\nmore\n'),
322
('b', 'some\nmore\nfor\nb\n')])
323
_append('a', StringIO('and\nthen\nsome\nmore\n'))
324
_append('b', StringIO('some\nmore\nfor\nb\n'))
325
else:
326
self.assertEqual((43, 15),
327
t.append_multi([('a', StringIO('and\nthen\nsome\nmore\n')),
328
('b', StringIO('some\nmore\nfor\nb\n'))]))
501
return
502
t.put_bytes('a', 'diff\ncontents for\na\n')
503
t.put_bytes('b', 'contents\nfor b\n')
504
505
self.assertEqual(20, self.applyDeprecated(zero_eleven,
506
t.append, 'a', StringIO('add\nsome\nmore\ncontents\n')))
507
508
self.check_transport_contents(
509
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
510
t, 'a')
511
512
# And we can create new files, too
513
self.assertEqual(0, self.applyDeprecated(zero_eleven,
514
t.append, 'c', StringIO('some text\nfor a missing file\n')))
515
self.check_transport_contents('some text\nfor a missing file\n',
516
t, 'c')
517
def test_append_file(self):
518
t = self.get_transport()
519
520
if t.is_readonly():
521
self.assertRaises(TransportNotPossible,
522
t.append_file, 'a', 'add\nsome\nmore\ncontents\n')
523
return
524
t.put_bytes('a', 'diff\ncontents for\na\n')
525
t.put_bytes('b', 'contents\nfor b\n')
526
527
self.assertEqual(20,
528
t.append_file('a', StringIO('add\nsome\nmore\ncontents\n')))
529
530
self.check_transport_contents(
531
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
532
t, 'a')
533
534
# a file with no parent should fail..
535
self.assertRaises(NoSuchFile,
536
t.append_file, 'missing/path', StringIO('content'))
537
538
# And we can create new files, too
539
self.assertEqual(0,
540
t.append_file('c', StringIO('some text\nfor a missing file\n')))
541
self.check_transport_contents('some text\nfor a missing file\n',
542
t, 'c')
543
544
def test_append_bytes(self):
545
t = self.get_transport()
546
547
if t.is_readonly():
548
self.assertRaises(TransportNotPossible,
549
t.append_bytes, 'a', 'add\nsome\nmore\ncontents\n')
550
return
551
552
self.assertEqual(0, t.append_bytes('a', 'diff\ncontents for\na\n'))
553
self.assertEqual(0, t.append_bytes('b', 'contents\nfor b\n'))
554
555
self.assertEqual(20,
556
t.append_bytes('a', 'add\nsome\nmore\ncontents\n'))
557
558
self.check_transport_contents(
559
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
560
t, 'a')
561
562
# a file with no parent should fail..
563
self.assertRaises(NoSuchFile,
564
t.append_bytes, 'missing/path', 'content')
565
566
def test_append_multi(self):
567
t = self.get_transport()
568
569
if t.is_readonly():
570
return
571
t.put_bytes('a', 'diff\ncontents for\na\n'
572
'add\nsome\nmore\ncontents\n')
573
t.put_bytes('b', 'contents\nfor b\n')
574
575
self.assertEqual((43, 15),
576
t.append_multi([('a', StringIO('and\nthen\nsome\nmore\n')),
577
('b', StringIO('some\nmore\nfor\nb\n'))]))
578
329
579
self.check_transport_contents(
330
580
'diff\ncontents for\na\n'
331
581
'add\nsome\nmore\ncontents\n'
336
586
'some\nmore\nfor\nb\n',
337
587
t, 'b')
338
588
339
if t.is_readonly():
340
_append('a', StringIO('a little bit more\n'))
341
_append('b', StringIO('from an iterator\n'))
342
else:
343
self.assertEqual((62, 31),
344
t.append_multi(iter([('a', StringIO('a little bit more\n')),
345
('b', StringIO('from an iterator\n'))])))
589
self.assertEqual((62, 31),
590
t.append_multi(iter([('a', StringIO('a little bit more\n')),
591
('b', StringIO('from an iterator\n'))])))
346
592
self.check_transport_contents(
347
593
'diff\ncontents for\na\n'
348
594
'add\nsome\nmore\ncontents\n'
355
601
'from an iterator\n',
356
602
t, 'b')
357
603
358
if t.is_readonly():
359
_append('c', StringIO('some text\nfor a missing file\n'))
360
_append('a', StringIO('some text in a\n'))
361
_append('d', StringIO('missing file r\n'))
362
else:
363
self.assertEqual(0,
364
t.append('c', StringIO('some text\nfor a missing file\n')))
365
self.assertEqual((80, 0),
366
t.append_multi([('a', StringIO('some text in a\n')),
367
('d', StringIO('missing file r\n'))]))
604
self.assertEqual((80, 0),
605
t.append_multi([('a', StringIO('some text in a\n')),
606
('d', StringIO('missing file r\n'))]))
607
368
608
self.check_transport_contents(
369
609
'diff\ncontents for\na\n'
370
610
'add\nsome\nmore\ncontents\n'
372
612
'a little bit more\n'
373
613
'some text in a\n',
374
614
t, 'a')
375
self.check_transport_contents('some text\nfor a missing file\n',
376
t, 'c')
377
615
self.check_transport_contents('missing file r\n', t, 'd')
378
379
# a file with no parent should fail..
380
if not t.is_readonly():
381
self.assertRaises(NoSuchFile,
382
t.append, 'missing/path',
383
StringIO('content'))
384
385
def test_append_file(self):
386
t = self.get_transport()
387
388
contents = [
389
('f1', StringIO('this is a string\nand some more stuff\n')),
390
('f2', StringIO('here is some text\nand a bit more\n')),
391
('f3', StringIO('some text for the\nthird file created\n')),
392
('f4', StringIO('this is a string\nand some more stuff\n')),
393
('f5', StringIO('here is some text\nand a bit more\n')),
394
('f6', StringIO('some text for the\nthird file created\n'))
395
]
396
397
if t.is_readonly():
398
for f, val in contents:
399
open(f, 'wb').write(val.read())
400
else:
401
t.put_multi(contents)
402
403
a1 = StringIO('appending to\none\n')
404
if t.is_readonly():
405
_append('f1', a1)
406
else:
407
t.append('f1', a1)
408
409
del a1
410
411
self.check_transport_contents(
412
'this is a string\nand some more stuff\n'
413
'appending to\none\n',
414
t, 'f1')
415
416
a2 = StringIO('adding more\ntext to two\n')
417
a3 = StringIO('some garbage\nto put in three\n')
418
419
if t.is_readonly():
420
_append('f2', a2)
421
_append('f3', a3)
422
else:
423
t.append_multi([('f2', a2), ('f3', a3)])
424
425
del a2, a3
426
427
self.check_transport_contents(
428
'here is some text\nand a bit more\n'
429
'adding more\ntext to two\n',
430
t, 'f2')
431
self.check_transport_contents(
432
'some text for the\nthird file created\n'
433
'some garbage\nto put in three\n',
434
t, 'f3')
435
436
# Test that an actual file object can be used with put
437
a4 = t.get('f1')
438
if t.is_readonly():
439
_append('f4', a4)
440
else:
441
t.append('f4', a4)
442
443
del a4
444
445
self.check_transport_contents(
446
'this is a string\nand some more stuff\n'
447
'this is a string\nand some more stuff\n'
448
'appending to\none\n',
449
t, 'f4')
450
451
a5 = t.get('f2')
452
a6 = t.get('f3')
453
if t.is_readonly():
454
_append('f5', a5)
455
_append('f6', a6)
456
else:
457
t.append_multi([('f5', a5), ('f6', a6)])
458
459
del a5, a6
460
461
self.check_transport_contents(
462
'here is some text\nand a bit more\n'
463
'here is some text\nand a bit more\n'
464
'adding more\ntext to two\n',
465
t, 'f5')
466
self.check_transport_contents(
467
'some text for the\nthird file created\n'
468
'some text for the\nthird file created\n'
469
'some garbage\nto put in three\n',
470
t, 'f6')
471
472
a5 = t.get('f2')
473
a6 = t.get('f2')
474
a7 = t.get('f3')
475
if t.is_readonly():
476
_append('c', a5)
477
_append('a', a6)
478
_append('d', a7)
479
else:
480
t.append('c', a5)
481
t.append_multi([('a', a6), ('d', a7)])
482
del a5, a6, a7
483
self.check_transport_contents(t.get('f2').read(), t, 'c')
484
self.check_transport_contents(t.get('f3').read(), t, 'd')
485
486
def test_append_mode(self):
616
617
def test_append_file_mode(self):
618
"""Check that append accepts a mode parameter"""
487
619
# check append accepts a mode
488
620
t = self.get_transport()
489
621
if t.is_readonly():
490
return
491
t.append('f', StringIO('f'), mode=None)
622
self.assertRaises(TransportNotPossible,
623
t.append_file, 'f', StringIO('f'), mode=None)
624
return
625
t.append_file('f', StringIO('f'), mode=None)
626
627
def test_append_bytes_mode(self):
628
# check append_bytes accepts a mode
629
t = self.get_transport()
630
if t.is_readonly():
631
self.assertRaises(TransportNotPossible,
632
t.append_bytes, 'f', 'f', mode=None)
633
return
634
t.append_bytes('f', 'f', mode=None)
492
635
493
636
def test_delete(self):
494
637
# TODO: Test Transport.delete
499
642
self.assertRaises(TransportNotPossible, t.delete, 'missing')
500
643
return
501
644
502
t.put('a', StringIO('a little bit of text\n'))
645
t.put_bytes('a', 'a little bit of text\n')
503
646
self.failUnless(t.has('a'))
504
647
t.delete('a')
505
648
self.failIf(t.has('a'))
506
649
507
650
self.assertRaises(NoSuchFile, t.delete, 'a')
508
651
509
t.put('a', StringIO('a text\n'))
510
t.put('b', StringIO('b text\n'))
511
t.put('c', StringIO('c text\n'))
652
t.put_bytes('a', 'a text\n')
653
t.put_bytes('b', 'b text\n')
654
t.put_bytes('c', 'c text\n')
512
655
self.assertEqual([True, True, True],
513
656
list(t.has_multi(['a', 'b', 'c'])))
514
657
t.delete_multi(['a', 'c'])
524
667
self.assertRaises(NoSuchFile,
525
668
t.delete_multi, iter(['a', 'b', 'c']))
526
669
527
t.put('a', StringIO('another a text\n'))
528
t.put('c', StringIO('another c text\n'))
670
t.put_bytes('a', 'another a text\n')
671
t.put_bytes('c', 'another c text\n')
529
672
t.delete_multi(iter(['a', 'b', 'c']))
530
673
531
674
# We should have deleted everything
629
772
# creates control files in the working directory
630
773
# perhaps all of this could be done in a subdirectory
631
774
632
t.put('a', StringIO('a first file\n'))
775
t.put_bytes('a', 'a first file\n')
633
776
self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
634
777
635
778
t.move('a', 'b')
640
783
self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
641
784
642
785
# Overwrite a file
643
t.put('c', StringIO('c this file\n'))
786
t.put_bytes('c', 'c this file\n')
644
787
t.move('c', 'b')
645
788
self.failIf(t.has('c'))
646
789
self.check_transport_contents('c this file\n', t, 'b')
655
798
if t.is_readonly():
656
799
return
657
800
658
t.put('a', StringIO('a file\n'))
801
t.put_bytes('a', 'a file\n')
659
802
t.copy('a', 'b')
660
803
self.check_transport_contents('a file\n', t, 'b')
661
804
664
807
# What should the assert be if you try to copy a
665
808
# file over a directory?
666
809
#self.assertRaises(Something, t.copy, 'a', 'c')
667
t.put('d', StringIO('text in d\n'))
810
t.put_bytes('d', 'text in d\n')
668
811
t.copy('d', 'b')
669
812
self.check_transport_contents('text in d\n', t, 'b')
670
813
816
959
if t1.is_readonly():
817
960
open('b/d', 'wb').write('newfile\n')
818
961
else:
819
t2.put('d', StringIO('newfile\n'))
962
t2.put_bytes('d', 'newfile\n')
820
963
821
964
self.failUnless(t1.has('b/d'))
822
965
self.failUnless(t2.has('d'))
916
1059
transport.iter_files_recursive)
917
1060
return
918
1061
if transport.is_readonly():
919
self.assertRaises(TransportNotPossible,
920
transport.put, 'a', 'some text for a\n')
921
1062
return
922
1063
self.build_tree(['from/',
923
1064
'from/dir/',
974
1115
transport = self.get_transport()
975
1116
if transport.is_readonly():
976
1117
return
977
transport.put('foo', StringIO('bar'))
1118
transport.put_bytes('foo', 'bar')
978
1119
transport2 = self.get_transport()
979
1120
self.check_transport_contents('bar', transport2, 'foo')
980
1121
# its base should be usable.
992
1133
if transport.is_readonly():
993
1134
self.assertRaises(TransportNotPossible, transport.lock_write, 'foo')
994
1135
return
995
transport.put('lock', StringIO())
1136
transport.put_bytes('lock', '')
996
1137
lock = transport.lock_write('lock')
997
1138
# TODO make this consistent on all platforms:
998
1139
# self.assertRaises(LockError, transport.lock_write, 'lock')
1003
1144
if transport.is_readonly():
1004
1145
file('lock', 'w').close()
1005
1146
else:
1006
transport.put('lock', StringIO())
1147
transport.put_bytes('lock', '')
1007
1148
lock = transport.lock_read('lock')
1008
1149
# TODO make this consistent on all platforms:
1009
1150
# self.assertRaises(LockError, transport.lock_read, 'lock')
1014
1155
if transport.is_readonly():
1015
1156
file('a', 'w').write('0123456789')
1016
1157
else:
1017
transport.put('a', StringIO('0123456789'))
1158
transport.put_bytes('a', '0123456789')
1018
1159
1019
1160
d = list(transport.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
1020
1161
self.assertEqual(d[0], (0, '0'))
1027
1168
if transport.is_readonly():
1028
1169
file('a', 'w').write('0123456789')
1029
1170
else:
1030
transport.put('a', StringIO('01234567890'))
1171
transport.put_bytes('a', '01234567890')
1031
1172
1032
1173
d = list(transport.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
1033
1174
self.assertEqual(d[0], (1, '1'))
Loggerhead is a web-based interface for Breezy
Version: 2.0.1