To get this branch, use:
bzr branch
http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar
« back to all changes in this revision
Viewing changes to bzrlib/tests/test_transport_implementations.py
- browse files at revision 1910.15.6
- compare with another revision
- compare with revision 1036
- download diff
- download tarball
- view history from revision 1910.15.6
- Committer: Andrew Bennetts
- Date: 2006-09-13 12:32:47 UTC
- mfrom: (1910.14.2 workingtree creation bug fix.)
- mto: (1910.16.4 lock_read/lock_write raise TransportNotPossible)
- mto: This revision was merged to the branch mainline in revision 2010.
- Revision ID: andrew.bennetts@canonical.com-20060913123247-48b9c5e047010a4d
Merge from bzr.dev
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
67
68
except excClass:
68
69
return
69
70
else:
70
if hasattr(excClass,'__name__'): excName = excClass.__name__
71
else: excName = str(excClass)
71
if getattr(excClass,'__name__', None) is not None:
72
excName = excClass.__name__
73
else:
74
excName = str(excClass)
72
75
raise self.failureException, "%s not raised" % excName
73
76
74
77
def test_has(self):
111
114
self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
112
115
self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
113
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
114
134
def test_put(self):
115
135
t = self.get_transport()
116
136
117
137
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])
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
self.assertRaises(NoSuchFile,
148
t.put, 'path/doesnt/exist/c', StringIO('contents'))
149
150
def test_put_bytes(self):
151
t = self.get_transport()
152
153
if t.is_readonly():
154
self.assertRaises(TransportNotPossible,
155
t.put_bytes, 'a', 'some text for a\n')
156
return
157
158
t.put_bytes('a', 'some text for a\n')
159
self.failUnless(t.has('a'))
160
self.check_transport_contents('some text for a\n', t, 'a')
161
162
# The contents should be overwritten
163
t.put_bytes('a', 'new text for a\n')
164
self.check_transport_contents('new text for a\n', t, 'a')
165
166
self.assertRaises(NoSuchFile,
167
t.put_bytes, 'path/doesnt/exist/c', 'contents')
168
169
def test_put_bytes_non_atomic(self):
170
t = self.get_transport()
171
172
if t.is_readonly():
173
self.assertRaises(TransportNotPossible,
174
t.put_bytes_non_atomic, 'a', 'some text for a\n')
175
return
176
177
self.failIf(t.has('a'))
178
t.put_bytes_non_atomic('a', 'some text for a\n')
179
self.failUnless(t.has('a'))
180
self.check_transport_contents('some text for a\n', t, 'a')
181
# Put also replaces contents
182
t.put_bytes_non_atomic('a', 'new\ncontents for\na\n')
183
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
184
185
# Make sure we can create another file
186
t.put_bytes_non_atomic('d', 'contents for\nd\n')
187
# And overwrite 'a' with empty contents
188
t.put_bytes_non_atomic('a', '')
189
self.check_transport_contents('contents for\nd\n', t, 'd')
190
self.check_transport_contents('', t, 'a')
191
192
self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'no/such/path',
193
'contents\n')
194
# Now test the create_parent flag
195
self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'dir/a',
196
'contents\n')
197
self.failIf(t.has('dir/a'))
198
t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',
199
create_parent_dir=True)
200
self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
201
202
# But we still get NoSuchFile if we can't make the parent dir
203
self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'not/there/a',
204
'contents\n',
205
create_parent_dir=True)
206
207
def test_put_bytes_permissions(self):
208
t = self.get_transport()
209
210
if t.is_readonly():
211
return
212
if not t._can_roundtrip_unix_modebits():
213
# Can't roundtrip, so no need to run this test
214
return
215
t.put_bytes('mode644', 'test text\n', mode=0644)
216
self.assertTransportMode(t, 'mode644', 0644)
217
t.put_bytes('mode666', 'test text\n', mode=0666)
218
self.assertTransportMode(t, 'mode666', 0666)
219
t.put_bytes('mode600', 'test text\n', mode=0600)
220
self.assertTransportMode(t, 'mode600', 0600)
221
# Yes, you can put_bytes a file such that it becomes readonly
222
t.put_bytes('mode400', 'test text\n', mode=0400)
223
self.assertTransportMode(t, 'mode400', 0400)
224
225
# The default permissions should be based on the current umask
226
umask = osutils.get_umask()
227
t.put_bytes('nomode', 'test text\n', mode=None)
228
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
229
230
def test_put_bytes_non_atomic_permissions(self):
231
t = self.get_transport()
232
233
if t.is_readonly():
234
return
235
if not t._can_roundtrip_unix_modebits():
236
# Can't roundtrip, so no need to run this test
237
return
238
t.put_bytes_non_atomic('mode644', 'test text\n', mode=0644)
239
self.assertTransportMode(t, 'mode644', 0644)
240
t.put_bytes_non_atomic('mode666', 'test text\n', mode=0666)
241
self.assertTransportMode(t, 'mode666', 0666)
242
t.put_bytes_non_atomic('mode600', 'test text\n', mode=0600)
243
self.assertTransportMode(t, 'mode600', 0600)
244
t.put_bytes_non_atomic('mode400', 'test text\n', mode=0400)
245
self.assertTransportMode(t, 'mode400', 0400)
246
247
# The default permissions should be based on the current umask
248
umask = osutils.get_umask()
249
t.put_bytes_non_atomic('nomode', 'test text\n', mode=None)
250
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
251
252
# We should also be able to set the mode for a parent directory
253
# when it is created
254
t.put_bytes_non_atomic('dir700/mode664', 'test text\n', mode=0664,
255
dir_mode=0700, create_parent_dir=True)
256
self.assertTransportMode(t, 'dir700', 0700)
257
t.put_bytes_non_atomic('dir770/mode664', 'test text\n', mode=0664,
258
dir_mode=0770, create_parent_dir=True)
259
self.assertTransportMode(t, 'dir770', 0770)
260
t.put_bytes_non_atomic('dir777/mode664', 'test text\n', mode=0664,
261
dir_mode=0777, create_parent_dir=True)
262
self.assertTransportMode(t, 'dir777', 0777)
263
264
def test_put_file(self):
265
t = self.get_transport()
266
267
if t.is_readonly():
268
self.assertRaises(TransportNotPossible,
269
t.put_file, 'a', StringIO('some text for a\n'))
270
return
271
272
t.put_file('a', StringIO('some text for a\n'))
273
self.failUnless(t.has('a'))
274
self.check_transport_contents('some text for a\n', t, 'a')
275
# Put also replaces contents
276
t.put_file('a', StringIO('new\ncontents for\na\n'))
277
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
278
self.assertRaises(NoSuchFile,
279
t.put_file, 'path/doesnt/exist/c',
280
StringIO('contents'))
281
282
def test_put_file_non_atomic(self):
283
t = self.get_transport()
284
285
if t.is_readonly():
286
self.assertRaises(TransportNotPossible,
287
t.put_file_non_atomic, 'a', StringIO('some text for a\n'))
288
return
289
290
self.failIf(t.has('a'))
291
t.put_file_non_atomic('a', StringIO('some text for a\n'))
292
self.failUnless(t.has('a'))
293
self.check_transport_contents('some text for a\n', t, 'a')
294
# Put also replaces contents
295
t.put_file_non_atomic('a', StringIO('new\ncontents for\na\n'))
296
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
297
298
# Make sure we can create another file
299
t.put_file_non_atomic('d', StringIO('contents for\nd\n'))
300
# And overwrite 'a' with empty contents
301
t.put_file_non_atomic('a', StringIO(''))
302
self.check_transport_contents('contents for\nd\n', t, 'd')
303
self.check_transport_contents('', t, 'a')
304
305
self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'no/such/path',
306
StringIO('contents\n'))
307
# Now test the create_parent flag
308
self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'dir/a',
309
StringIO('contents\n'))
310
self.failIf(t.has('dir/a'))
311
t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
312
create_parent_dir=True)
313
self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
314
315
# But we still get NoSuchFile if we can't make the parent dir
316
self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'not/there/a',
317
StringIO('contents\n'),
318
create_parent_dir=True)
319
320
def test_put_file_permissions(self):
321
322
t = self.get_transport()
323
324
if t.is_readonly():
325
return
326
if not t._can_roundtrip_unix_modebits():
327
# Can't roundtrip, so no need to run this test
328
return
329
t.put_file('mode644', StringIO('test text\n'), mode=0644)
330
self.assertTransportMode(t, 'mode644', 0644)
331
t.put_file('mode666', StringIO('test text\n'), mode=0666)
332
self.assertTransportMode(t, 'mode666', 0666)
333
t.put_file('mode600', StringIO('test text\n'), mode=0600)
334
self.assertTransportMode(t, 'mode600', 0600)
335
# Yes, you can put a file such that it becomes readonly
336
t.put_file('mode400', StringIO('test text\n'), mode=0400)
337
self.assertTransportMode(t, 'mode400', 0400)
338
339
# XXX: put_multi is deprecated, so do we really care anymore?
340
self.applyDeprecated(zero_eleven, t.put_multi,
341
[('mmode644', StringIO('text\n'))], mode=0644)
342
self.assertTransportMode(t, 'mmode644', 0644)
343
344
# The default permissions should be based on the current umask
345
umask = osutils.get_umask()
346
t.put_file('nomode', StringIO('test text\n'), mode=None)
347
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
348
349
def test_put_file_non_atomic_permissions(self):
350
t = self.get_transport()
351
352
if t.is_readonly():
353
return
354
if not t._can_roundtrip_unix_modebits():
355
# Can't roundtrip, so no need to run this test
356
return
357
t.put_file_non_atomic('mode644', StringIO('test text\n'), mode=0644)
358
self.assertTransportMode(t, 'mode644', 0644)
359
t.put_file_non_atomic('mode666', StringIO('test text\n'), mode=0666)
360
self.assertTransportMode(t, 'mode666', 0666)
361
t.put_file_non_atomic('mode600', StringIO('test text\n'), mode=0600)
362
self.assertTransportMode(t, 'mode600', 0600)
363
# Yes, you can put_file_non_atomic a file such that it becomes readonly
364
t.put_file_non_atomic('mode400', StringIO('test text\n'), mode=0400)
365
self.assertTransportMode(t, 'mode400', 0400)
366
367
# The default permissions should be based on the current umask
368
umask = osutils.get_umask()
369
t.put_file_non_atomic('nomode', StringIO('test text\n'), mode=None)
370
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
371
372
# We should also be able to set the mode for a parent directory
373
# when it is created
374
sio = StringIO()
375
t.put_file_non_atomic('dir700/mode664', sio, mode=0664,
376
dir_mode=0700, create_parent_dir=True)
377
self.assertTransportMode(t, 'dir700', 0700)
378
t.put_file_non_atomic('dir770/mode664', sio, mode=0664,
379
dir_mode=0770, create_parent_dir=True)
380
self.assertTransportMode(t, 'dir770', 0770)
381
t.put_file_non_atomic('dir777/mode664', sio, mode=0664,
382
dir_mode=0777, create_parent_dir=True)
383
self.assertTransportMode(t, 'dir777', 0777)
384
385
def test_put_multi(self):
386
t = self.get_transport()
387
388
if t.is_readonly():
389
return
390
self.assertEqual(2, self.applyDeprecated(zero_eleven,
391
t.put_multi, [('a', StringIO('new\ncontents for\na\n')),
392
('d', StringIO('contents\nfor d\n'))]
393
))
394
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd'])),
395
[True, False, False, True])
134
396
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
135
397
self.check_transport_contents('contents\nfor d\n', t, 'd')
136
398
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)
399
self.assertEqual(2, self.applyDeprecated(zero_eleven,
400
t.put_multi, iter([('a', StringIO('diff\ncontents for\na\n')),
401
('d', StringIO('another contents\nfor d\n'))])
402
))
141
403
self.check_transport_contents('diff\ncontents for\na\n', t, 'a')
142
404
self.check_transport_contents('another contents\nfor d\n', t, 'd')
143
405
144
self.assertRaises(NoSuchFile,
145
t.put, 'path/doesnt/exist/c', StringIO('contents'))
146
147
406
def test_put_permissions(self):
148
407
t = self.get_transport()
149
408
152
411
if not t._can_roundtrip_unix_modebits():
153
412
# Can't roundtrip, so no need to run this test
154
413
return
155
t.put('mode644', StringIO('test text\n'), mode=0644)
414
self.applyDeprecated(zero_eleven, t.put, 'mode644',
415
StringIO('test text\n'), mode=0644)
156
416
self.assertTransportMode(t, 'mode644', 0644)
157
t.put('mode666', StringIO('test text\n'), mode=0666)
417
self.applyDeprecated(zero_eleven, t.put, 'mode666',
418
StringIO('test text\n'), mode=0666)
158
419
self.assertTransportMode(t, 'mode666', 0666)
159
t.put('mode600', StringIO('test text\n'), mode=0600)
420
self.applyDeprecated(zero_eleven, t.put, 'mode600',
421
StringIO('test text\n'), mode=0600)
160
422
self.assertTransportMode(t, 'mode600', 0600)
161
423
# Yes, you can put a file such that it becomes readonly
162
t.put('mode400', StringIO('test text\n'), mode=0400)
424
self.applyDeprecated(zero_eleven, t.put, 'mode400',
425
StringIO('test text\n'), mode=0400)
163
426
self.assertTransportMode(t, 'mode400', 0400)
164
t.put_multi([('mmode644', StringIO('text\n'))], mode=0644)
427
self.applyDeprecated(zero_eleven, t.put_multi,
428
[('mmode644', StringIO('text\n'))], mode=0644)
165
429
self.assertTransportMode(t, 'mmode644', 0644)
166
430
167
431
# The default permissions should be based on the current umask
168
432
umask = osutils.get_umask()
169
t.put('nomode', StringIO('test text\n'), mode=None)
433
self.applyDeprecated(zero_eleven, t.put, 'nomode',
434
StringIO('test text\n'), mode=None)
170
435
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
171
436
172
437
def test_mkdir(self):
207
472
self.assertRaises(FileExists, t.mkdir, 'dir_g')
208
473
209
474
# 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'))]))
475
t.put_bytes('dir_a/a', 'contents of dir_a/a')
476
t.put_file('dir_b/b', StringIO('contents of dir_b/b'))
213
477
self.check_transport_contents('contents of dir_a/a', t, 'dir_a/a')
214
478
self.check_transport_contents('contents of dir_b/b', t, 'dir_b/b')
215
479
270
534
self.build_tree(['e/', 'e/f'])
271
535
else:
272
536
t.mkdir('e')
273
t.put('e/f', StringIO('contents of e'))
537
t.put_bytes('e/f', 'contents of e')
274
538
self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], temp_transport)
275
539
temp_transport.mkdir('e')
276
540
t.copy_to(['e/f'], temp_transport)
295
559
t = self.get_transport()
296
560
297
561
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'))]))
562
return
563
t.put_bytes('a', 'diff\ncontents for\na\n')
564
t.put_bytes('b', 'contents\nfor b\n')
565
566
self.assertEqual(20, self.applyDeprecated(zero_eleven,
567
t.append, 'a', StringIO('add\nsome\nmore\ncontents\n')))
568
569
self.check_transport_contents(
570
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
571
t, 'a')
572
573
# And we can create new files, too
574
self.assertEqual(0, self.applyDeprecated(zero_eleven,
575
t.append, 'c', StringIO('some text\nfor a missing file\n')))
576
self.check_transport_contents('some text\nfor a missing file\n',
577
t, 'c')
578
def test_append_file(self):
579
t = self.get_transport()
580
581
if t.is_readonly():
582
self.assertRaises(TransportNotPossible,
583
t.append_file, 'a', 'add\nsome\nmore\ncontents\n')
584
return
585
t.put_bytes('a', 'diff\ncontents for\na\n')
586
t.put_bytes('b', 'contents\nfor b\n')
587
588
self.assertEqual(20,
589
t.append_file('a', StringIO('add\nsome\nmore\ncontents\n')))
590
591
self.check_transport_contents(
592
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
593
t, 'a')
594
595
# a file with no parent should fail..
596
self.assertRaises(NoSuchFile,
597
t.append_file, 'missing/path', StringIO('content'))
598
599
# And we can create new files, too
600
self.assertEqual(0,
601
t.append_file('c', StringIO('some text\nfor a missing file\n')))
602
self.check_transport_contents('some text\nfor a missing file\n',
603
t, 'c')
604
605
def test_append_bytes(self):
606
t = self.get_transport()
607
608
if t.is_readonly():
609
self.assertRaises(TransportNotPossible,
610
t.append_bytes, 'a', 'add\nsome\nmore\ncontents\n')
611
return
612
613
self.assertEqual(0, t.append_bytes('a', 'diff\ncontents for\na\n'))
614
self.assertEqual(0, t.append_bytes('b', 'contents\nfor b\n'))
615
616
self.assertEqual(20,
617
t.append_bytes('a', 'add\nsome\nmore\ncontents\n'))
618
619
self.check_transport_contents(
620
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
621
t, 'a')
622
623
# a file with no parent should fail..
624
self.assertRaises(NoSuchFile,
625
t.append_bytes, 'missing/path', 'content')
626
627
def test_append_multi(self):
628
t = self.get_transport()
629
630
if t.is_readonly():
631
return
632
t.put_bytes('a', 'diff\ncontents for\na\n'
633
'add\nsome\nmore\ncontents\n')
634
t.put_bytes('b', 'contents\nfor b\n')
635
636
self.assertEqual((43, 15),
637
t.append_multi([('a', StringIO('and\nthen\nsome\nmore\n')),
638
('b', StringIO('some\nmore\nfor\nb\n'))]))
639
329
640
self.check_transport_contents(
330
641
'diff\ncontents for\na\n'
331
642
'add\nsome\nmore\ncontents\n'
336
647
'some\nmore\nfor\nb\n',
337
648
t, 'b')
338
649
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'))])))
650
self.assertEqual((62, 31),
651
t.append_multi(iter([('a', StringIO('a little bit more\n')),
652
('b', StringIO('from an iterator\n'))])))
346
653
self.check_transport_contents(
347
654
'diff\ncontents for\na\n'
348
655
'add\nsome\nmore\ncontents\n'
355
662
'from an iterator\n',
356
663
t, 'b')
357
664
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'))]))
665
self.assertEqual((80, 0),
666
t.append_multi([('a', StringIO('some text in a\n')),
667
('d', StringIO('missing file r\n'))]))
668
368
669
self.check_transport_contents(
369
670
'diff\ncontents for\na\n'
370
671
'add\nsome\nmore\ncontents\n'
372
673
'a little bit more\n'
373
674
'some text in a\n',
374
675
t, 'a')
375
self.check_transport_contents('some text\nfor a missing file\n',
376
t, 'c')
377
676
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):
677
678
def test_append_file_mode(self):
679
"""Check that append accepts a mode parameter"""
487
680
# check append accepts a mode
488
681
t = self.get_transport()
489
682
if t.is_readonly():
490
return
491
t.append('f', StringIO('f'), mode=None)
683
self.assertRaises(TransportNotPossible,
684
t.append_file, 'f', StringIO('f'), mode=None)
685
return
686
t.append_file('f', StringIO('f'), mode=None)
687
688
def test_append_bytes_mode(self):
689
# check append_bytes accepts a mode
690
t = self.get_transport()
691
if t.is_readonly():
692
self.assertRaises(TransportNotPossible,
693
t.append_bytes, 'f', 'f', mode=None)
694
return
695
t.append_bytes('f', 'f', mode=None)
492
696
493
697
def test_delete(self):
494
698
# TODO: Test Transport.delete
499
703
self.assertRaises(TransportNotPossible, t.delete, 'missing')
500
704
return
501
705
502
t.put('a', StringIO('a little bit of text\n'))
706
t.put_bytes('a', 'a little bit of text\n')
503
707
self.failUnless(t.has('a'))
504
708
t.delete('a')
505
709
self.failIf(t.has('a'))
506
710
507
711
self.assertRaises(NoSuchFile, t.delete, 'a')
508
712
509
t.put('a', StringIO('a text\n'))
510
t.put('b', StringIO('b text\n'))
511
t.put('c', StringIO('c text\n'))
713
t.put_bytes('a', 'a text\n')
714
t.put_bytes('b', 'b text\n')
715
t.put_bytes('c', 'c text\n')
512
716
self.assertEqual([True, True, True],
513
717
list(t.has_multi(['a', 'b', 'c'])))
514
718
t.delete_multi(['a', 'c'])
524
728
self.assertRaises(NoSuchFile,
525
729
t.delete_multi, iter(['a', 'b', 'c']))
526
730
527
t.put('a', StringIO('another a text\n'))
528
t.put('c', StringIO('another c text\n'))
731
t.put_bytes('a', 'another a text\n')
732
t.put_bytes('c', 'another c text\n')
529
733
t.delete_multi(iter(['a', 'b', 'c']))
530
734
531
735
# We should have deleted everything
543
747
t.mkdir('adir')
544
748
t.mkdir('adir/bdir')
545
749
t.rmdir('adir/bdir')
546
self.assertFalse(t.has('adir/bdir'))
750
# ftp may not be able to raise NoSuchFile for lack of
751
# details when failing
752
self.assertRaises((NoSuchFile, PathError), t.rmdir, 'adir/bdir')
547
753
t.rmdir('adir')
548
self.assertFalse(t.has('adir'))
754
self.assertRaises((NoSuchFile, PathError), t.rmdir, 'adir')
549
755
550
756
def test_rmdir_not_empty(self):
551
757
"""Deleting a non-empty directory raises an exception
629
835
# creates control files in the working directory
630
836
# perhaps all of this could be done in a subdirectory
631
837
632
t.put('a', StringIO('a first file\n'))
838
t.put_bytes('a', 'a first file\n')
633
839
self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
634
840
635
841
t.move('a', 'b')
640
846
self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
641
847
642
848
# Overwrite a file
643
t.put('c', StringIO('c this file\n'))
849
t.put_bytes('c', 'c this file\n')
644
850
t.move('c', 'b')
645
851
self.failIf(t.has('c'))
646
852
self.check_transport_contents('c this file\n', t, 'b')
655
861
if t.is_readonly():
656
862
return
657
863
658
t.put('a', StringIO('a file\n'))
864
t.put_bytes('a', 'a file\n')
659
865
t.copy('a', 'b')
660
866
self.check_transport_contents('a file\n', t, 'b')
661
867
664
870
# What should the assert be if you try to copy a
665
871
# file over a directory?
666
872
#self.assertRaises(Something, t.copy, 'a', 'c')
667
t.put('d', StringIO('text in d\n'))
873
t.put_bytes('d', 'text in d\n')
668
874
t.copy('d', 'b')
669
875
self.check_transport_contents('text in d\n', t, 'b')
670
876
820
1026
if t1.is_readonly():
821
1027
open('b/d', 'wb').write('newfile\n')
822
1028
else:
823
t2.put('d', StringIO('newfile\n'))
1029
t2.put_bytes('d', 'newfile\n')
824
1030
825
1031
self.failUnless(t1.has('b/d'))
826
1032
self.failUnless(t2.has('d'))
951
1157
transport.iter_files_recursive)
952
1158
return
953
1159
if transport.is_readonly():
954
self.assertRaises(TransportNotPossible,
955
transport.put, 'a', 'some text for a\n')
956
1160
return
957
1161
self.build_tree(['from/',
958
1162
'from/dir/',
1009
1213
transport = self.get_transport()
1010
1214
if transport.is_readonly():
1011
1215
return
1012
transport.put('foo', StringIO('bar'))
1216
transport.put_bytes('foo', 'bar')
1013
1217
transport2 = self.get_transport()
1014
1218
self.check_transport_contents('bar', transport2, 'foo')
1015
1219
# its base should be usable.
1027
1231
if transport.is_readonly():
1028
1232
self.assertRaises(TransportNotPossible, transport.lock_write, 'foo')
1029
1233
return
1030
transport.put('lock', StringIO())
1234
transport.put_bytes('lock', '')
1031
1235
lock = transport.lock_write('lock')
1032
1236
# TODO make this consistent on all platforms:
1033
1237
# self.assertRaises(LockError, transport.lock_write, 'lock')
1038
1242
if transport.is_readonly():
1039
1243
file('lock', 'w').close()
1040
1244
else:
1041
transport.put('lock', StringIO())
1245
transport.put_bytes('lock', '')
1042
1246
lock = transport.lock_read('lock')
1043
1247
# TODO make this consistent on all platforms:
1044
1248
# self.assertRaises(LockError, transport.lock_read, 'lock')
1049
1253
if transport.is_readonly():
1050
1254
file('a', 'w').write('0123456789')
1051
1255
else:
1052
transport.put('a', StringIO('0123456789'))
1256
transport.put_bytes('a', '0123456789')
1053
1257
1054
1258
d = list(transport.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
1055
1259
self.assertEqual(d[0], (0, '0'))
1062
1266
if transport.is_readonly():
1063
1267
file('a', 'w').write('0123456789')
1064
1268
else:
1065
transport.put('a', StringIO('01234567890'))
1269
transport.put_bytes('a', '01234567890')
1066
1270
1067
1271
d = list(transport.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
1068
1272
self.assertEqual(d[0], (1, '1'))
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0