« back to all changes in this revision
Viewing changes to bzrlib/tests/test_transport_implementations.py
- browse files at revision 1908.11.1
- compare with another revision
- compare with revision 5807.4.5
- download diff
- download tarball
- view history from revision 1908.11.1
- Committer: Robert Collins
- Date: 2006-09-08 01:58:47 UTC
- mfrom: (1993 +trunk)
- mto: (1852.16.2 Tree.walkdirs)
- mto: This revision was merged to the branch mainline in revision 2322.
- Revision ID: robertc@robertcollins.net-20060908015847-69da178f0ec194d4
Add a new method ``Tree.revision_tree`` which allows access to cached
trees for arbitrary revisions. This allows the in development dirstate
tree format to provide access to the callers to cached copies of
inventory data which are cheaper to access than inventories from the
repository. (Robert Collins, Martin Pool)
trees for arbitrary revisions. This allows the in development dirstate
tree format to provide access to the callers to cached copies of
inventory data which are cheaper to access than inventories from the
repository. (Robert Collins, Martin Pool)
- 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
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
# We should also be able to set the mode for a parent directory
248
# when it is created
249
t.put_bytes_non_atomic('dir700/mode664', 'test text\n', mode=0664,
250
dir_mode=0700, create_parent_dir=True)
251
self.assertTransportMode(t, 'dir700', 0700)
252
t.put_bytes_non_atomic('dir770/mode664', 'test text\n', mode=0664,
253
dir_mode=0770, create_parent_dir=True)
254
self.assertTransportMode(t, 'dir770', 0770)
255
t.put_bytes_non_atomic('dir777/mode664', 'test text\n', mode=0664,
256
dir_mode=0777, create_parent_dir=True)
257
self.assertTransportMode(t, 'dir777', 0777)
258
259
def test_put_file(self):
260
t = self.get_transport()
261
262
if t.is_readonly():
263
self.assertRaises(TransportNotPossible,
264
t.put_file, 'a', StringIO('some text for a\n'))
265
return
266
267
t.put_file('a', StringIO('some text for a\n'))
268
self.failUnless(t.has('a'))
269
self.check_transport_contents('some text for a\n', t, 'a')
270
# Put also replaces contents
271
t.put_file('a', StringIO('new\ncontents for\na\n'))
272
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
273
self.assertRaises(NoSuchFile,
274
t.put_file, 'path/doesnt/exist/c',
275
StringIO('contents'))
276
277
def test_put_file_non_atomic(self):
278
t = self.get_transport()
279
280
if t.is_readonly():
281
self.assertRaises(TransportNotPossible,
282
t.put_file_non_atomic, 'a', StringIO('some text for a\n'))
283
return
284
285
self.failIf(t.has('a'))
286
t.put_file_non_atomic('a', StringIO('some text for a\n'))
287
self.failUnless(t.has('a'))
288
self.check_transport_contents('some text for a\n', t, 'a')
289
# Put also replaces contents
290
t.put_file_non_atomic('a', StringIO('new\ncontents for\na\n'))
291
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
292
293
# Make sure we can create another file
294
t.put_file_non_atomic('d', StringIO('contents for\nd\n'))
295
# And overwrite 'a' with empty contents
296
t.put_file_non_atomic('a', StringIO(''))
297
self.check_transport_contents('contents for\nd\n', t, 'd')
298
self.check_transport_contents('', t, 'a')
299
300
self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'no/such/path',
301
StringIO('contents\n'))
302
# Now test the create_parent flag
303
self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'dir/a',
304
StringIO('contents\n'))
305
self.failIf(t.has('dir/a'))
306
t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
307
create_parent_dir=True)
308
self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
309
310
# But we still get NoSuchFile if we can't make the parent dir
311
self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'not/there/a',
312
StringIO('contents\n'),
313
create_parent_dir=True)
314
315
def test_put_file_permissions(self):
316
317
t = self.get_transport()
318
319
if t.is_readonly():
320
return
321
if not t._can_roundtrip_unix_modebits():
322
# Can't roundtrip, so no need to run this test
323
return
324
t.put_file('mode644', StringIO('test text\n'), mode=0644)
325
self.assertTransportMode(t, 'mode644', 0644)
326
t.put_file('mode666', StringIO('test text\n'), mode=0666)
327
self.assertTransportMode(t, 'mode666', 0666)
328
t.put_file('mode600', StringIO('test text\n'), mode=0600)
329
self.assertTransportMode(t, 'mode600', 0600)
330
# Yes, you can put a file such that it becomes readonly
331
t.put_file('mode400', StringIO('test text\n'), mode=0400)
332
self.assertTransportMode(t, 'mode400', 0400)
333
334
# XXX: put_multi is deprecated, so do we really care anymore?
335
self.applyDeprecated(zero_eleven, t.put_multi,
336
[('mmode644', StringIO('text\n'))], mode=0644)
337
self.assertTransportMode(t, 'mmode644', 0644)
338
339
# The default permissions should be based on the current umask
340
umask = osutils.get_umask()
341
t.put_file('nomode', StringIO('test text\n'), mode=None)
342
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
343
344
def test_put_file_non_atomic_permissions(self):
345
t = self.get_transport()
346
347
if t.is_readonly():
348
return
349
if not t._can_roundtrip_unix_modebits():
350
# Can't roundtrip, so no need to run this test
351
return
352
t.put_file_non_atomic('mode644', StringIO('test text\n'), mode=0644)
353
self.assertTransportMode(t, 'mode644', 0644)
354
t.put_file_non_atomic('mode666', StringIO('test text\n'), mode=0666)
355
self.assertTransportMode(t, 'mode666', 0666)
356
t.put_file_non_atomic('mode600', StringIO('test text\n'), mode=0600)
357
self.assertTransportMode(t, 'mode600', 0600)
358
# Yes, you can put_file_non_atomic a file such that it becomes readonly
359
t.put_file_non_atomic('mode400', StringIO('test text\n'), mode=0400)
360
self.assertTransportMode(t, 'mode400', 0400)
361
362
# The default permissions should be based on the current umask
363
umask = osutils.get_umask()
364
t.put_file_non_atomic('nomode', StringIO('test text\n'), mode=None)
365
self.assertTransportMode(t, 'nomode', 0666 & ~umask)
366
367
# We should also be able to set the mode for a parent directory
368
# when it is created
369
sio = StringIO()
370
t.put_file_non_atomic('dir700/mode664', sio, mode=0664,
371
dir_mode=0700, create_parent_dir=True)
372
self.assertTransportMode(t, 'dir700', 0700)
373
t.put_file_non_atomic('dir770/mode664', sio, mode=0664,
374
dir_mode=0770, create_parent_dir=True)
375
self.assertTransportMode(t, 'dir770', 0770)
376
t.put_file_non_atomic('dir777/mode664', sio, mode=0664,
377
dir_mode=0777, create_parent_dir=True)
378
self.assertTransportMode(t, 'dir777', 0777)
379
380
def test_put_multi(self):
381
t = self.get_transport()
382
383
if t.is_readonly():
384
return
385
self.assertEqual(2, self.applyDeprecated(zero_eleven,
386
t.put_multi, [('a', StringIO('new\ncontents for\na\n')),
387
('d', StringIO('contents\nfor d\n'))]
388
))
389
self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd'])),
390
[True, False, False, True])
134
391
self.check_transport_contents('new\ncontents for\na\n', t, 'a')
135
392
self.check_transport_contents('contents\nfor d\n', t, 'd')
136
393
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)
394
self.assertEqual(2, self.applyDeprecated(zero_eleven,
395
t.put_multi, iter([('a', StringIO('diff\ncontents for\na\n')),
396
('d', StringIO('another contents\nfor d\n'))])
397
))
141
398
self.check_transport_contents('diff\ncontents for\na\n', t, 'a')
142
399
self.check_transport_contents('another contents\nfor d\n', t, 'd')
143
400
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
401
def test_mkdir(self):
173
402
t = self.get_transport()
174
403
207
436
self.assertRaises(FileExists, t.mkdir, 'dir_g')
208
437
209
438
# 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'))]))
439
t.put_bytes('dir_a/a', 'contents of dir_a/a')
440
t.put_file('dir_b/b', StringIO('contents of dir_b/b'))
213
441
self.check_transport_contents('contents of dir_a/a', t, 'dir_a/a')
214
442
self.check_transport_contents('contents of dir_b/b', t, 'dir_b/b')
215
443
270
498
self.build_tree(['e/', 'e/f'])
271
499
else:
272
500
t.mkdir('e')
273
t.put('e/f', StringIO('contents of e'))
501
t.put_bytes('e/f', 'contents of e')
274
502
self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], temp_transport)
275
503
temp_transport.mkdir('e')
276
504
t.copy_to(['e/f'], temp_transport)
295
523
t = self.get_transport()
296
524
297
525
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'))]))
526
return
527
t.put_bytes('a', 'diff\ncontents for\na\n')
528
t.put_bytes('b', 'contents\nfor b\n')
529
530
self.assertEqual(20, self.applyDeprecated(zero_eleven,
531
t.append, 'a', StringIO('add\nsome\nmore\ncontents\n')))
532
533
self.check_transport_contents(
534
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
535
t, 'a')
536
537
# And we can create new files, too
538
self.assertEqual(0, self.applyDeprecated(zero_eleven,
539
t.append, 'c', StringIO('some text\nfor a missing file\n')))
540
self.check_transport_contents('some text\nfor a missing file\n',
541
t, 'c')
542
def test_append_file(self):
543
t = self.get_transport()
544
545
if t.is_readonly():
546
self.assertRaises(TransportNotPossible,
547
t.append_file, 'a', 'add\nsome\nmore\ncontents\n')
548
return
549
t.put_bytes('a', 'diff\ncontents for\na\n')
550
t.put_bytes('b', 'contents\nfor b\n')
551
552
self.assertEqual(20,
553
t.append_file('a', StringIO('add\nsome\nmore\ncontents\n')))
554
555
self.check_transport_contents(
556
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
557
t, 'a')
558
559
# a file with no parent should fail..
560
self.assertRaises(NoSuchFile,
561
t.append_file, 'missing/path', StringIO('content'))
562
563
# And we can create new files, too
564
self.assertEqual(0,
565
t.append_file('c', StringIO('some text\nfor a missing file\n')))
566
self.check_transport_contents('some text\nfor a missing file\n',
567
t, 'c')
568
569
def test_append_bytes(self):
570
t = self.get_transport()
571
572
if t.is_readonly():
573
self.assertRaises(TransportNotPossible,
574
t.append_bytes, 'a', 'add\nsome\nmore\ncontents\n')
575
return
576
577
self.assertEqual(0, t.append_bytes('a', 'diff\ncontents for\na\n'))
578
self.assertEqual(0, t.append_bytes('b', 'contents\nfor b\n'))
579
580
self.assertEqual(20,
581
t.append_bytes('a', 'add\nsome\nmore\ncontents\n'))
582
583
self.check_transport_contents(
584
'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',
585
t, 'a')
586
587
# a file with no parent should fail..
588
self.assertRaises(NoSuchFile,
589
t.append_bytes, 'missing/path', 'content')
590
591
def test_append_multi(self):
592
t = self.get_transport()
593
594
if t.is_readonly():
595
return
596
t.put_bytes('a', 'diff\ncontents for\na\n'
597
'add\nsome\nmore\ncontents\n')
598
t.put_bytes('b', 'contents\nfor b\n')
599
600
self.assertEqual((43, 15),
601
t.append_multi([('a', StringIO('and\nthen\nsome\nmore\n')),
602
('b', StringIO('some\nmore\nfor\nb\n'))]))
603
329
604
self.check_transport_contents(
330
605
'diff\ncontents for\na\n'
331
606
'add\nsome\nmore\ncontents\n'
336
611
'some\nmore\nfor\nb\n',
337
612
t, 'b')
338
613
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'))])))
614
self.assertEqual((62, 31),
615
t.append_multi(iter([('a', StringIO('a little bit more\n')),
616
('b', StringIO('from an iterator\n'))])))
346
617
self.check_transport_contents(
347
618
'diff\ncontents for\na\n'
348
619
'add\nsome\nmore\ncontents\n'
355
626
'from an iterator\n',
356
627
t, 'b')
357
628
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'))]))
629
self.assertEqual((80, 0),
630
t.append_multi([('a', StringIO('some text in a\n')),
631
('d', StringIO('missing file r\n'))]))
632
368
633
self.check_transport_contents(
369
634
'diff\ncontents for\na\n'
370
635
'add\nsome\nmore\ncontents\n'
372
637
'a little bit more\n'
373
638
'some text in a\n',
374
639
t, 'a')
375
self.check_transport_contents('some text\nfor a missing file\n',
376
t, 'c')
377
640
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):
641
642
def test_append_file_mode(self):
643
"""Check that append accepts a mode parameter"""
487
644
# check append accepts a mode
488
645
t = self.get_transport()
489
646
if t.is_readonly():
490
return
491
t.append('f', StringIO('f'), mode=None)
647
self.assertRaises(TransportNotPossible,
648
t.append_file, 'f', StringIO('f'), mode=None)
649
return
650
t.append_file('f', StringIO('f'), mode=None)
651
652
def test_append_bytes_mode(self):
653
# check append_bytes accepts a mode
654
t = self.get_transport()
655
if t.is_readonly():
656
self.assertRaises(TransportNotPossible,
657
t.append_bytes, 'f', 'f', mode=None)
658
return
659
t.append_bytes('f', 'f', mode=None)
492
660
493
661
def test_delete(self):
494
662
# TODO: Test Transport.delete
499
667
self.assertRaises(TransportNotPossible, t.delete, 'missing')
500
668
return
501
669
502
t.put('a', StringIO('a little bit of text\n'))
670
t.put_bytes('a', 'a little bit of text\n')
503
671
self.failUnless(t.has('a'))
504
672
t.delete('a')
505
673
self.failIf(t.has('a'))
506
674
507
675
self.assertRaises(NoSuchFile, t.delete, 'a')
508
676
509
t.put('a', StringIO('a text\n'))
510
t.put('b', StringIO('b text\n'))
511
t.put('c', StringIO('c text\n'))
677
t.put_bytes('a', 'a text\n')
678
t.put_bytes('b', 'b text\n')
679
t.put_bytes('c', 'c text\n')
512
680
self.assertEqual([True, True, True],
513
681
list(t.has_multi(['a', 'b', 'c'])))
514
682
t.delete_multi(['a', 'c'])
524
692
self.assertRaises(NoSuchFile,
525
693
t.delete_multi, iter(['a', 'b', 'c']))
526
694
527
t.put('a', StringIO('another a text\n'))
528
t.put('c', StringIO('another c text\n'))
695
t.put_bytes('a', 'another a text\n')
696
t.put_bytes('c', 'another c text\n')
529
697
t.delete_multi(iter(['a', 'b', 'c']))
530
698
531
699
# We should have deleted everything
629
797
# creates control files in the working directory
630
798
# perhaps all of this could be done in a subdirectory
631
799
632
t.put('a', StringIO('a first file\n'))
800
t.put_bytes('a', 'a first file\n')
633
801
self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
634
802
635
803
t.move('a', 'b')
640
808
self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
641
809
642
810
# Overwrite a file
643
t.put('c', StringIO('c this file\n'))
811
t.put_bytes('c', 'c this file\n')
644
812
t.move('c', 'b')
645
813
self.failIf(t.has('c'))
646
814
self.check_transport_contents('c this file\n', t, 'b')
655
823
if t.is_readonly():
656
824
return
657
825
658
t.put('a', StringIO('a file\n'))
826
t.put_bytes('a', 'a file\n')
659
827
t.copy('a', 'b')
660
828
self.check_transport_contents('a file\n', t, 'b')
661
829
664
832
# What should the assert be if you try to copy a
665
833
# file over a directory?
666
834
#self.assertRaises(Something, t.copy, 'a', 'c')
667
t.put('d', StringIO('text in d\n'))
835
t.put_bytes('d', 'text in d\n')
668
836
t.copy('d', 'b')
669
837
self.check_transport_contents('text in d\n', t, 'b')
670
838
816
984
if t1.is_readonly():
817
985
open('b/d', 'wb').write('newfile\n')
818
986
else:
819
t2.put('d', StringIO('newfile\n'))
987
t2.put_bytes('d', 'newfile\n')
820
988
821
989
self.failUnless(t1.has('b/d'))
822
990
self.failUnless(t2.has('d'))
916
1084
transport.iter_files_recursive)
917
1085
return
918
1086
if transport.is_readonly():
919
self.assertRaises(TransportNotPossible,
920
transport.put, 'a', 'some text for a\n')
921
1087
return
922
1088
self.build_tree(['from/',
923
1089
'from/dir/',
974
1140
transport = self.get_transport()
975
1141
if transport.is_readonly():
976
1142
return
977
transport.put('foo', StringIO('bar'))
1143
transport.put_bytes('foo', 'bar')
978
1144
transport2 = self.get_transport()
979
1145
self.check_transport_contents('bar', transport2, 'foo')
980
1146
# its base should be usable.
992
1158
if transport.is_readonly():
993
1159
self.assertRaises(TransportNotPossible, transport.lock_write, 'foo')
994
1160
return
995
transport.put('lock', StringIO())
1161
transport.put_bytes('lock', '')
996
1162
lock = transport.lock_write('lock')
997
1163
# TODO make this consistent on all platforms:
998
1164
# self.assertRaises(LockError, transport.lock_write, 'lock')
1003
1169
if transport.is_readonly():
1004
1170
file('lock', 'w').close()
1005
1171
else:
1006
transport.put('lock', StringIO())
1172
transport.put_bytes('lock', '')
1007
1173
lock = transport.lock_read('lock')
1008
1174
# TODO make this consistent on all platforms:
1009
1175
# self.assertRaises(LockError, transport.lock_read, 'lock')
1014
1180
if transport.is_readonly():
1015
1181
file('a', 'w').write('0123456789')
1016
1182
else:
1017
transport.put('a', StringIO('0123456789'))
1183
transport.put_bytes('a', '0123456789')
1018
1184
1019
1185
d = list(transport.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
1020
1186
self.assertEqual(d[0], (0, '0'))
1027
1193
if transport.is_readonly():
1028
1194
file('a', 'w').write('0123456789')
1029
1195
else:
1030
transport.put('a', StringIO('01234567890'))
1196
transport.put_bytes('a', '01234567890')
1031
1197
1032
1198
d = list(transport.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
1033
1199
self.assertEqual(d[0], (1, '1'))
Loggerhead is a web-based interface for Breezy
Version: 2.0.1