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 1907.4.14
- compare with another revision
- compare with revision 1036
- download diff
- download tarball
- view history from revision 1907.4.14
- Committer: Matthieu Moy
- Date: 2006-09-11 09:34:51 UTC
- mfrom: (1997 +trunk)
- mto: (1998.1.1 jam-integration)
- mto: This revision was merged to the branch mainline in revision 1999.
- Revision ID: Matthieu.Moy@imag.fr-20060911093451-24ef653833ff7477
merge
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
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])
134
393
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
135
394
self.check_transport_contents('contents\nfor d\n', t, 'd')
136
395
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)
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
))
141
400
self.check_transport_contents('diff\ncontents for\na\n', t, 'a')
142
401
self.check_transport_contents('another contents\nfor d\n', t, 'd')
143
402
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
403
def test_mkdir(self):
173
404
t = self.get_transport()
174
405
207
438
self.assertRaises(FileExists, t.mkdir, 'dir_g')
208
439
209
440
# 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'))]))
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'))
213
443
self.check_transport_contents('contents of dir_a/a', t, 'dir_a/a')
214
444
self.check_transport_contents('contents of dir_b/b', t, 'dir_b/b')
215
445
270
500
self.build_tree(['e/', 'e/f'])
271
501
else:
272
502
t.mkdir('e')
273
t.put('e/f', StringIO('contents of e'))
503
t.put_bytes('e/f', 'contents of e')
274
504
self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], temp_transport)
275
505
temp_transport.mkdir('e')
276
506
t.copy_to(['e/f'], temp_transport)
295
525
t = self.get_transport()
296
526
297
527
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'))]))
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
329
606
self.check_transport_contents(
330
607
'diff\ncontents for\na\n'
331
608
'add\nsome\nmore\ncontents\n'
336
613
'some\nmore\nfor\nb\n',
337
614
t, 'b')
338
615
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'))])))
616
self.assertEqual((62, 31),
617
t.append_multi(iter([('a', StringIO('a little bit more\n')),
618
('b', StringIO('from an iterator\n'))])))
346
619
self.check_transport_contents(
347
620
'diff\ncontents for\na\n'
348
621
'add\nsome\nmore\ncontents\n'
355
628
'from an iterator\n',
356
629
t, 'b')
357
630
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'))]))
631
self.assertEqual((80, 0),
632
t.append_multi([('a', StringIO('some text in a\n')),
633
('d', StringIO('missing file r\n'))]))
634
368
635
self.check_transport_contents(
369
636
'diff\ncontents for\na\n'
370
637
'add\nsome\nmore\ncontents\n'
372
639
'a little bit more\n'
373
640
'some text in a\n',
374
641
t, 'a')
375
self.check_transport_contents('some text\nfor a missing file\n',
376
t, 'c')
377
642
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):
643
644
def test_append_file_mode(self):
645
"""Check that append accepts a mode parameter"""
487
646
# check append accepts a mode
488
647
t = self.get_transport()
489
648
if t.is_readonly():
490
return
491
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)
492
662
493
663
def test_delete(self):
494
664
# TODO: Test Transport.delete
499
669
self.assertRaises(TransportNotPossible, t.delete, 'missing')
500
670
return
501
671
502
t.put('a', StringIO('a little bit of text\n'))
672
t.put_bytes('a', 'a little bit of text\n')
503
673
self.failUnless(t.has('a'))
504
674
t.delete('a')
505
675
self.failIf(t.has('a'))
506
676
507
677
self.assertRaises(NoSuchFile, t.delete, 'a')
508
678
509
t.put('a', StringIO('a text\n'))
510
t.put('b', StringIO('b text\n'))
511
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')
512
682
self.assertEqual([True, True, True],
513
683
list(t.has_multi(['a', 'b', 'c'])))
514
684
t.delete_multi(['a', 'c'])
524
694
self.assertRaises(NoSuchFile,
525
695
t.delete_multi, iter(['a', 'b', 'c']))
526
696
527
t.put('a', StringIO('another a text\n'))
528
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')
529
699
t.delete_multi(iter(['a', 'b', 'c']))
530
700
531
701
# We should have deleted everything
629
799
# creates control files in the working directory
630
800
# perhaps all of this could be done in a subdirectory
631
801
632
t.put('a', StringIO('a first file\n'))
802
t.put_bytes('a', 'a first file\n')
633
803
self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
634
804
635
805
t.move('a', 'b')
640
810
self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
641
811
642
812
# Overwrite a file
643
t.put('c', StringIO('c this file\n'))
813
t.put_bytes('c', 'c this file\n')
644
814
t.move('c', 'b')
645
815
self.failIf(t.has('c'))
646
816
self.check_transport_contents('c this file\n', t, 'b')
655
825
if t.is_readonly():
656
826
return
657
827
658
t.put('a', StringIO('a file\n'))
828
t.put_bytes('a', 'a file\n')
659
829
t.copy('a', 'b')
660
830
self.check_transport_contents('a file\n', t, 'b')
661
831
664
834
# What should the assert be if you try to copy a
665
835
# file over a directory?
666
836
#self.assertRaises(Something, t.copy, 'a', 'c')
667
t.put('d', StringIO('text in d\n'))
837
t.put_bytes('d', 'text in d\n')
668
838
t.copy('d', 'b')
669
839
self.check_transport_contents('text in d\n', t, 'b')
670
840
816
986
if t1.is_readonly():
817
987
open('b/d', 'wb').write('newfile\n')
818
988
else:
819
t2.put('d', StringIO('newfile\n'))
989
t2.put_bytes('d', 'newfile\n')
820
990
821
991
self.failUnless(t1.has('b/d'))
822
992
self.failUnless(t2.has('d'))
916
1086
transport.iter_files_recursive)
917
1087
return
918
1088
if transport.is_readonly():
919
self.assertRaises(TransportNotPossible,
920
transport.put, 'a', 'some text for a\n')
921
1089
return
922
1090
self.build_tree(['from/',
923
1091
'from/dir/',
974
1142
transport = self.get_transport()
975
1143
if transport.is_readonly():
976
1144
return
977
transport.put('foo', StringIO('bar'))
1145
transport.put_bytes('foo', 'bar')
978
1146
transport2 = self.get_transport()
979
1147
self.check_transport_contents('bar', transport2, 'foo')
980
1148
# its base should be usable.
992
1160
if transport.is_readonly():
993
1161
self.assertRaises(TransportNotPossible, transport.lock_write, 'foo')
994
1162
return
995
transport.put('lock', StringIO())
1163
transport.put_bytes('lock', '')
996
1164
lock = transport.lock_write('lock')
997
1165
# TODO make this consistent on all platforms:
998
1166
# self.assertRaises(LockError, transport.lock_write, 'lock')
1003
1171
if transport.is_readonly():
1004
1172
file('lock', 'w').close()
1005
1173
else:
1006
transport.put('lock', StringIO())
1174
transport.put_bytes('lock', '')
1007
1175
lock = transport.lock_read('lock')
1008
1176
# TODO make this consistent on all platforms:
1009
1177
# self.assertRaises(LockError, transport.lock_read, 'lock')
1014
1182
if transport.is_readonly():
1015
1183
file('a', 'w').write('0123456789')
1016
1184
else:
1017
transport.put('a', StringIO('0123456789'))
1185
transport.put_bytes('a', '0123456789')
1018
1186
1019
1187
d = list(transport.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
1020
1188
self.assertEqual(d[0], (0, '0'))
1027
1195
if transport.is_readonly():
1028
1196
file('a', 'w').write('0123456789')
1029
1197
else:
1030
transport.put('a', StringIO('01234567890'))
1198
transport.put_bytes('a', '01234567890')
1031
1199
1032
1200
d = list(transport.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
1033
1201
self.assertEqual(d[0], (1, '1'))
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0