208
208
self.build_tree(files, transport=t, line_endings='binary')
209
209
self.assertRaises(NoSuchFile, t.get, 'c')
210
def iterate_and_close(func, *args):
211
for f in func(*args):
212
# We call f.read() here because things like paramiko actually
213
# spawn a thread to prefetch the content, which we want to
214
# consume before we close the handle.
217
self.assertRaises(NoSuchFile, iterate_and_close,
218
t.get_multi, ['a', 'b', 'c'])
219
self.assertRaises(NoSuchFile, iterate_and_close,
220
t.get_multi, iter(['a', 'b', 'c']))
210
self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
211
self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
222
213
def test_get_directory_read_gives_ReadError(self):
223
214
"""consistent errors for read() on a file returned by get()."""
296
287
t.put_bytes('a', 'some text for a\n')
297
self.assertTrue(t.has('a'))
288
self.failUnless(t.has('a'))
298
289
self.check_transport_contents('some text for a\n', t, 'a')
300
291
# The contents should be overwritten
312
303
t.put_bytes_non_atomic, 'a', 'some text for a\n')
315
self.assertFalse(t.has('a'))
306
self.failIf(t.has('a'))
316
307
t.put_bytes_non_atomic('a', 'some text for a\n')
317
self.assertTrue(t.has('a'))
308
self.failUnless(t.has('a'))
318
309
self.check_transport_contents('some text for a\n', t, 'a')
319
310
# Put also replaces contents
320
311
t.put_bytes_non_atomic('a', 'new\ncontents for\na\n')
332
323
# Now test the create_parent flag
333
324
self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'dir/a',
335
self.assertFalse(t.has('dir/a'))
326
self.failIf(t.has('dir/a'))
336
327
t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',
337
328
create_parent_dir=True)
338
329
self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
410
401
result = t.put_file('a', StringIO('some text for a\n'))
411
402
# put_file returns the length of the data written
412
403
self.assertEqual(16, result)
413
self.assertTrue(t.has('a'))
404
self.failUnless(t.has('a'))
414
405
self.check_transport_contents('some text for a\n', t, 'a')
415
406
# Put also replaces contents
416
407
result = t.put_file('a', StringIO('new\ncontents for\na\n'))
428
419
t.put_file_non_atomic, 'a', StringIO('some text for a\n'))
431
self.assertFalse(t.has('a'))
422
self.failIf(t.has('a'))
432
423
t.put_file_non_atomic('a', StringIO('some text for a\n'))
433
self.assertTrue(t.has('a'))
424
self.failUnless(t.has('a'))
434
425
self.check_transport_contents('some text for a\n', t, 'a')
435
426
# Put also replaces contents
436
427
t.put_file_non_atomic('a', StringIO('new\ncontents for\na\n'))
448
439
# Now test the create_parent flag
449
440
self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'dir/a',
450
441
StringIO('contents\n'))
451
self.assertFalse(t.has('dir/a'))
442
self.failIf(t.has('dir/a'))
452
443
t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
453
444
create_parent_dir=True)
454
445
self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
833
824
t.put_bytes('a', 'a little bit of text\n')
834
self.assertTrue(t.has('a'))
825
self.failUnless(t.has('a'))
836
self.assertFalse(t.has('a'))
827
self.failIf(t.has('a'))
838
829
self.assertRaises(NoSuchFile, t.delete, 'a')
845
836
t.delete_multi(['a', 'c'])
846
837
self.assertEqual([False, True, False],
847
838
list(t.has_multi(['a', 'b', 'c'])))
848
self.assertFalse(t.has('a'))
849
self.assertTrue(t.has('b'))
850
self.assertFalse(t.has('c'))
839
self.failIf(t.has('a'))
840
self.failUnless(t.has('b'))
841
self.failIf(t.has('c'))
852
843
self.assertRaises(NoSuchFile,
853
844
t.delete_multi, ['a', 'b', 'c'])
914
905
t.mkdir('foo-baz')
916
907
self.assertRaises((NoSuchFile, PathError), t.rmdir, 'foo')
917
self.assertTrue(t.has('foo-bar'))
908
self.failUnless(t.has('foo-bar'))
919
910
def test_rename_dir_succeeds(self):
920
911
t = self.get_transport()
1003
994
self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
1005
996
t.move('a', 'b')
1006
self.assertTrue(t.has('b'))
1007
self.assertFalse(t.has('a'))
997
self.failUnless(t.has('b'))
998
self.failIf(t.has('a'))
1009
1000
self.check_transport_contents('a first file\n', t, 'b')
1010
1001
self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
1012
1003
# Overwrite a file
1013
1004
t.put_bytes('c', 'c this file\n')
1014
1005
t.move('c', 'b')
1015
self.assertFalse(t.has('c'))
1006
self.failIf(t.has('c'))
1016
1007
self.check_transport_contents('c this file\n', t, 'b')
1018
1009
# TODO: Try to write a test for atomicity
1050
1041
except NotImplementedError:
1051
1042
raise TestSkipped("Transport %s has no bogus URL support." %
1052
1043
self._server.__class__)
1053
t = _mod_transport.get_transport_from_url(url)
1044
t = _mod_transport.get_transport(url)
1054
1045
self.assertRaises((ConnectionError, NoSuchFile), t.get, '.bzr/branch')
1056
1047
def test_stat(self):
1072
1063
for path, size in zip(paths, sizes):
1073
1064
st = t.stat(path)
1074
1065
if path.endswith('/'):
1075
self.assertTrue(S_ISDIR(st.st_mode))
1066
self.failUnless(S_ISDIR(st.st_mode))
1076
1067
# directory sizes are meaningless
1078
self.assertTrue(S_ISREG(st.st_mode))
1069
self.failUnless(S_ISREG(st.st_mode))
1079
1070
self.assertEqual(size, st.st_size)
1081
1072
remote_stats = list(t.stat_multi(paths))
1088
1079
self.assertListRaises(NoSuchFile, t.stat_multi, iter(['a', 'c', 'd']))
1089
1080
self.build_tree(['subdir/', 'subdir/file'], transport=t)
1090
1081
subdir = t.clone('subdir')
1091
st = subdir.stat('./file')
1092
st = subdir.stat('.')
1082
subdir.stat('./file')
1094
1085
def test_hardlink(self):
1095
1086
from stat import ST_NLINK
1105
1096
t.hardlink(source_name, link_name)
1107
self.assertTrue(t.has(source_name))
1108
self.assertTrue(t.has(link_name))
1098
self.failUnless(t.has(source_name))
1099
self.failUnless(t.has(link_name))
1110
1101
st = t.stat(link_name)
1111
self.assertEqual(st[ST_NLINK], 2)
1102
self.failUnlessEqual(st[ST_NLINK], 2)
1112
1103
except TransportNotPossible:
1113
1104
raise TestSkipped("Transport %s does not support hardlinks." %
1114
1105
self._server.__class__)
1127
1118
t.symlink(source_name, link_name)
1129
self.assertTrue(t.has(source_name))
1130
self.assertTrue(t.has(link_name))
1120
self.failUnless(t.has(source_name))
1121
self.failUnless(t.has(link_name))
1132
1123
st = t.stat(link_name)
1133
self.assertTrue(S_ISLNK(st.st_mode),
1124
self.failUnless(S_ISLNK(st.st_mode),
1134
1125
"expected symlink, got mode %o" % st.st_mode)
1135
1126
except TransportNotPossible:
1136
1127
raise TestSkipped("Transport %s does not support symlinks." %
1137
1128
self._server.__class__)
1138
1129
except IOError:
1139
self.knownFailure("Paramiko fails to create symlinks during tests")
1130
raise tests.KnownFailure("Paramiko fails to create symlinks during tests")
1141
1132
def test_list_dir(self):
1142
1133
# TODO: Test list_dir, just try once, and if it throws, stop testing
1206
1197
raise TestSkipped("not a connected transport")
1208
1199
t2 = t1.clone('subdir')
1209
self.assertEquals(t1._parsed_url.scheme, t2._parsed_url.scheme)
1210
self.assertEquals(t1._parsed_url.user, t2._parsed_url.user)
1211
self.assertEquals(t1._parsed_url.password, t2._parsed_url.password)
1212
self.assertEquals(t1._parsed_url.host, t2._parsed_url.host)
1213
self.assertEquals(t1._parsed_url.port, t2._parsed_url.port)
1200
self.assertEquals(t1._scheme, t2._scheme)
1201
self.assertEquals(t1._user, t2._user)
1202
self.assertEquals(t1._password, t2._password)
1203
self.assertEquals(t1._host, t2._host)
1204
self.assertEquals(t1._port, t2._port)
1215
1206
def test__reuse_for(self):
1216
1207
t = self.get_transport()
1224
1215
Only the parameters different from None will be changed.
1226
if scheme is None: scheme = t._parsed_url.scheme
1227
if user is None: user = t._parsed_url.user
1228
if password is None: password = t._parsed_url.password
1229
if user is None: user = t._parsed_url.user
1230
if host is None: host = t._parsed_url.host
1231
if port is None: port = t._parsed_url.port
1232
if path is None: path = t._parsed_url.path
1233
return str(urlutils.URL(scheme, user, password, host, port, path))
1217
if scheme is None: scheme = t._scheme
1218
if user is None: user = t._user
1219
if password is None: password = t._password
1220
if user is None: user = t._user
1221
if host is None: host = t._host
1222
if port is None: port = t._port
1223
if path is None: path = t._path
1224
return t._unsplit_url(scheme, user, password, host, port, path)
1235
if t._parsed_url.scheme == 'ftp':
1226
if t._scheme == 'ftp':
1236
1227
scheme = 'sftp'
1239
1230
self.assertIsNot(t, t._reuse_for(new_url(scheme=scheme)))
1240
if t._parsed_url.user == 'me':
1254
1245
# (they may be typed by the user when prompted for example)
1255
1246
self.assertIs(t, t._reuse_for(new_url(password='from space')))
1256
1247
# We will not connect, we can use a invalid host
1257
self.assertIsNot(t, t._reuse_for(new_url(host=t._parsed_url.host + 'bar')))
1258
if t._parsed_url.port == 1234:
1248
self.assertIsNot(t, t._reuse_for(new_url(host=t._host + 'bar')))
1303
1294
self.build_tree(['a', 'b/', 'b/c'], transport=t1)
1305
self.assertTrue(t1.has('a'))
1306
self.assertTrue(t1.has('b/c'))
1307
self.assertFalse(t1.has('c'))
1296
self.failUnless(t1.has('a'))
1297
self.failUnless(t1.has('b/c'))
1298
self.failIf(t1.has('c'))
1309
1300
t2 = t1.clone('b')
1310
1301
self.assertEqual(t1.base + 'b/', t2.base)
1312
self.assertTrue(t2.has('c'))
1313
self.assertFalse(t2.has('a'))
1303
self.failUnless(t2.has('c'))
1304
self.failIf(t2.has('a'))
1315
1306
t3 = t2.clone('..')
1316
self.assertTrue(t3.has('a'))
1317
self.assertFalse(t3.has('c'))
1307
self.failUnless(t3.has('a'))
1308
self.failIf(t3.has('c'))
1319
self.assertFalse(t1.has('b/d'))
1320
self.assertFalse(t2.has('d'))
1321
self.assertFalse(t3.has('b/d'))
1310
self.failIf(t1.has('b/d'))
1311
self.failIf(t2.has('d'))
1312
self.failIf(t3.has('b/d'))
1323
1314
if t1.is_readonly():
1324
1315
self.build_tree_contents([('b/d', 'newfile\n')])
1326
1317
t2.put_bytes('d', 'newfile\n')
1328
self.assertTrue(t1.has('b/d'))
1329
self.assertTrue(t2.has('d'))
1330
self.assertTrue(t3.has('b/d'))
1319
self.failUnless(t1.has('b/d'))
1320
self.failUnless(t2.has('d'))
1321
self.failUnless(t3.has('b/d'))
1332
1323
def test_clone_to_root(self):
1333
1324
orig_transport = self.get_transport()
1419
1410
# smoke test for abspath on win32.
1420
1411
# a transport based on 'file:///' never fully qualifies the drive.
1421
transport = _mod_transport.get_transport_from_url("file:///")
1422
self.assertEqual(transport.abspath("/"), "file:///")
1412
transport = _mod_transport.get_transport("file:///")
1413
self.failUnlessEqual(transport.abspath("/"), "file:///")
1424
1415
# but a transport that starts with a drive spec must keep it.
1425
transport = _mod_transport.get_transport_from_url("file:///C:/")
1426
self.assertEqual(transport.abspath("/"), "file:///C:/")
1416
transport = _mod_transport.get_transport("file:///C:/")
1417
self.failUnlessEqual(transport.abspath("/"), "file:///C:/")
1428
1419
def test_local_abspath(self):
1429
1420
transport = self.get_transport()
1556
1547
no_unicode_support = getattr(self._server, 'no_unicode_support', False)
1557
1548
if no_unicode_support:
1558
self.knownFailure("test server cannot handle unicode paths")
1549
raise tests.KnownFailure("test server cannot handle unicode paths")
1561
1552
self.build_tree(files, transport=t, line_endings='binary')
1626
1617
def test_readv(self):
1627
1618
transport = self.get_transport()
1628
1619
if transport.is_readonly():
1629
with file('a', 'w') as f: f.write('0123456789')
1620
file('a', 'w').write('0123456789')
1631
1622
transport.put_bytes('a', '0123456789')
1642
1633
def test_readv_out_of_order(self):
1643
1634
transport = self.get_transport()
1644
1635
if transport.is_readonly():
1645
with file('a', 'w') as f: f.write('0123456789')
1636
file('a', 'w').write('0123456789')
1647
1638
transport.put_bytes('a', '01234567890')
1720
1711
transport = self.get_transport()
1721
1712
# test from observed failure case.
1722
1713
if transport.is_readonly():
1723
with file('a', 'w') as f: f.write('a'*1024*1024)
1714
file('a', 'w').write('a'*1024*1024)
1725
1716
transport.put_bytes('a', 'a'*1024*1024)
1726
1717
broken_vector = [(465219, 800), (225221, 800), (445548, 800),
1760
1751
def test_readv_short_read(self):
1761
1752
transport = self.get_transport()
1762
1753
if transport.is_readonly():
1763
with file('a', 'w') as f: f.write('0123456789')
1754
file('a', 'w').write('0123456789')
1765
1756
transport.put_bytes('a', '01234567890')
1776
1767
self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),
1777
1768
transport.readv, 'a', [(12,2)])
1779
def test_no_segment_parameters(self):
1780
"""Segment parameters should be stripped and stored in
1781
transport.segment_parameters."""
1782
transport = self.get_transport("foo")
1783
self.assertEquals({}, transport.get_segment_parameters())
1785
def test_segment_parameters(self):
1786
"""Segment parameters should be stripped and stored in
1787
transport.get_segment_parameters()."""
1788
base_url = self._server.get_url()
1789
parameters = {"key1": "val1", "key2": "val2"}
1790
url = urlutils.join_segment_parameters(base_url, parameters)
1791
transport = _mod_transport.get_transport_from_url(url)
1792
self.assertEquals(parameters, transport.get_segment_parameters())
1794
1770
def test_stat_symlink(self):
1795
1771
# if a transport points directly to a symlink (and supports symlinks
1796
1772
# at all) you can tell this. helps with bug 32669.
added
removed
bzrlib/tests/per_transport.py
