« back to all changes in this revision
Viewing changes to bzrlib/tests/test_transport_implementations.py
- browse files at revision 1957.1.6
- compare with another revision
- compare with revision 5200.3.3
- download diff
- download tarball
- view history from revision 1957.1.6
- Committer: John Arbash Meinel
- Date: 2006-09-15 00:44:57 UTC
- mfrom: (2009 +trunk)
- mto: This revision was merged to the branch mainline in revision 2050.
- Revision ID: john@arbash-meinel.com-20060915004457-902cec0526a39337
[merge] bzr.dev 2009
- files added:
- bzrlib/inspect_for_copy.py
- files modified:
- HACKING
- setup.py *
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', '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
543
713
t.mkdir('adir')
544
714
t.mkdir('adir/bdir')
545
715
t.rmdir('adir/bdir')
546
self.assertRaises(NoSuchFile, t.stat, 'adir/bdir')
716
# ftp may not be able to raise NoSuchFile for lack of
717
# details when failing
718
self.assertRaises((NoSuchFile, PathError), t.rmdir, 'adir/bdir')
547
719
t.rmdir('adir')
548
self.assertRaises(NoSuchFile, t.stat, 'adir')
720
self.assertRaises((NoSuchFile, PathError), t.rmdir, 'adir')
549
721
550
722
def test_rmdir_not_empty(self):
551
723
"""Deleting a non-empty directory raises an exception
579
751
t.mkdir('adir/asubdir')
580
752
t.mkdir('bdir')
581
753
t.mkdir('bdir/bsubdir')
754
# any kind of PathError would be OK, though we normally expect
755
# DirectoryNotEmpty
582
756
self.assertRaises(PathError, t.rename, 'bdir', 'adir')
583
757
# nothing was changed so it should still be as before
584
758
self.assertTrue(t.has('bdir/bsubdir'))
627
801
# creates control files in the working directory
628
802
# perhaps all of this could be done in a subdirectory
629
803
630
t.put('a', StringIO('a first file\n'))
804
t.put_bytes('a', 'a first file\n')
631
805
self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
632
806
633
807
t.move('a', 'b')
638
812
self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
639
813
640
814
# Overwrite a file
641
t.put('c', StringIO('c this file\n'))
815
t.put_bytes('c', 'c this file\n')
642
816
t.move('c', 'b')
643
817
self.failIf(t.has('c'))
644
818
self.check_transport_contents('c this file\n', t, 'b')
653
827
if t.is_readonly():
654
828
return
655
829
656
t.put('a', StringIO('a file\n'))
830
t.put_bytes('a', 'a file\n')
657
831
t.copy('a', 'b')
658
832
self.check_transport_contents('a file\n', t, 'b')
659
833
662
836
# What should the assert be if you try to copy a
663
837
# file over a directory?
664
838
#self.assertRaises(Something, t.copy, 'a', 'c')
665
t.put('d', StringIO('text in d\n'))
839
t.put_bytes('d', 'text in d\n')
666
840
t.copy('d', 'b')
667
841
self.check_transport_contents('text in d\n', t, 'b')
668
842
669
843
# TODO: test copy_multi
670
844
671
845
def test_connection_error(self):
672
"""ConnectionError is raised when connection is impossible"""
846
"""ConnectionError is raised when connection is impossible.
847
848
The error may be raised from either the constructor or the first
849
operation on the transport.
850
"""
673
851
try:
674
852
url = self._server.get_bogus_url()
675
853
except NotImplementedError:
745
923
os.mkdir('wd')
746
924
t = t.clone('wd')
747
925
748
self.assertEqual([], sorted_list(u'.'))
926
self.assertEqual([], sorted_list('.'))
749
927
# c2 is precisely one letter longer than c here to test that
750
928
# suffixing is not confused.
929
# a%25b checks that quoting is done consistently across transports
930
tree_names = ['a', 'a%25b', 'b', 'c/', 'c/d', 'c/e', 'c2/']
751
931
if not t.is_readonly():
752
self.build_tree(['a', 'b', 'c/', 'c/d', 'c/e', 'c2/'], transport=t)
932
self.build_tree(tree_names, transport=t)
753
933
else:
754
self.build_tree(['wd/a', 'wd/b', 'wd/c/', 'wd/c/d', 'wd/c/e', 'wd/c2/'])
934
self.build_tree(['wd/' + name for name in tree_names])
755
935
756
self.assertEqual([u'a', u'b', u'c', u'c2'], sorted_list(u'.'))
757
self.assertEqual([u'd', u'e'], sorted_list(u'c'))
936
self.assertEqual(
937
['a', 'a%2525b', 'b', 'c', 'c2'], sorted_list('.'))
938
self.assertEqual(['d', 'e'], sorted_list('c'))
758
939
759
940
if not t.is_readonly():
760
941
t.delete('c/d')
763
944
os.unlink('wd/c/d')
764
945
os.unlink('wd/b')
765
946
766
self.assertEqual([u'a', u'c', u'c2'], sorted_list('.'))
767
self.assertEqual([u'e'], sorted_list(u'c'))
947
self.assertEqual(['a', 'a%2525b', 'c', 'c2'], sorted_list('.'))
948
self.assertEqual(['e'], sorted_list('c'))
768
949
769
950
self.assertListRaises(PathError, t.list_dir, 'q')
770
951
self.assertListRaises(PathError, t.list_dir, 'c/f')
771
952
self.assertListRaises(PathError, t.list_dir, 'a')
772
953
954
def test_list_dir_result_is_url_escaped(self):
955
t = self.get_transport()
956
if not t.listable():
957
raise TestSkipped("transport not listable")
958
959
if not t.is_readonly():
960
self.build_tree(['a/', 'a/%'], transport=t)
961
else:
962
self.build_tree(['a/', 'a/%'])
963
964
names = list(t.list_dir('a'))
965
self.assertEqual(['%25'], names)
966
self.assertIsInstance(names[0], str)
967
773
968
def test_clone(self):
774
969
# TODO: Test that clone moves up and down the filesystem
775
970
t1 = self.get_transport()
797
992
if t1.is_readonly():
798
993
open('b/d', 'wb').write('newfile\n')
799
994
else:
800
t2.put('d', StringIO('newfile\n'))
995
t2.put_bytes('d', 'newfile\n')
801
996
802
997
self.failUnless(t1.has('b/d'))
803
998
self.failUnless(t2.has('d'))
872
1067
'isolated/dir/',
873
1068
'isolated/dir/foo',
874
1069
'isolated/dir/bar',
1070
'isolated/dir/b%25z', # make sure quoting is correct
875
1071
'isolated/bar'],
876
1072
transport=transport)
877
1073
paths = set(transport.iter_files_recursive())
879
1075
self.assertEqual(paths,
880
1076
set(['isolated/dir/foo',
881
1077
'isolated/dir/bar',
1078
'isolated/dir/b%2525z',
882
1079
'isolated/bar']))
883
1080
sub_transport = transport.clone('isolated')
884
1081
paths = set(sub_transport.iter_files_recursive())
885
self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
1082
self.assertEqual(paths,
1083
set(['dir/foo', 'dir/bar', 'dir/b%2525z', 'bar']))
1084
1085
def test_copy_tree(self):
1086
# TODO: test file contents and permissions are preserved. This test was
1087
# added just to ensure that quoting was handled correctly.
1088
# -- David Allouche 2006-08-11
1089
transport = self.get_transport()
1090
if not transport.listable():
1091
self.assertRaises(TransportNotPossible,
1092
transport.iter_files_recursive)
1093
return
1094
if transport.is_readonly():
1095
return
1096
self.build_tree(['from/',
1097
'from/dir/',
1098
'from/dir/foo',
1099
'from/dir/bar',
1100
'from/dir/b%25z', # make sure quoting is correct
1101
'from/bar'],
1102
transport=transport)
1103
transport.copy_tree('from', 'to')
1104
paths = set(transport.iter_files_recursive())
1105
self.assertEqual(paths,
1106
set(['from/dir/foo',
1107
'from/dir/bar',
1108
'from/dir/b%2525z',
1109
'from/bar',
1110
'to/dir/foo',
1111
'to/dir/bar',
1112
'to/dir/b%2525z',
1113
'to/bar',]))
886
1114
887
1115
def test_unicode_paths(self):
888
1116
"""Test that we can read/write files with Unicode names."""
920
1148
transport = self.get_transport()
921
1149
if transport.is_readonly():
922
1150
return
923
transport.put('foo', StringIO('bar'))
1151
transport.put_bytes('foo', 'bar')
924
1152
transport2 = self.get_transport()
925
1153
self.check_transport_contents('bar', transport2, 'foo')
926
1154
# its base should be usable.
938
1166
if transport.is_readonly():
939
1167
self.assertRaises(TransportNotPossible, transport.lock_write, 'foo')
940
1168
return
941
transport.put('lock', StringIO())
1169
transport.put_bytes('lock', '')
942
1170
lock = transport.lock_write('lock')
943
1171
# TODO make this consistent on all platforms:
944
1172
# self.assertRaises(LockError, transport.lock_write, 'lock')
949
1177
if transport.is_readonly():
950
1178
file('lock', 'w').close()
951
1179
else:
952
transport.put('lock', StringIO())
1180
transport.put_bytes('lock', '')
953
1181
lock = transport.lock_read('lock')
954
1182
# TODO make this consistent on all platforms:
955
1183
# self.assertRaises(LockError, transport.lock_read, 'lock')
960
1188
if transport.is_readonly():
961
1189
file('a', 'w').write('0123456789')
962
1190
else:
963
transport.put('a', StringIO('0123456789'))
1191
transport.put_bytes('a', '0123456789')
964
1192
965
1193
d = list(transport.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
966
1194
self.assertEqual(d[0], (0, '0'))
973
1201
if transport.is_readonly():
974
1202
file('a', 'w').write('0123456789')
975
1203
else:
976
transport.put('a', StringIO('01234567890'))
1204
transport.put_bytes('a', '01234567890')
977
1205
978
1206
d = list(transport.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
979
1207
self.assertEqual(d[0], (1, '1'))
Loggerhead is a web-based interface for Breezy
Version: 2.0.1