← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/per_transport.py

  • Committer: Jelmer Vernooij
  • Date: 2011-10-14 13:56:45 UTC
  • mfrom: (6215 +trunk)
  • mto: This revision was merged to the branch mainline in revision 6216.
  • Revision ID: jelmer@samba.org-20111014135645-phc3q3y21k2ks0s2
Merge bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/per_transport.py
1
 
# Copyright (C) 2004, 2005, 2006, 2007 Canonical Ltd
 
1
# Copyright (C) 2005-2011 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
30
29
 
31
30
from bzrlib import (
32
31
    errors,
33
32
    osutils,
 
33
    pyutils,
34
34
    tests,
 
35
    transport as _mod_transport,
35
36
    urlutils,
36
37
    )
37
38
from bzrlib.errors import (ConnectionError,
38
 
                           DirectoryNotEmpty,
39
39
                           FileExists,
40
40
                           InvalidURL,
41
 
                           LockError,
42
41
                           NoSuchFile,
43
 
                           NotLocalUrl,
44
42
                           PathError,
45
43
                           TransportNotPossible,
46
44
                           )
47
45
from bzrlib.osutils import getcwd
48
46
from bzrlib.smart import medium
49
47
from bzrlib.tests import (
50
 
    TestCaseInTempDir,
51
48
    TestSkipped,
52
49
    TestNotApplicable,
53
50
    multiply_tests,
54
51
    )
 
52
from bzrlib.tests import test_server
55
53
from bzrlib.tests.test_transport import TestTransportImplementation
56
54
from bzrlib.transport import (
57
55
    ConnectedTransport,
58
 
    get_transport,
59
56
    _get_transport_modules,
60
57
    )
61
58
from bzrlib.transport.memory import MemoryTransport
77
74
    for module in _get_transport_modules():
78
75
        try:
79
76
            permutations = get_transport_test_permutations(
80
 
                reduce(getattr, (module).split('.')[1:], __import__(module)))
 
77
                pyutils.get_named_object(module))
81
78
            for (klass, server_factory) in permutations:
82
 
                scenario = (server_factory.__name__,
 
79
                scenario = ('%s,%s' % (klass.__name__, server_factory.__name__),
83
80
                    {"transport_class":klass,
84
81
                     "transport_server":server_factory})
85
82
                result.append(scenario)
101
98
 
102
99
    def setUp(self):
103
100
        super(TransportTests, self).setUp()
104
 
        self._captureVar('BZR_NO_SMART_VFS', None)
 
101
        self.overrideEnv('BZR_NO_SMART_VFS', None)
105
102
 
106
103
    def check_transport_contents(self, content, transport, relpath):
107
 
        """Check that transport.get(relpath).read() == content."""
108
 
        self.assertEqualDiff(content, transport.get(relpath).read())
 
104
        """Check that transport.get_bytes(relpath) == content."""
 
105
        self.assertEqualDiff(content, transport.get_bytes(relpath))
109
106
 
110
107
    def test_ensure_base_missing(self):
111
108
        """.ensure_base() should create the directory if it doesn't exist"""
170
167
        self.assertEqual(True, t.has_any(['b', 'b', 'b']))
171
168
 
172
169
    def test_has_root_works(self):
173
 
        from bzrlib.smart import server
174
 
        if self.transport_server is server.SmartTCPServer_for_testing:
 
170
        if self.transport_server is test_server.SmartTCPServer_for_testing:
175
171
            raise TestNotApplicable(
176
172
                "SmartTCPServer_for_testing intentionally does not allow "
177
173
                "access to /.")
211
207
                    ]
212
208
        self.build_tree(files, transport=t, line_endings='binary')
213
209
        self.assertRaises(NoSuchFile, t.get, 'c')
214
 
        self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])
215
 
        self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', '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']))
216
221
 
217
222
    def test_get_directory_read_gives_ReadError(self):
218
223
        """consistent errors for read() on a file returned by get()."""
251
256
 
252
257
    def test_get_bytes_unknown_file(self):
253
258
        t = self.get_transport()
254
 
 
255
259
        self.assertRaises(NoSuchFile, t.get_bytes, 'c')
256
260
 
257
261
    def test_get_with_open_write_stream_sees_all_content(self):
261
265
        handle = t.open_write_stream('foo')
262
266
        try:
263
267
            handle.write('b')
264
 
            self.assertEqual('b', t.get('foo').read())
 
268
            self.assertEqual('b', t.get_bytes('foo'))
265
269
        finally:
266
270
            handle.close()
267
271
 
273
277
        try:
274
278
            handle.write('b')
275
279
            self.assertEqual('b', t.get_bytes('foo'))
276
 
            self.assertEqual('b', t.get('foo').read())
 
280
            f = t.get('foo')
 
281
            try:
 
282
                self.assertEqual('b', f.read())
 
283
            finally:
 
284
                f.close()
277
285
        finally:
278
286
            handle.close()
279
287
 
286
294
            return
287
295
 
288
296
        t.put_bytes('a', 'some text for a\n')
289
 
        self.failUnless(t.has('a'))
 
297
        self.assertTrue(t.has('a'))
290
298
        self.check_transport_contents('some text for a\n', t, 'a')
291
299
 
292
300
        # The contents should be overwritten
304
312
                    t.put_bytes_non_atomic, 'a', 'some text for a\n')
305
313
            return
306
314
 
307
 
        self.failIf(t.has('a'))
 
315
        self.assertFalse(t.has('a'))
308
316
        t.put_bytes_non_atomic('a', 'some text for a\n')
309
 
        self.failUnless(t.has('a'))
 
317
        self.assertTrue(t.has('a'))
310
318
        self.check_transport_contents('some text for a\n', t, 'a')
311
319
        # Put also replaces contents
312
320
        t.put_bytes_non_atomic('a', 'new\ncontents for\na\n')
324
332
        # Now test the create_parent flag
325
333
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'dir/a',
326
334
                                       'contents\n')
327
 
        self.failIf(t.has('dir/a'))
 
335
        self.assertFalse(t.has('dir/a'))
328
336
        t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',
329
337
                               create_parent_dir=True)
330
338
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
402
410
        result = t.put_file('a', StringIO('some text for a\n'))
403
411
        # put_file returns the length of the data written
404
412
        self.assertEqual(16, result)
405
 
        self.failUnless(t.has('a'))
 
413
        self.assertTrue(t.has('a'))
406
414
        self.check_transport_contents('some text for a\n', t, 'a')
407
415
        # Put also replaces contents
408
416
        result = t.put_file('a', StringIO('new\ncontents for\na\n'))
420
428
                    t.put_file_non_atomic, 'a', StringIO('some text for a\n'))
421
429
            return
422
430
 
423
 
        self.failIf(t.has('a'))
 
431
        self.assertFalse(t.has('a'))
424
432
        t.put_file_non_atomic('a', StringIO('some text for a\n'))
425
 
        self.failUnless(t.has('a'))
 
433
        self.assertTrue(t.has('a'))
426
434
        self.check_transport_contents('some text for a\n', t, 'a')
427
435
        # Put also replaces contents
428
436
        t.put_file_non_atomic('a', StringIO('new\ncontents for\na\n'))
440
448
        # Now test the create_parent flag
441
449
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'dir/a',
442
450
                                       StringIO('contents\n'))
443
 
        self.failIf(t.has('dir/a'))
 
451
        self.assertFalse(t.has('dir/a'))
444
452
        t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
445
453
                              create_parent_dir=True)
446
454
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
645
653
            self.build_tree(files, transport=transport_from)
646
654
            self.assertEqual(4, transport_from.copy_to(files, transport_to))
647
655
            for f in files:
648
 
                self.check_transport_contents(transport_to.get(f).read(),
 
656
                self.check_transport_contents(transport_to.get_bytes(f),
649
657
                                              transport_from, f)
650
658
 
651
659
        t = self.get_transport()
674
682
        files = ['a', 'b', 'c', 'd']
675
683
        t.copy_to(iter(files), temp_transport)
676
684
        for f in files:
677
 
            self.check_transport_contents(temp_transport.get(f).read(),
 
685
            self.check_transport_contents(temp_transport.get_bytes(f),
678
686
                                          t, f)
679
687
        del temp_transport
680
688
 
823
831
            return
824
832
 
825
833
        t.put_bytes('a', 'a little bit of text\n')
826
 
        self.failUnless(t.has('a'))
 
834
        self.assertTrue(t.has('a'))
827
835
        t.delete('a')
828
 
        self.failIf(t.has('a'))
 
836
        self.assertFalse(t.has('a'))
829
837
 
830
838
        self.assertRaises(NoSuchFile, t.delete, 'a')
831
839
 
837
845
        t.delete_multi(['a', 'c'])
838
846
        self.assertEqual([False, True, False],
839
847
                list(t.has_multi(['a', 'b', 'c'])))
840
 
        self.failIf(t.has('a'))
841
 
        self.failUnless(t.has('b'))
842
 
        self.failIf(t.has('c'))
 
848
        self.assertFalse(t.has('a'))
 
849
        self.assertTrue(t.has('b'))
 
850
        self.assertFalse(t.has('c'))
843
851
 
844
852
        self.assertRaises(NoSuchFile,
845
853
                t.delete_multi, ['a', 'b', 'c'])
906
914
        t.mkdir('foo-baz')
907
915
        t.rmdir('foo')
908
916
        self.assertRaises((NoSuchFile, PathError), t.rmdir, 'foo')
909
 
        self.failUnless(t.has('foo-bar'))
 
917
        self.assertTrue(t.has('foo-bar'))
910
918
 
911
919
    def test_rename_dir_succeeds(self):
912
920
        t = self.get_transport()
995
1003
        self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
996
1004
 
997
1005
        t.move('a', 'b')
998
 
        self.failUnless(t.has('b'))
999
 
        self.failIf(t.has('a'))
 
1006
        self.assertTrue(t.has('b'))
 
1007
        self.assertFalse(t.has('a'))
1000
1008
 
1001
1009
        self.check_transport_contents('a first file\n', t, 'b')
1002
1010
        self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
1004
1012
        # Overwrite a file
1005
1013
        t.put_bytes('c', 'c this file\n')
1006
1014
        t.move('c', 'b')
1007
 
        self.failIf(t.has('c'))
 
1015
        self.assertFalse(t.has('c'))
1008
1016
        self.check_transport_contents('c this file\n', t, 'b')
1009
1017
 
1010
1018
        # TODO: Try to write a test for atomicity
1042
1050
        except NotImplementedError:
1043
1051
            raise TestSkipped("Transport %s has no bogus URL support." %
1044
1052
                              self._server.__class__)
1045
 
        t = get_transport(url)
 
1053
        t = _mod_transport.get_transport_from_url(url)
1046
1054
        self.assertRaises((ConnectionError, NoSuchFile), t.get, '.bzr/branch')
1047
1055
 
1048
1056
    def test_stat(self):
1064
1072
        for path, size in zip(paths, sizes):
1065
1073
            st = t.stat(path)
1066
1074
            if path.endswith('/'):
1067
 
                self.failUnless(S_ISDIR(st.st_mode))
 
1075
                self.assertTrue(S_ISDIR(st.st_mode))
1068
1076
                # directory sizes are meaningless
1069
1077
            else:
1070
 
                self.failUnless(S_ISREG(st.st_mode))
 
1078
                self.assertTrue(S_ISREG(st.st_mode))
1071
1079
                self.assertEqual(size, st.st_size)
1072
1080
 
1073
1081
        remote_stats = list(t.stat_multi(paths))
1080
1088
        self.assertListRaises(NoSuchFile, t.stat_multi, iter(['a', 'c', 'd']))
1081
1089
        self.build_tree(['subdir/', 'subdir/file'], transport=t)
1082
1090
        subdir = t.clone('subdir')
1083
 
        subdir.stat('./file')
1084
 
        subdir.stat('.')
 
1091
        st = subdir.stat('./file')
 
1092
        st = subdir.stat('.')
 
1093
 
 
1094
    def test_hardlink(self):
 
1095
        from stat import ST_NLINK
 
1096
 
 
1097
        t = self.get_transport()
 
1098
 
 
1099
        source_name = "original_target"
 
1100
        link_name = "target_link"
 
1101
 
 
1102
        self.build_tree([source_name], transport=t)
 
1103
 
 
1104
        try:
 
1105
            t.hardlink(source_name, link_name)
 
1106
 
 
1107
            self.assertTrue(t.has(source_name))
 
1108
            self.assertTrue(t.has(link_name))
 
1109
 
 
1110
            st = t.stat(link_name)
 
1111
            self.assertEqual(st[ST_NLINK], 2)
 
1112
        except TransportNotPossible:
 
1113
            raise TestSkipped("Transport %s does not support hardlinks." %
 
1114
                              self._server.__class__)
 
1115
 
 
1116
    def test_symlink(self):
 
1117
        from stat import S_ISLNK
 
1118
 
 
1119
        t = self.get_transport()
 
1120
 
 
1121
        source_name = "original_target"
 
1122
        link_name = "target_link"
 
1123
 
 
1124
        self.build_tree([source_name], transport=t)
 
1125
 
 
1126
        try:
 
1127
            t.symlink(source_name, link_name)
 
1128
 
 
1129
            self.assertTrue(t.has(source_name))
 
1130
            self.assertTrue(t.has(link_name))
 
1131
 
 
1132
            st = t.stat(link_name)
 
1133
            self.assertTrue(S_ISLNK(st.st_mode),
 
1134
                "expected symlink, got mode %o" % st.st_mode)
 
1135
        except TransportNotPossible:
 
1136
            raise TestSkipped("Transport %s does not support symlinks." %
 
1137
                              self._server.__class__)
 
1138
        except IOError:
 
1139
            self.knownFailure("Paramiko fails to create symlinks during tests")
1085
1140
 
1086
1141
    def test_list_dir(self):
1087
1142
        # TODO: Test list_dir, just try once, and if it throws, stop testing
1151
1206
            raise TestSkipped("not a connected transport")
1152
1207
 
1153
1208
        t2 = t1.clone('subdir')
1154
 
        self.assertEquals(t1._scheme, t2._scheme)
1155
 
        self.assertEquals(t1._user, t2._user)
1156
 
        self.assertEquals(t1._password, t2._password)
1157
 
        self.assertEquals(t1._host, t2._host)
1158
 
        self.assertEquals(t1._port, t2._port)
 
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)
1159
1214
 
1160
1215
    def test__reuse_for(self):
1161
1216
        t = self.get_transport()
1168
1223
 
1169
1224
            Only the parameters different from None will be changed.
1170
1225
            """
1171
 
            if scheme   is None: scheme   = t._scheme
1172
 
            if user     is None: user     = t._user
1173
 
            if password is None: password = t._password
1174
 
            if user     is None: user     = t._user
1175
 
            if host     is None: host     = t._host
1176
 
            if port     is None: port     = t._port
1177
 
            if path     is None: path     = t._path
1178
 
            return t._unsplit_url(scheme, user, password, host, port, path)
 
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))
1179
1234
 
1180
 
        if t._scheme == 'ftp':
 
1235
        if t._parsed_url.scheme == 'ftp':
1181
1236
            scheme = 'sftp'
1182
1237
        else:
1183
1238
            scheme = 'ftp'
1184
1239
        self.assertIsNot(t, t._reuse_for(new_url(scheme=scheme)))
1185
 
        if t._user == 'me':
 
1240
        if t._parsed_url.user == 'me':
1186
1241
            user = 'you'
1187
1242
        else:
1188
1243
            user = 'me'
1199
1254
        #   (they may be typed by the user when prompted for example)
1200
1255
        self.assertIs(t, t._reuse_for(new_url(password='from space')))
1201
1256
        # We will not connect, we can use a invalid host
1202
 
        self.assertIsNot(t, t._reuse_for(new_url(host=t._host + 'bar')))
1203
 
        if t._port == 1234:
 
1257
        self.assertIsNot(t, t._reuse_for(new_url(host=t._parsed_url.host + 'bar')))
 
1258
        if t._parsed_url.port == 1234:
1204
1259
            port = 4321
1205
1260
        else:
1206
1261
            port = 1234
1219
1274
        self.assertIs(t._get_connection(), c._get_connection())
1220
1275
 
1221
1276
        # Temporary failure, we need to create a new dummy connection
1222
 
        new_connection = object()
 
1277
        new_connection = None
1223
1278
        t._set_connection(new_connection)
1224
1279
        # Check that both transports use the same connection
1225
1280
        self.assertIs(new_connection, t._get_connection())
1247
1302
 
1248
1303
        self.build_tree(['a', 'b/', 'b/c'], transport=t1)
1249
1304
 
1250
 
        self.failUnless(t1.has('a'))
1251
 
        self.failUnless(t1.has('b/c'))
1252
 
        self.failIf(t1.has('c'))
 
1305
        self.assertTrue(t1.has('a'))
 
1306
        self.assertTrue(t1.has('b/c'))
 
1307
        self.assertFalse(t1.has('c'))
1253
1308
 
1254
1309
        t2 = t1.clone('b')
1255
1310
        self.assertEqual(t1.base + 'b/', t2.base)
1256
1311
 
1257
 
        self.failUnless(t2.has('c'))
1258
 
        self.failIf(t2.has('a'))
 
1312
        self.assertTrue(t2.has('c'))
 
1313
        self.assertFalse(t2.has('a'))
1259
1314
 
1260
1315
        t3 = t2.clone('..')
1261
 
        self.failUnless(t3.has('a'))
1262
 
        self.failIf(t3.has('c'))
 
1316
        self.assertTrue(t3.has('a'))
 
1317
        self.assertFalse(t3.has('c'))
1263
1318
 
1264
 
        self.failIf(t1.has('b/d'))
1265
 
        self.failIf(t2.has('d'))
1266
 
        self.failIf(t3.has('b/d'))
 
1319
        self.assertFalse(t1.has('b/d'))
 
1320
        self.assertFalse(t2.has('d'))
 
1321
        self.assertFalse(t3.has('b/d'))
1267
1322
 
1268
1323
        if t1.is_readonly():
1269
1324
            self.build_tree_contents([('b/d', 'newfile\n')])
1270
1325
        else:
1271
1326
            t2.put_bytes('d', 'newfile\n')
1272
1327
 
1273
 
        self.failUnless(t1.has('b/d'))
1274
 
        self.failUnless(t2.has('d'))
1275
 
        self.failUnless(t3.has('b/d'))
 
1328
        self.assertTrue(t1.has('b/d'))
 
1329
        self.assertTrue(t2.has('d'))
 
1330
        self.assertTrue(t3.has('b/d'))
1276
1331
 
1277
1332
    def test_clone_to_root(self):
1278
1333
        orig_transport = self.get_transport()
1352
1407
        self.assertEqual(transport.clone("/").abspath('foo'),
1353
1408
                         transport.abspath("/foo"))
1354
1409
 
 
1410
    # GZ 2011-01-26: Test in per_transport but not using self.get_transport?
1355
1411
    def test_win32_abspath(self):
1356
1412
        # Note: we tried to set sys.platform='win32' so we could test on
1357
1413
        # other platforms too, but then osutils does platform specific
1362
1418
 
1363
1419
        # smoke test for abspath on win32.
1364
1420
        # a transport based on 'file:///' never fully qualifies the drive.
1365
 
        transport = get_transport("file:///")
1366
 
        self.failUnlessEqual(transport.abspath("/"), "file:///")
 
1421
        transport = _mod_transport.get_transport_from_url("file:///")
 
1422
        self.assertEqual(transport.abspath("/"), "file:///")
1367
1423
 
1368
1424
        # but a transport that starts with a drive spec must keep it.
1369
 
        transport = get_transport("file:///C:/")
1370
 
        self.failUnlessEqual(transport.abspath("/"), "file:///C:/")
 
1425
        transport = _mod_transport.get_transport_from_url("file:///C:/")
 
1426
        self.assertEqual(transport.abspath("/"), "file:///C:/")
1371
1427
 
1372
1428
    def test_local_abspath(self):
1373
1429
        transport = self.get_transport()
1497
1553
                 u'\u65e5', # Kanji person
1498
1554
                ]
1499
1555
 
 
1556
        no_unicode_support = getattr(self._server, 'no_unicode_support', False)
 
1557
        if no_unicode_support:
 
1558
            self.knownFailure("test server cannot handle unicode paths")
 
1559
 
1500
1560
        try:
1501
1561
            self.build_tree(files, transport=t, line_endings='binary')
1502
1562
        except UnicodeError:
1566
1626
    def test_readv(self):
1567
1627
        transport = self.get_transport()
1568
1628
        if transport.is_readonly():
1569
 
            file('a', 'w').write('0123456789')
 
1629
            with file('a', 'w') as f: f.write('0123456789')
1570
1630
        else:
1571
1631
            transport.put_bytes('a', '0123456789')
1572
1632
 
1582
1642
    def test_readv_out_of_order(self):
1583
1643
        transport = self.get_transport()
1584
1644
        if transport.is_readonly():
1585
 
            file('a', 'w').write('0123456789')
 
1645
            with file('a', 'w') as f: f.write('0123456789')
1586
1646
        else:
1587
1647
            transport.put_bytes('a', '01234567890')
1588
1648
 
1660
1720
        transport = self.get_transport()
1661
1721
        # test from observed failure case.
1662
1722
        if transport.is_readonly():
1663
 
            file('a', 'w').write('a'*1024*1024)
 
1723
            with file('a', 'w') as f: f.write('a'*1024*1024)
1664
1724
        else:
1665
1725
            transport.put_bytes('a', 'a'*1024*1024)
1666
1726
        broken_vector = [(465219, 800), (225221, 800), (445548, 800),
1700
1760
    def test_readv_short_read(self):
1701
1761
        transport = self.get_transport()
1702
1762
        if transport.is_readonly():
1703
 
            file('a', 'w').write('0123456789')
 
1763
            with file('a', 'w') as f: f.write('0123456789')
1704
1764
        else:
1705
1765
            transport.put_bytes('a', '01234567890')
1706
1766
 
1715
1775
        # also raise a special error
1716
1776
        self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),
1717
1777
                              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))