~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/per_transport.py

  • Committer: Robert Collins
  • Date: 2010-05-06 11:08:10 UTC
  • mto: This revision was merged to the branch mainline in revision 5223.
  • Revision ID: robertc@robertcollins.net-20100506110810-h3j07fh5gmw54s25
Cleaner matcher matching revised unlocking protocol.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005-2011 Canonical Ltd
 
1
# Copyright (C) 2005-2010 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
26
26
from StringIO import StringIO as pyStringIO
27
27
import stat
28
28
import sys
 
29
import unittest
29
30
 
30
31
from bzrlib import (
31
32
    errors,
32
33
    osutils,
33
 
    pyutils,
34
34
    tests,
35
 
    transport as _mod_transport,
36
35
    urlutils,
37
36
    )
38
37
from bzrlib.errors import (ConnectionError,
 
38
                           DirectoryNotEmpty,
39
39
                           FileExists,
40
40
                           InvalidURL,
 
41
                           LockError,
41
42
                           NoSuchFile,
 
43
                           NotLocalUrl,
42
44
                           PathError,
43
45
                           TransportNotPossible,
44
46
                           )
45
47
from bzrlib.osutils import getcwd
46
48
from bzrlib.smart import medium
47
49
from bzrlib.tests import (
 
50
    TestCaseInTempDir,
48
51
    TestSkipped,
49
52
    TestNotApplicable,
50
53
    multiply_tests,
53
56
from bzrlib.tests.test_transport import TestTransportImplementation
54
57
from bzrlib.transport import (
55
58
    ConnectedTransport,
 
59
    get_transport,
56
60
    _get_transport_modules,
57
61
    )
58
62
from bzrlib.transport.memory import MemoryTransport
74
78
    for module in _get_transport_modules():
75
79
        try:
76
80
            permutations = get_transport_test_permutations(
77
 
                pyutils.get_named_object(module))
 
81
                reduce(getattr, (module).split('.')[1:], __import__(module)))
78
82
            for (klass, server_factory) in permutations:
79
83
                scenario = ('%s,%s' % (klass.__name__, server_factory.__name__),
80
84
                    {"transport_class":klass,
98
102
 
99
103
    def setUp(self):
100
104
        super(TransportTests, self).setUp()
101
 
        self.overrideEnv('BZR_NO_SMART_VFS', None)
 
105
        self._captureVar('BZR_NO_SMART_VFS', None)
102
106
 
103
107
    def check_transport_contents(self, content, transport, relpath):
104
 
        """Check that transport.get_bytes(relpath) == content."""
105
 
        self.assertEqualDiff(content, transport.get_bytes(relpath))
 
108
        """Check that transport.get(relpath).read() == content."""
 
109
        self.assertEqualDiff(content, transport.get(relpath).read())
106
110
 
107
111
    def test_ensure_base_missing(self):
108
112
        """.ensure_base() should create the directory if it doesn't exist"""
207
211
                    ]
208
212
        self.build_tree(files, transport=t, line_endings='binary')
209
213
        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.
215
 
                content = f.read()
216
 
                f.close()
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']))
 
214
        self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
 
215
        self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))
221
216
 
222
217
    def test_get_directory_read_gives_ReadError(self):
223
218
        """consistent errors for read() on a file returned by get()."""
256
251
 
257
252
    def test_get_bytes_unknown_file(self):
258
253
        t = self.get_transport()
 
254
 
259
255
        self.assertRaises(NoSuchFile, t.get_bytes, 'c')
260
256
 
261
257
    def test_get_with_open_write_stream_sees_all_content(self):
265
261
        handle = t.open_write_stream('foo')
266
262
        try:
267
263
            handle.write('b')
268
 
            self.assertEqual('b', t.get_bytes('foo'))
 
264
            self.assertEqual('b', t.get('foo').read())
269
265
        finally:
270
266
            handle.close()
271
267
 
277
273
        try:
278
274
            handle.write('b')
279
275
            self.assertEqual('b', t.get_bytes('foo'))
280
 
            f = t.get('foo')
281
 
            try:
282
 
                self.assertEqual('b', f.read())
283
 
            finally:
284
 
                f.close()
 
276
            self.assertEqual('b', t.get('foo').read())
285
277
        finally:
286
278
            handle.close()
287
279
 
294
286
            return
295
287
 
296
288
        t.put_bytes('a', 'some text for a\n')
297
 
        self.assertTrue(t.has('a'))
 
289
        self.failUnless(t.has('a'))
298
290
        self.check_transport_contents('some text for a\n', t, 'a')
299
291
 
300
292
        # The contents should be overwritten
312
304
                    t.put_bytes_non_atomic, 'a', 'some text for a\n')
313
305
            return
314
306
 
315
 
        self.assertFalse(t.has('a'))
 
307
        self.failIf(t.has('a'))
316
308
        t.put_bytes_non_atomic('a', 'some text for a\n')
317
 
        self.assertTrue(t.has('a'))
 
309
        self.failUnless(t.has('a'))
318
310
        self.check_transport_contents('some text for a\n', t, 'a')
319
311
        # Put also replaces contents
320
312
        t.put_bytes_non_atomic('a', 'new\ncontents for\na\n')
332
324
        # Now test the create_parent flag
333
325
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'dir/a',
334
326
                                       'contents\n')
335
 
        self.assertFalse(t.has('dir/a'))
 
327
        self.failIf(t.has('dir/a'))
336
328
        t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',
337
329
                               create_parent_dir=True)
338
330
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
410
402
        result = t.put_file('a', StringIO('some text for a\n'))
411
403
        # put_file returns the length of the data written
412
404
        self.assertEqual(16, result)
413
 
        self.assertTrue(t.has('a'))
 
405
        self.failUnless(t.has('a'))
414
406
        self.check_transport_contents('some text for a\n', t, 'a')
415
407
        # Put also replaces contents
416
408
        result = t.put_file('a', StringIO('new\ncontents for\na\n'))
428
420
                    t.put_file_non_atomic, 'a', StringIO('some text for a\n'))
429
421
            return
430
422
 
431
 
        self.assertFalse(t.has('a'))
 
423
        self.failIf(t.has('a'))
432
424
        t.put_file_non_atomic('a', StringIO('some text for a\n'))
433
 
        self.assertTrue(t.has('a'))
 
425
        self.failUnless(t.has('a'))
434
426
        self.check_transport_contents('some text for a\n', t, 'a')
435
427
        # Put also replaces contents
436
428
        t.put_file_non_atomic('a', StringIO('new\ncontents for\na\n'))
448
440
        # Now test the create_parent flag
449
441
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'dir/a',
450
442
                                       StringIO('contents\n'))
451
 
        self.assertFalse(t.has('dir/a'))
 
443
        self.failIf(t.has('dir/a'))
452
444
        t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
453
445
                              create_parent_dir=True)
454
446
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
653
645
            self.build_tree(files, transport=transport_from)
654
646
            self.assertEqual(4, transport_from.copy_to(files, transport_to))
655
647
            for f in files:
656
 
                self.check_transport_contents(transport_to.get_bytes(f),
 
648
                self.check_transport_contents(transport_to.get(f).read(),
657
649
                                              transport_from, f)
658
650
 
659
651
        t = self.get_transport()
682
674
        files = ['a', 'b', 'c', 'd']
683
675
        t.copy_to(iter(files), temp_transport)
684
676
        for f in files:
685
 
            self.check_transport_contents(temp_transport.get_bytes(f),
 
677
            self.check_transport_contents(temp_transport.get(f).read(),
686
678
                                          t, f)
687
679
        del temp_transport
688
680
 
831
823
            return
832
824
 
833
825
        t.put_bytes('a', 'a little bit of text\n')
834
 
        self.assertTrue(t.has('a'))
 
826
        self.failUnless(t.has('a'))
835
827
        t.delete('a')
836
 
        self.assertFalse(t.has('a'))
 
828
        self.failIf(t.has('a'))
837
829
 
838
830
        self.assertRaises(NoSuchFile, t.delete, 'a')
839
831
 
845
837
        t.delete_multi(['a', 'c'])
846
838
        self.assertEqual([False, True, False],
847
839
                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'))
 
840
        self.failIf(t.has('a'))
 
841
        self.failUnless(t.has('b'))
 
842
        self.failIf(t.has('c'))
851
843
 
852
844
        self.assertRaises(NoSuchFile,
853
845
                t.delete_multi, ['a', 'b', 'c'])
914
906
        t.mkdir('foo-baz')
915
907
        t.rmdir('foo')
916
908
        self.assertRaises((NoSuchFile, PathError), t.rmdir, 'foo')
917
 
        self.assertTrue(t.has('foo-bar'))
 
909
        self.failUnless(t.has('foo-bar'))
918
910
 
919
911
    def test_rename_dir_succeeds(self):
920
912
        t = self.get_transport()
1003
995
        self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
1004
996
 
1005
997
        t.move('a', 'b')
1006
 
        self.assertTrue(t.has('b'))
1007
 
        self.assertFalse(t.has('a'))
 
998
        self.failUnless(t.has('b'))
 
999
        self.failIf(t.has('a'))
1008
1000
 
1009
1001
        self.check_transport_contents('a first file\n', t, 'b')
1010
1002
        self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
1012
1004
        # Overwrite a file
1013
1005
        t.put_bytes('c', 'c this file\n')
1014
1006
        t.move('c', 'b')
1015
 
        self.assertFalse(t.has('c'))
 
1007
        self.failIf(t.has('c'))
1016
1008
        self.check_transport_contents('c this file\n', t, 'b')
1017
1009
 
1018
1010
        # TODO: Try to write a test for atomicity
1050
1042
        except NotImplementedError:
1051
1043
            raise TestSkipped("Transport %s has no bogus URL support." %
1052
1044
                              self._server.__class__)
1053
 
        t = _mod_transport.get_transport_from_url(url)
 
1045
        t = get_transport(url)
1054
1046
        self.assertRaises((ConnectionError, NoSuchFile), t.get, '.bzr/branch')
1055
1047
 
1056
1048
    def test_stat(self):
1072
1064
        for path, size in zip(paths, sizes):
1073
1065
            st = t.stat(path)
1074
1066
            if path.endswith('/'):
1075
 
                self.assertTrue(S_ISDIR(st.st_mode))
 
1067
                self.failUnless(S_ISDIR(st.st_mode))
1076
1068
                # directory sizes are meaningless
1077
1069
            else:
1078
 
                self.assertTrue(S_ISREG(st.st_mode))
 
1070
                self.failUnless(S_ISREG(st.st_mode))
1079
1071
                self.assertEqual(size, st.st_size)
1080
1072
 
1081
1073
        remote_stats = list(t.stat_multi(paths))
1088
1080
        self.assertListRaises(NoSuchFile, t.stat_multi, iter(['a', 'c', 'd']))
1089
1081
        self.build_tree(['subdir/', 'subdir/file'], transport=t)
1090
1082
        subdir = t.clone('subdir')
1091
 
        st = subdir.stat('./file')
1092
 
        st = subdir.stat('.')
 
1083
        subdir.stat('./file')
 
1084
        subdir.stat('.')
1093
1085
 
1094
1086
    def test_hardlink(self):
1095
1087
        from stat import ST_NLINK
1104
1096
        try:
1105
1097
            t.hardlink(source_name, link_name)
1106
1098
 
1107
 
            self.assertTrue(t.has(source_name))
1108
 
            self.assertTrue(t.has(link_name))
 
1099
            self.failUnless(t.has(source_name))
 
1100
            self.failUnless(t.has(link_name))
1109
1101
 
1110
1102
            st = t.stat(link_name)
1111
 
            self.assertEqual(st[ST_NLINK], 2)
 
1103
            self.failUnlessEqual(st[ST_NLINK], 2)
1112
1104
        except TransportNotPossible:
1113
1105
            raise TestSkipped("Transport %s does not support hardlinks." %
1114
1106
                              self._server.__class__)
1126
1118
        try:
1127
1119
            t.symlink(source_name, link_name)
1128
1120
 
1129
 
            self.assertTrue(t.has(source_name))
1130
 
            self.assertTrue(t.has(link_name))
 
1121
            self.failUnless(t.has(source_name))
 
1122
            self.failUnless(t.has(link_name))
1131
1123
 
1132
1124
            st = t.stat(link_name)
1133
 
            self.assertTrue(S_ISLNK(st.st_mode),
1134
 
                "expected symlink, got mode %o" % st.st_mode)
 
1125
            self.failUnless(S_ISLNK(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")
1140
1131
 
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")
1207
1198
 
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)
1214
1205
 
1215
1206
    def test__reuse_for(self):
1216
1207
        t = self.get_transport()
1223
1214
 
1224
1215
            Only the parameters different from None will be changed.
1225
1216
            """
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)
1234
1225
 
1235
 
        if t._parsed_url.scheme == 'ftp':
 
1226
        if t._scheme == 'ftp':
1236
1227
            scheme = 'sftp'
1237
1228
        else:
1238
1229
            scheme = 'ftp'
1239
1230
        self.assertIsNot(t, t._reuse_for(new_url(scheme=scheme)))
1240
 
        if t._parsed_url.user == 'me':
 
1231
        if t._user == 'me':
1241
1232
            user = 'you'
1242
1233
        else:
1243
1234
            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')))
 
1249
        if t._port == 1234:
1259
1250
            port = 4321
1260
1251
        else:
1261
1252
            port = 1234
1274
1265
        self.assertIs(t._get_connection(), c._get_connection())
1275
1266
 
1276
1267
        # Temporary failure, we need to create a new dummy connection
1277
 
        new_connection = None
 
1268
        new_connection = object()
1278
1269
        t._set_connection(new_connection)
1279
1270
        # Check that both transports use the same connection
1280
1271
        self.assertIs(new_connection, t._get_connection())
1302
1293
 
1303
1294
        self.build_tree(['a', 'b/', 'b/c'], transport=t1)
1304
1295
 
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'))
1308
1299
 
1309
1300
        t2 = t1.clone('b')
1310
1301
        self.assertEqual(t1.base + 'b/', t2.base)
1311
1302
 
1312
 
        self.assertTrue(t2.has('c'))
1313
 
        self.assertFalse(t2.has('a'))
 
1303
        self.failUnless(t2.has('c'))
 
1304
        self.failIf(t2.has('a'))
1314
1305
 
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'))
1318
1309
 
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'))
1322
1313
 
1323
1314
        if t1.is_readonly():
1324
1315
            self.build_tree_contents([('b/d', 'newfile\n')])
1325
1316
        else:
1326
1317
            t2.put_bytes('d', 'newfile\n')
1327
1318
 
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'))
1331
1322
 
1332
1323
    def test_clone_to_root(self):
1333
1324
        orig_transport = self.get_transport()
1407
1398
        self.assertEqual(transport.clone("/").abspath('foo'),
1408
1399
                         transport.abspath("/foo"))
1409
1400
 
1410
 
    # GZ 2011-01-26: Test in per_transport but not using self.get_transport?
1411
1401
    def test_win32_abspath(self):
1412
1402
        # Note: we tried to set sys.platform='win32' so we could test on
1413
1403
        # other platforms too, but then osutils does platform specific
1418
1408
 
1419
1409
        # smoke test for abspath on win32.
1420
1410
        # 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:///")
 
1411
        transport = get_transport("file:///")
 
1412
        self.failUnlessEqual(transport.abspath("/"), "file:///")
1423
1413
 
1424
1414
        # 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:/")
 
1415
        transport = get_transport("file:///C:/")
 
1416
        self.failUnlessEqual(transport.abspath("/"), "file:///C:/")
1427
1417
 
1428
1418
    def test_local_abspath(self):
1429
1419
        transport = self.get_transport()
1555
1545
 
1556
1546
        no_unicode_support = getattr(self._server, 'no_unicode_support', False)
1557
1547
        if no_unicode_support:
1558
 
            self.knownFailure("test server cannot handle unicode paths")
 
1548
            raise tests.KnownFailure("test server cannot handle unicode paths")
1559
1549
 
1560
1550
        try:
1561
1551
            self.build_tree(files, transport=t, line_endings='binary')
1626
1616
    def test_readv(self):
1627
1617
        transport = self.get_transport()
1628
1618
        if transport.is_readonly():
1629
 
            with file('a', 'w') as f: f.write('0123456789')
 
1619
            file('a', 'w').write('0123456789')
1630
1620
        else:
1631
1621
            transport.put_bytes('a', '0123456789')
1632
1622
 
1642
1632
    def test_readv_out_of_order(self):
1643
1633
        transport = self.get_transport()
1644
1634
        if transport.is_readonly():
1645
 
            with file('a', 'w') as f: f.write('0123456789')
 
1635
            file('a', 'w').write('0123456789')
1646
1636
        else:
1647
1637
            transport.put_bytes('a', '01234567890')
1648
1638
 
1720
1710
        transport = self.get_transport()
1721
1711
        # test from observed failure case.
1722
1712
        if transport.is_readonly():
1723
 
            with file('a', 'w') as f: f.write('a'*1024*1024)
 
1713
            file('a', 'w').write('a'*1024*1024)
1724
1714
        else:
1725
1715
            transport.put_bytes('a', 'a'*1024*1024)
1726
1716
        broken_vector = [(465219, 800), (225221, 800), (445548, 800),
1760
1750
    def test_readv_short_read(self):
1761
1751
        transport = self.get_transport()
1762
1752
        if transport.is_readonly():
1763
 
            with file('a', 'w') as f: f.write('0123456789')
 
1753
            file('a', 'w').write('0123456789')
1764
1754
        else:
1765
1755
            transport.put_bytes('a', '01234567890')
1766
1756
 
1775
1765
        # also raise a special error
1776
1766
        self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),
1777
1767
                              transport.readv, 'a', [(12,2)])
1778
 
 
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())
1784
 
 
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())
1793
 
 
1794
 
    def test_stat_symlink(self):
1795
 
        # if a transport points directly to a symlink (and supports symlinks
1796
 
        # at all) you can tell this.  helps with bug 32669.
1797
 
        t = self.get_transport()
1798
 
        try:
1799
 
            t.symlink('target', 'link')
1800
 
        except TransportNotPossible:
1801
 
            raise TestSkipped("symlinks not supported")
1802
 
        t2 = t.clone('link')
1803
 
        st = t2.stat('')
1804
 
        self.assertTrue(stat.S_ISLNK(st.st_mode))