« back to all changes in this revision
Viewing changes to bzrlib/tests/test_transport_implementations.py
- browse files at revision 1963.2.7
- compare with another revision
- compare with revision 5200.3.3
- download diff
- download tarball
- view history from revision 1963.2.7
- Committer: Robey Pointer
- Date: 2006-09-08 18:52:17 UTC
- mfrom: (1993 +trunk)
- mto: This revision was merged to the branch mainline in revision 1996.
- Revision ID: robey@lag.net-20060908185217-6a4406e1d41753f5
merge from bzr.dev
- files modified:
- Makefile
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
113
114
self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
114
115
self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
115
116
117
def test_get_bytes(self):
118
t = self.get_transport()
119
120
files = ['a', 'b', 'e', 'g']
121
contents = ['contents of a\n',
122
'contents of b\n',
123
'contents of e\n',
124
'contents of g\n',
125
]
126
self.build_tree(files, transport=t, line_endings='binary')
127
self.check_transport_contents('contents of a\n', t, 'a')
128
129
for content, fname in zip(contents, files):
130
self.assertEqual(content, t.get_bytes(fname))
131
132
self.assertRaises(NoSuchFile, t.get_bytes, 'c')
133
116
134
def test_put(self):
117
135
t = self.get_transport()
118
136
119
137
if t.is_readonly():
120
self.assertRaises(TransportNotPossible,
121
t.put, 'a', 'some text for a\n')
122
return
123
124
t.put('a', StringIO('some text for a\n'))
125
self.failUnless(t.has('a'))
126
self.check_transport_contents('some text for a\n', t, 'a')
127
# Make sure 'has' is updated
128
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
129
[True, False, False, False, False])
130
# Put also replaces contents
131
self.assertEqual(t.put_multi([('a', StringIO('new\ncontents for\na\n')),
132
('d', StringIO('contents\nfor d\n'))]),
133
2)
134
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
135
[True, False, False, True, False])
138
return
139
140
self.applyDeprecated(zero_eleven, t.put, 'a', 'string\ncontents\n')
141
self.check_transport_contents('string\ncontents\n', t, 'a')
142
143
self.applyDeprecated(zero_eleven,
144
t.put, 'b', StringIO('file-like\ncontents\n'))
145
self.check_transport_contents('file-like\ncontents\n', t, 'b')
146
147
def test_put_bytes(self):
148
t = self.get_transport()
149
150
if t.is_readonly():
151
self.assertRaises(TransportNotPossible,
152
t.put_bytes, 'a', 'some text for a\n')
153
return
154
155
t.put_bytes('a', 'some text for a\n')
156
self.failUnless(t.has('a'))
157
self.check_transport_contents('some text for a\n', t, 'a')
158
159
# The contents should be overwritten
160
t.put_bytes('a', 'new text for a\n')
161
self.check_transport_contents('new text for a\n', t, 'a')
162
163
self.assertRaises(NoSuchFile,
164
t.put_bytes, 'path/doesnt/exist/c', 'contents')
165
166
def test_put_bytes_non_atomic(self):
167
t = self.get_transport()
168
169
if t.is_readonly():
170
self.assertRaises(TransportNotPossible,
171
t.put_bytes_non_atomic, 'a', 'some text for a\n')
172
return
173
174
self.failIf(t.has('a'))
175
t.put_bytes_non_atomic('a', 'some text for a\n')
176
self.failUnless(t.has('a'))
177
self.check_transport_contents('some text for a\n', t, 'a')
178
# Put also replaces contents
179
t.put_bytes_non_atomic('a', 'new\ncontents for\na\n')
180
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
181
182
# Make sure we can create another file
183
t.put_bytes_non_atomic('d', 'contents for\nd\n')
184
# And overwrite 'a' with empty contents
185
t.put_bytes_non_atomic('a', '')
186
self.check_transport_contents('contents for\nd\n', t, 'd')
187
self.check_transport_contents('', t, 'a')
188
189
self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'no/such/path',
190
'contents\n')
191
# Now test the create_parent flag
192
self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'dir/a',
193
'contents\n')
194
self.failIf(t.has('dir/a'))
195
t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',
196
create_parent_dir=True)
197
self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
198
199
# But we still get NoSuchFile if we can't make the parent dir
200
self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'not/there/a',
201
'contents\n',
202
create_parent_dir=True)
203
204
def test_put_bytes_permissions(self):
205
t = self.get_transport()
206
207
if t.is_readonly():
208
return
209
if not t._can_roundtrip_unix_modebits():
210
# Can't roundtrip, so no need to run this test
211
return
212
t.put_bytes('mode644', 'test text\n', mode=0644)
213
self.assertTransportMode(t, 'mode644', 0644)
214
t.put_bytes('mode666', 'test text\n', mode=0666)
215
self.assertTransportMode(t, 'mode666', 0666)
216
t.put_bytes('mode600', 'test text\n', mode=0600)
217
self.assertTransportMode(t, 'mode600', 0600)
218
# Yes, you can put_bytes a file such that it becomes readonly
219
t.put_bytes('mode400', 'test text\n', mode=0400)
220
self.assertTransportMode(t, 'mode400', 0400)
221
222
# The default permissions should be based on the current umask
223
umask = osutils.get_umask()
224
t.put_bytes('nomode', 'test text\n', mode=None)
225
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
226
227
def test_put_bytes_non_atomic_permissions(self):
228
t = self.get_transport()
229
230
if t.is_readonly():
231
return
232
if not t._can_roundtrip_unix_modebits():
233
# Can't roundtrip, so no need to run this test
234
return
235
t.put_bytes_non_atomic('mode644', 'test text\n', mode=0644)
236
self.assertTransportMode(t, 'mode644', 0644)
237
t.put_bytes_non_atomic('mode666', 'test text\n', mode=0666)
238
self.assertTransportMode(t, 'mode666', 0666)
239
t.put_bytes_non_atomic('mode600', 'test text\n', mode=0600)
240
self.assertTransportMode(t, 'mode600', 0600)
241
t.put_bytes_non_atomic('mode400', 'test text\n', mode=0400)
242
self.assertTransportMode(t, 'mode400', 0400)
243
244
# The default permissions should be based on the current umask
245
umask = osutils.get_umask()
246
t.put_bytes_non_atomic('nomode', 'test text\n', mode=None)
247
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
248
249
# We should also be able to set the mode for a parent directory
250
# when it is created
251
t.put_bytes_non_atomic('dir700/mode664', 'test text\n', mode=0664,
252
dir_mode=0700, create_parent_dir=True)
253
self.assertTransportMode(t, 'dir700', 0700)
254
t.put_bytes_non_atomic('dir770/mode664', 'test text\n', mode=0664,
255
dir_mode=0770, create_parent_dir=True)
256
self.assertTransportMode(t, 'dir770', 0770)
257
t.put_bytes_non_atomic('dir777/mode664', 'test text\n', mode=0664,
258
dir_mode=0777, create_parent_dir=True)
259
self.assertTransportMode(t, 'dir777', 0777)
260
261
def test_put_file(self):
262
t = self.get_transport()
263
264
if t.is_readonly():
265
self.assertRaises(TransportNotPossible,
266
t.put_file, 'a', StringIO('some text for a\n'))
267
return
268
269
t.put_file('a', StringIO('some text for a\n'))
270
self.failUnless(t.has('a'))
271
self.check_transport_contents('some text for a\n', t, 'a')
272
# Put also replaces contents
273
t.put_file('a', StringIO('new\ncontents for\na\n'))
274
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
275
self.assertRaises(NoSuchFile,
276
t.put_file, 'path/doesnt/exist/c',
277
StringIO('contents'))
278
279
def test_put_file_non_atomic(self):
280
t = self.get_transport()
281
282
if t.is_readonly():
283
self.assertRaises(TransportNotPossible,
284
t.put_file_non_atomic, 'a', StringIO('some text for a\n'))
285
return
286
287
self.failIf(t.has('a'))
288
t.put_file_non_atomic('a', StringIO('some text for a\n'))
289
self.failUnless(t.has('a'))
290
self.check_transport_contents('some text for a\n', t, 'a')
291
# Put also replaces contents
292
t.put_file_non_atomic('a', StringIO('new\ncontents for\na\n'))
293
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
294
295
# Make sure we can create another file
296
t.put_file_non_atomic('d', StringIO('contents for\nd\n'))
297
# And overwrite 'a' with empty contents
298
t.put_file_non_atomic('a', StringIO(''))
299
self.check_transport_contents('contents for\nd\n', t, 'd')
300
self.check_transport_contents('', t, 'a')
301
302
self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'no/such/path',
303
StringIO('contents\n'))
304
# Now test the create_parent flag
305
self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'dir/a',
306
StringIO('contents\n'))
307
self.failIf(t.has('dir/a'))
308
t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
309
create_parent_dir=True)
310
self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
311
312
# But we still get NoSuchFile if we can't make the parent dir
313
self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'not/there/a',
314
StringIO('contents\n'),
315
create_parent_dir=True)
316
317
def test_put_file_permissions(self):
318
319
t = self.get_transport()
320
321
if t.is_readonly():
322
return
323
if not t._can_roundtrip_unix_modebits():
324
# Can't roundtrip, so no need to run this test
325
return
326
t.put_file('mode644', StringIO('test text\n'), mode=0644)
327
self.assertTransportMode(t, 'mode644', 0644)
328
t.put_file('mode666', StringIO('test text\n'), mode=0666)
329
self.assertTransportMode(t, 'mode666', 0666)
330
t.put_file('mode600', StringIO('test text\n'), mode=0600)
331
self.assertTransportMode(t, 'mode600', 0600)
332
# Yes, you can put a file such that it becomes readonly
333
t.put_file('mode400', StringIO('test text\n'), mode=0400)
334
self.assertTransportMode(t, 'mode400', 0400)
335
336
# XXX: put_multi is deprecated, so do we really care anymore?
337
self.applyDeprecated(zero_eleven, t.put_multi,
338
[('mmode644', StringIO('text\n'))], mode=0644)
339
self.assertTransportMode(t, 'mmode644', 0644)
340
341
# The default permissions should be based on the current umask
342
umask = osutils.get_umask()
343
t.put_file('nomode', StringIO('test text\n'), mode=None)
344
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
345
346
def test_put_file_non_atomic_permissions(self):
347
t = self.get_transport()
348
349
if t.is_readonly():
350
return
351
if not t._can_roundtrip_unix_modebits():
352
# Can't roundtrip, so no need to run this test
353
return
354
t.put_file_non_atomic('mode644', StringIO('test text\n'), mode=0644)
355
self.assertTransportMode(t, 'mode644', 0644)
356
t.put_file_non_atomic('mode666', StringIO('test text\n'), mode=0666)
357
self.assertTransportMode(t, 'mode666', 0666)
358
t.put_file_non_atomic('mode600', StringIO('test text\n'), mode=0600)
359
self.assertTransportMode(t, 'mode600', 0600)
360
# Yes, you can put_file_non_atomic a file such that it becomes readonly
361
t.put_file_non_atomic('mode400', StringIO('test text\n'), mode=0400)
362
self.assertTransportMode(t, 'mode400', 0400)
363
364
# The default permissions should be based on the current umask
365
umask = osutils.get_umask()
366
t.put_file_non_atomic('nomode', StringIO('test text\n'), mode=None)
367
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
368
369
# We should also be able to set the mode for a parent directory
370
# when it is created
371
sio = StringIO()
372
t.put_file_non_atomic('dir700/mode664', sio, mode=0664,
373
dir_mode=0700, create_parent_dir=True)
374
self.assertTransportMode(t, 'dir700', 0700)
375
t.put_file_non_atomic('dir770/mode664', sio, mode=0664,
376
dir_mode=0770, create_parent_dir=True)
377
self.assertTransportMode(t, 'dir770', 0770)
378
t.put_file_non_atomic('dir777/mode664', sio, mode=0664,
379
dir_mode=0777, create_parent_dir=True)
380
self.assertTransportMode(t, 'dir777', 0777)
381
382
def test_put_multi(self):
383
t = self.get_transport()
384
385
if t.is_readonly():
386
return
387
self.assertEqual(2, self.applyDeprecated(zero_eleven,
388
t.put_multi, [('a', StringIO('new\ncontents for\na\n')),
389
('d', StringIO('contents\nfor d\n'))]
390
))
391
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd'])),
392
[True, False, False, True])
136
393
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
137
394
self.check_transport_contents('contents\nfor d\n', t, 'd')
138
395
139
self.assertEqual(
140
t.put_multi(iter([('a', StringIO('diff\ncontents for\na\n')),
141
('d', StringIO('another contents\nfor d\n'))])),
142
2)
396
self.assertEqual(2, self.applyDeprecated(zero_eleven,
397
t.put_multi, iter([('a', StringIO('diff\ncontents for\na\n')),
398
('d', StringIO('another contents\nfor d\n'))])
399
))
143
400
self.check_transport_contents('diff\ncontents for\na\n', t, 'a')
144
401
self.check_transport_contents('another contents\nfor d\n', t, 'd')
145
402
146
self.assertRaises(NoSuchFile,
147
t.put, 'path/doesnt/exist/c', StringIO('contents'))
148
149
def test_put_permissions(self):
150
t = self.get_transport()
151
152
if t.is_readonly():
153
return
154
if not t._can_roundtrip_unix_modebits():
155
# Can't roundtrip, so no need to run this test
156
return
157
t.put('mode644', StringIO('test text\n'), mode=0644)
158
self.assertTransportMode(t, 'mode644', 0644)
159
t.put('mode666', StringIO('test text\n'), mode=0666)
160
self.assertTransportMode(t, 'mode666', 0666)
161
t.put('mode600', StringIO('test text\n'), mode=0600)
162
self.assertTransportMode(t, 'mode600', 0600)
163
# Yes, you can put a file such that it becomes readonly
164
t.put('mode400', StringIO('test text\n'), mode=0400)
165
self.assertTransportMode(t, 'mode400', 0400)
166
t.put_multi([('mmode644', StringIO('text\n'))], mode=0644)
167
self.assertTransportMode(t, 'mmode644', 0644)
168
169
# The default permissions should be based on the current umask
170
umask = osutils.get_umask()
171
t.put('nomode', StringIO('test text\n'), mode=None)
172
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
173
174
403
def test_mkdir(self):
175
404
t = self.get_transport()
176
405
209
438
self.assertRaises(FileExists, t.mkdir, 'dir_g')
210
439
211
440
# Test get/put in sub-directories
212
self.assertEqual(2,
213
t.put_multi([('dir_a/a', StringIO('contents of dir_a/a')),
214
('dir_b/b', StringIO('contents of dir_b/b'))]))
441
t.put_bytes('dir_a/a', 'contents of dir_a/a')
442
t.put_file('dir_b/b', StringIO('contents of dir_b/b'))
215
443
self.check_transport_contents('contents of dir_a/a', t, 'dir_a/a')
216
444
self.check_transport_contents('contents of dir_b/b', t, 'dir_b/b')
217
445
272
500
self.build_tree(['e/', 'e/f'])
273
501
else:
274
502
t.mkdir('e')
275
t.put('e/f', StringIO('contents of e'))
503
t.put_bytes('e/f', 'contents of e')
276
504
self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], temp_transport)
277
505
temp_transport.mkdir('e')
278
506
t.copy_to(['e/f'], temp_transport)
297
525
t = self.get_transport()
298
526
299
527
if t.is_readonly():
300
open('a', 'wb').write('diff\ncontents for\na\n')
301
open('b', 'wb').write('contents\nfor b\n')
302
else:
303
t.put_multi([
304
('a', StringIO('diff\ncontents for\na\n')),
305
('b', StringIO('contents\nfor b\n'))
306
])
307
308
if t.is_readonly():
309
self.assertRaises(TransportNotPossible,
310
t.append, 'a', 'add\nsome\nmore\ncontents\n')
311
_append('a', StringIO('add\nsome\nmore\ncontents\n'))
312
else:
313
self.assertEqual(20,
314
t.append('a', StringIO('add\nsome\nmore\ncontents\n')))
315
316
self.check_transport_contents(
317
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
318
t, 'a')
319
320
if t.is_readonly():
321
self.assertRaises(TransportNotPossible,
322
t.append_multi,
323
[('a', 'and\nthen\nsome\nmore\n'),
324
('b', 'some\nmore\nfor\nb\n')])
325
_append('a', StringIO('and\nthen\nsome\nmore\n'))
326
_append('b', StringIO('some\nmore\nfor\nb\n'))
327
else:
328
self.assertEqual((43, 15),
329
t.append_multi([('a', StringIO('and\nthen\nsome\nmore\n')),
330
('b', StringIO('some\nmore\nfor\nb\n'))]))
528
return
529
t.put_bytes('a', 'diff\ncontents for\na\n')
530
t.put_bytes('b', 'contents\nfor b\n')
531
532
self.assertEqual(20, self.applyDeprecated(zero_eleven,
533
t.append, 'a', StringIO('add\nsome\nmore\ncontents\n')))
534
535
self.check_transport_contents(
536
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
537
t, 'a')
538
539
# And we can create new files, too
540
self.assertEqual(0, self.applyDeprecated(zero_eleven,
541
t.append, 'c', StringIO('some text\nfor a missing file\n')))
542
self.check_transport_contents('some text\nfor a missing file\n',
543
t, 'c')
544
def test_append_file(self):
545
t = self.get_transport()
546
547
if t.is_readonly():
548
self.assertRaises(TransportNotPossible,
549
t.append_file, 'a', 'add\nsome\nmore\ncontents\n')
550
return
551
t.put_bytes('a', 'diff\ncontents for\na\n')
552
t.put_bytes('b', 'contents\nfor b\n')
553
554
self.assertEqual(20,
555
t.append_file('a', StringIO('add\nsome\nmore\ncontents\n')))
556
557
self.check_transport_contents(
558
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
559
t, 'a')
560
561
# a file with no parent should fail..
562
self.assertRaises(NoSuchFile,
563
t.append_file, 'missing/path', StringIO('content'))
564
565
# And we can create new files, too
566
self.assertEqual(0,
567
t.append_file('c', StringIO('some text\nfor a missing file\n')))
568
self.check_transport_contents('some text\nfor a missing file\n',
569
t, 'c')
570
571
def test_append_bytes(self):
572
t = self.get_transport()
573
574
if t.is_readonly():
575
self.assertRaises(TransportNotPossible,
576
t.append_bytes, 'a', 'add\nsome\nmore\ncontents\n')
577
return
578
579
self.assertEqual(0, t.append_bytes('a', 'diff\ncontents for\na\n'))
580
self.assertEqual(0, t.append_bytes('b', 'contents\nfor b\n'))
581
582
self.assertEqual(20,
583
t.append_bytes('a', 'add\nsome\nmore\ncontents\n'))
584
585
self.check_transport_contents(
586
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
587
t, 'a')
588
589
# a file with no parent should fail..
590
self.assertRaises(NoSuchFile,
591
t.append_bytes, 'missing/path', 'content')
592
593
def test_append_multi(self):
594
t = self.get_transport()
595
596
if t.is_readonly():
597
return
598
t.put_bytes('a', 'diff\ncontents for\na\n'
599
'add\nsome\nmore\ncontents\n')
600
t.put_bytes('b', 'contents\nfor b\n')
601
602
self.assertEqual((43, 15),
603
t.append_multi([('a', StringIO('and\nthen\nsome\nmore\n')),
604
('b', StringIO('some\nmore\nfor\nb\n'))]))
605
331
606
self.check_transport_contents(
332
607
'diff\ncontents for\na\n'
333
608
'add\nsome\nmore\ncontents\n'
338
613
'some\nmore\nfor\nb\n',
339
614
t, 'b')
340
615
341
if t.is_readonly():
342
_append('a', StringIO('a little bit more\n'))
343
_append('b', StringIO('from an iterator\n'))
344
else:
345
self.assertEqual((62, 31),
346
t.append_multi(iter([('a', StringIO('a little bit more\n')),
347
('b', StringIO('from an iterator\n'))])))
616
self.assertEqual((62, 31),
617
t.append_multi(iter([('a', StringIO('a little bit more\n')),
618
('b', StringIO('from an iterator\n'))])))
348
619
self.check_transport_contents(
349
620
'diff\ncontents for\na\n'
350
621
'add\nsome\nmore\ncontents\n'
357
628
'from an iterator\n',
358
629
t, 'b')
359
630
360
if t.is_readonly():
361
_append('c', StringIO('some text\nfor a missing file\n'))
362
_append('a', StringIO('some text in a\n'))
363
_append('d', StringIO('missing file r\n'))
364
else:
365
self.assertEqual(0,
366
t.append('c', StringIO('some text\nfor a missing file\n')))
367
self.assertEqual((80, 0),
368
t.append_multi([('a', StringIO('some text in a\n')),
369
('d', StringIO('missing file r\n'))]))
631
self.assertEqual((80, 0),
632
t.append_multi([('a', StringIO('some text in a\n')),
633
('d', StringIO('missing file r\n'))]))
634
370
635
self.check_transport_contents(
371
636
'diff\ncontents for\na\n'
372
637
'add\nsome\nmore\ncontents\n'
374
639
'a little bit more\n'
375
640
'some text in a\n',
376
641
t, 'a')
377
self.check_transport_contents('some text\nfor a missing file\n',
378
t, 'c')
379
642
self.check_transport_contents('missing file r\n', t, 'd')
380
381
# a file with no parent should fail..
382
if not t.is_readonly():
383
self.assertRaises(NoSuchFile,
384
t.append, 'missing/path',
385
StringIO('content'))
386
387
def test_append_file(self):
388
t = self.get_transport()
389
390
contents = [
391
('f1', StringIO('this is a string\nand some more stuff\n')),
392
('f2', StringIO('here is some text\nand a bit more\n')),
393
('f3', StringIO('some text for the\nthird file created\n')),
394
('f4', StringIO('this is a string\nand some more stuff\n')),
395
('f5', StringIO('here is some text\nand a bit more\n')),
396
('f6', StringIO('some text for the\nthird file created\n'))
397
]
398
399
if t.is_readonly():
400
for f, val in contents:
401
open(f, 'wb').write(val.read())
402
else:
403
t.put_multi(contents)
404
405
a1 = StringIO('appending to\none\n')
406
if t.is_readonly():
407
_append('f1', a1)
408
else:
409
t.append('f1', a1)
410
411
del a1
412
413
self.check_transport_contents(
414
'this is a string\nand some more stuff\n'
415
'appending to\none\n',
416
t, 'f1')
417
418
a2 = StringIO('adding more\ntext to two\n')
419
a3 = StringIO('some garbage\nto put in three\n')
420
421
if t.is_readonly():
422
_append('f2', a2)
423
_append('f3', a3)
424
else:
425
t.append_multi([('f2', a2), ('f3', a3)])
426
427
del a2, a3
428
429
self.check_transport_contents(
430
'here is some text\nand a bit more\n'
431
'adding more\ntext to two\n',
432
t, 'f2')
433
self.check_transport_contents(
434
'some text for the\nthird file created\n'
435
'some garbage\nto put in three\n',
436
t, 'f3')
437
438
# Test that an actual file object can be used with put
439
a4 = t.get('f1')
440
if t.is_readonly():
441
_append('f4', a4)
442
else:
443
t.append('f4', a4)
444
445
del a4
446
447
self.check_transport_contents(
448
'this is a string\nand some more stuff\n'
449
'this is a string\nand some more stuff\n'
450
'appending to\none\n',
451
t, 'f4')
452
453
a5 = t.get('f2')
454
a6 = t.get('f3')
455
if t.is_readonly():
456
_append('f5', a5)
457
_append('f6', a6)
458
else:
459
t.append_multi([('f5', a5), ('f6', a6)])
460
461
del a5, a6
462
463
self.check_transport_contents(
464
'here is some text\nand a bit more\n'
465
'here is some text\nand a bit more\n'
466
'adding more\ntext to two\n',
467
t, 'f5')
468
self.check_transport_contents(
469
'some text for the\nthird file created\n'
470
'some text for the\nthird file created\n'
471
'some garbage\nto put in three\n',
472
t, 'f6')
473
474
a5 = t.get('f2')
475
a6 = t.get('f2')
476
a7 = t.get('f3')
477
if t.is_readonly():
478
_append('c', a5)
479
_append('a', a6)
480
_append('d', a7)
481
else:
482
t.append('c', a5)
483
t.append_multi([('a', a6), ('d', a7)])
484
del a5, a6, a7
485
self.check_transport_contents(t.get('f2').read(), t, 'c')
486
self.check_transport_contents(t.get('f3').read(), t, 'd')
487
488
def test_append_mode(self):
643
644
def test_append_file_mode(self):
645
"""Check that append accepts a mode parameter"""
489
646
# check append accepts a mode
490
647
t = self.get_transport()
491
648
if t.is_readonly():
492
return
493
t.append('f', StringIO('f'), mode=None)
649
self.assertRaises(TransportNotPossible,
650
t.append_file, 'f', StringIO('f'), mode=None)
651
return
652
t.append_file('f', StringIO('f'), mode=None)
653
654
def test_append_bytes_mode(self):
655
# check append_bytes accepts a mode
656
t = self.get_transport()
657
if t.is_readonly():
658
self.assertRaises(TransportNotPossible,
659
t.append_bytes, 'f', 'f', mode=None)
660
return
661
t.append_bytes('f', 'f', mode=None)
494
662
495
663
def test_delete(self):
496
664
# TODO: Test Transport.delete
501
669
self.assertRaises(TransportNotPossible, t.delete, 'missing')
502
670
return
503
671
504
t.put('a', StringIO('a little bit of text\n'))
672
t.put_bytes('a', 'a little bit of text\n')
505
673
self.failUnless(t.has('a'))
506
674
t.delete('a')
507
675
self.failIf(t.has('a'))
508
676
509
677
self.assertRaises(NoSuchFile, t.delete, 'a')
510
678
511
t.put('a', StringIO('a text\n'))
512
t.put('b', StringIO('b text\n'))
513
t.put('c', StringIO('c text\n'))
679
t.put_bytes('a', 'a text\n')
680
t.put_bytes('b', 'b text\n')
681
t.put_bytes('c', 'c text\n')
514
682
self.assertEqual([True, True, True],
515
683
list(t.has_multi(['a', 'b', 'c'])))
516
684
t.delete_multi(['a', 'c'])
526
694
self.assertRaises(NoSuchFile,
527
695
t.delete_multi, iter(['a', 'b', 'c']))
528
696
529
t.put('a', StringIO('another a text\n'))
530
t.put('c', StringIO('another c text\n'))
697
t.put_bytes('a', 'another a text\n')
698
t.put_bytes('c', 'another c text\n')
531
699
t.delete_multi(iter(['a', 'b', 'c']))
532
700
533
701
# We should have deleted everything
631
799
# creates control files in the working directory
632
800
# perhaps all of this could be done in a subdirectory
633
801
634
t.put('a', StringIO('a first file\n'))
802
t.put_bytes('a', 'a first file\n')
635
803
self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
636
804
637
805
t.move('a', 'b')
642
810
self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
643
811
644
812
# Overwrite a file
645
t.put('c', StringIO('c this file\n'))
813
t.put_bytes('c', 'c this file\n')
646
814
t.move('c', 'b')
647
815
self.failIf(t.has('c'))
648
816
self.check_transport_contents('c this file\n', t, 'b')
657
825
if t.is_readonly():
658
826
return
659
827
660
t.put('a', StringIO('a file\n'))
828
t.put_bytes('a', 'a file\n')
661
829
t.copy('a', 'b')
662
830
self.check_transport_contents('a file\n', t, 'b')
663
831
666
834
# What should the assert be if you try to copy a
667
835
# file over a directory?
668
836
#self.assertRaises(Something, t.copy, 'a', 'c')
669
t.put('d', StringIO('text in d\n'))
837
t.put_bytes('d', 'text in d\n')
670
838
t.copy('d', 'b')
671
839
self.check_transport_contents('text in d\n', t, 'b')
672
840
818
986
if t1.is_readonly():
819
987
open('b/d', 'wb').write('newfile\n')
820
988
else:
821
t2.put('d', StringIO('newfile\n'))
989
t2.put_bytes('d', 'newfile\n')
822
990
823
991
self.failUnless(t1.has('b/d'))
824
992
self.failUnless(t2.has('d'))
918
1086
transport.iter_files_recursive)
919
1087
return
920
1088
if transport.is_readonly():
921
self.assertRaises(TransportNotPossible,
922
transport.put, 'a', 'some text for a\n')
923
1089
return
924
1090
self.build_tree(['from/',
925
1091
'from/dir/',
976
1142
transport = self.get_transport()
977
1143
if transport.is_readonly():
978
1144
return
979
transport.put('foo', StringIO('bar'))
1145
transport.put_bytes('foo', 'bar')
980
1146
transport2 = self.get_transport()
981
1147
self.check_transport_contents('bar', transport2, 'foo')
982
1148
# its base should be usable.
994
1160
if transport.is_readonly():
995
1161
self.assertRaises(TransportNotPossible, transport.lock_write, 'foo')
996
1162
return
997
transport.put('lock', StringIO())
1163
transport.put_bytes('lock', '')
998
1164
lock = transport.lock_write('lock')
999
1165
# TODO make this consistent on all platforms:
1000
1166
# self.assertRaises(LockError, transport.lock_write, 'lock')
1005
1171
if transport.is_readonly():
1006
1172
file('lock', 'w').close()
1007
1173
else:
1008
transport.put('lock', StringIO())
1174
transport.put_bytes('lock', '')
1009
1175
lock = transport.lock_read('lock')
1010
1176
# TODO make this consistent on all platforms:
1011
1177
# self.assertRaises(LockError, transport.lock_read, 'lock')
1016
1182
if transport.is_readonly():
1017
1183
file('a', 'w').write('0123456789')
1018
1184
else:
1019
transport.put('a', StringIO('0123456789'))
1185
transport.put_bytes('a', '0123456789')
1020
1186
1021
1187
d = list(transport.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
1022
1188
self.assertEqual(d[0], (0, '0'))
1029
1195
if transport.is_readonly():
1030
1196
file('a', 'w').write('0123456789')
1031
1197
else:
1032
transport.put('a', StringIO('01234567890'))
1198
transport.put_bytes('a', '01234567890')
1033
1199
1034
1200
d = list(transport.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
1035
1201
self.assertEqual(d[0], (1, '1'))
Loggerhead is a web-based interface for Breezy
Version: 2.0.1