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.2.47
- compare with another revision
- compare with revision 5200.3.1
- download diff
- download tarball
- view history from revision 1910.2.47
- Committer: Aaron Bentley
- Date: 2006-09-09 18:52:57 UTC
- mfrom: (1996 +trunk)
- mto: This revision was merged to the branch mainline in revision 1997.
- Revision ID: aaron.bentley@utoronto.ca-20060909185257-ce0ee03ee5125ff1
Merge bzr.dev
- files added:
- bzrlib/tests/blackbox/test_locale.py
- files modified:
- Makefile
- 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
627
799
# creates control files in the working directory
628
800
# perhaps all of this could be done in a subdirectory
629
801
630
t.put('a', StringIO('a first file\n'))
802
t.put_bytes('a', 'a first file\n')
631
803
self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
632
804
633
805
t.move('a', 'b')
638
810
self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
639
811
640
812
# Overwrite a file
641
t.put('c', StringIO('c this file\n'))
813
t.put_bytes('c', 'c this file\n')
642
814
t.move('c', 'b')
643
815
self.failIf(t.has('c'))
644
816
self.check_transport_contents('c this file\n', t, 'b')
653
825
if t.is_readonly():
654
826
return
655
827
656
t.put('a', StringIO('a file\n'))
828
t.put_bytes('a', 'a file\n')
657
829
t.copy('a', 'b')
658
830
self.check_transport_contents('a file\n', t, 'b')
659
831
662
834
# What should the assert be if you try to copy a
663
835
# file over a directory?
664
836
#self.assertRaises(Something, t.copy, 'a', 'c')
665
t.put('d', StringIO('text in d\n'))
837
t.put_bytes('d', 'text in d\n')
666
838
t.copy('d', 'b')
667
839
self.check_transport_contents('text in d\n', t, 'b')
668
840
745
917
os.mkdir('wd')
746
918
t = t.clone('wd')
747
919
748
self.assertEqual([], sorted_list(u'.'))
920
self.assertEqual([], sorted_list('.'))
749
921
# c2 is precisely one letter longer than c here to test that
750
922
# suffixing is not confused.
923
# a%25b checks that quoting is done consistently across transports
924
tree_names = ['a', 'a%25b', 'b', 'c/', 'c/d', 'c/e', 'c2/']
751
925
if not t.is_readonly():
752
self.build_tree(['a', 'b', 'c/', 'c/d', 'c/e', 'c2/'], transport=t)
926
self.build_tree(tree_names, transport=t)
753
927
else:
754
self.build_tree(['wd/a', 'wd/b', 'wd/c/', 'wd/c/d', 'wd/c/e', 'wd/c2/'])
928
self.build_tree(['wd/' + name for name in tree_names])
755
929
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'))
930
self.assertEqual(
931
['a', 'a%2525b', 'b', 'c', 'c2'], sorted_list('.'))
932
self.assertEqual(['d', 'e'], sorted_list('c'))
758
933
759
934
if not t.is_readonly():
760
935
t.delete('c/d')
763
938
os.unlink('wd/c/d')
764
939
os.unlink('wd/b')
765
940
766
self.assertEqual([u'a', u'c', u'c2'], sorted_list('.'))
767
self.assertEqual([u'e'], sorted_list(u'c'))
941
self.assertEqual(['a', 'a%2525b', 'c', 'c2'], sorted_list('.'))
942
self.assertEqual(['e'], sorted_list('c'))
768
943
769
944
self.assertListRaises(PathError, t.list_dir, 'q')
770
945
self.assertListRaises(PathError, t.list_dir, 'c/f')
771
946
self.assertListRaises(PathError, t.list_dir, 'a')
772
947
948
def test_list_dir_result_is_url_escaped(self):
949
t = self.get_transport()
950
if not t.listable():
951
raise TestSkipped("transport not listable")
952
953
if not t.is_readonly():
954
self.build_tree(['a/', 'a/%'], transport=t)
955
else:
956
self.build_tree(['a/', 'a/%'])
957
958
names = list(t.list_dir('a'))
959
self.assertEqual(['%25'], names)
960
self.assertIsInstance(names[0], str)
961
773
962
def test_clone(self):
774
963
# TODO: Test that clone moves up and down the filesystem
775
964
t1 = self.get_transport()
797
986
if t1.is_readonly():
798
987
open('b/d', 'wb').write('newfile\n')
799
988
else:
800
t2.put('d', StringIO('newfile\n'))
989
t2.put_bytes('d', 'newfile\n')
801
990
802
991
self.failUnless(t1.has('b/d'))
803
992
self.failUnless(t2.has('d'))
872
1061
'isolated/dir/',
873
1062
'isolated/dir/foo',
874
1063
'isolated/dir/bar',
1064
'isolated/dir/b%25z', # make sure quoting is correct
875
1065
'isolated/bar'],
876
1066
transport=transport)
877
1067
paths = set(transport.iter_files_recursive())
879
1069
self.assertEqual(paths,
880
1070
set(['isolated/dir/foo',
881
1071
'isolated/dir/bar',
1072
'isolated/dir/b%2525z',
882
1073
'isolated/bar']))
883
1074
sub_transport = transport.clone('isolated')
884
1075
paths = set(sub_transport.iter_files_recursive())
885
self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
1076
self.assertEqual(paths,
1077
set(['dir/foo', 'dir/bar', 'dir/b%2525z', 'bar']))
1078
1079
def test_copy_tree(self):
1080
# TODO: test file contents and permissions are preserved. This test was
1081
# added just to ensure that quoting was handled correctly.
1082
# -- David Allouche 2006-08-11
1083
transport = self.get_transport()
1084
if not transport.listable():
1085
self.assertRaises(TransportNotPossible,
1086
transport.iter_files_recursive)
1087
return
1088
if transport.is_readonly():
1089
return
1090
self.build_tree(['from/',
1091
'from/dir/',
1092
'from/dir/foo',
1093
'from/dir/bar',
1094
'from/dir/b%25z', # make sure quoting is correct
1095
'from/bar'],
1096
transport=transport)
1097
transport.copy_tree('from', 'to')
1098
paths = set(transport.iter_files_recursive())
1099
self.assertEqual(paths,
1100
set(['from/dir/foo',
1101
'from/dir/bar',
1102
'from/dir/b%2525z',
1103
'from/bar',
1104
'to/dir/foo',
1105
'to/dir/bar',
1106
'to/dir/b%2525z',
1107
'to/bar',]))
886
1108
887
1109
def test_unicode_paths(self):
888
1110
"""Test that we can read/write files with Unicode names."""
920
1142
transport = self.get_transport()
921
1143
if transport.is_readonly():
922
1144
return
923
transport.put('foo', StringIO('bar'))
1145
transport.put_bytes('foo', 'bar')
924
1146
transport2 = self.get_transport()
925
1147
self.check_transport_contents('bar', transport2, 'foo')
926
1148
# its base should be usable.
938
1160
if transport.is_readonly():
939
1161
self.assertRaises(TransportNotPossible, transport.lock_write, 'foo')
940
1162
return
941
transport.put('lock', StringIO())
1163
transport.put_bytes('lock', '')
942
1164
lock = transport.lock_write('lock')
943
1165
# TODO make this consistent on all platforms:
944
1166
# self.assertRaises(LockError, transport.lock_write, 'lock')
949
1171
if transport.is_readonly():
950
1172
file('lock', 'w').close()
951
1173
else:
952
transport.put('lock', StringIO())
1174
transport.put_bytes('lock', '')
953
1175
lock = transport.lock_read('lock')
954
1176
# TODO make this consistent on all platforms:
955
1177
# self.assertRaises(LockError, transport.lock_read, 'lock')
960
1182
if transport.is_readonly():
961
1183
file('a', 'w').write('0123456789')
962
1184
else:
963
transport.put('a', StringIO('0123456789'))
1185
transport.put_bytes('a', '0123456789')
964
1186
965
1187
d = list(transport.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
966
1188
self.assertEqual(d[0], (0, '0'))
973
1195
if transport.is_readonly():
974
1196
file('a', 'w').write('0123456789')
975
1197
else:
976
transport.put('a', StringIO('01234567890'))
1198
transport.put_bytes('a', '01234567890')
977
1199
978
1200
d = list(transport.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
979
1201
self.assertEqual(d[0], (1, '1'))
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0