← 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: Patch Queue Manager
  • Date: 2015-09-30 16:43:21 UTC
  • mfrom: (6603.2.2 fix-keep-dirty)
  • Revision ID: pqm@pqm.ubuntu.com-20150930164321-ct2v2qnmvimqt8qf
(vila) Avoid associating dirty patch headers with the previous file in the
 patch. (Colin Watson)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/per_transport.py
1
 
# Copyright (C) 2005-2010 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,
56
53
from bzrlib.tests.test_transport import TestTransportImplementation
57
54
from bzrlib.transport import (
58
55
    ConnectedTransport,
59
 
    get_transport,
 
56
    Transport,
60
57
    _get_transport_modules,
61
58
    )
62
59
from bzrlib.transport.memory import MemoryTransport
 
60
from bzrlib.transport.remote import RemoteTransport
63
61
 
64
62
 
65
63
def get_transport_test_permutations(module):
78
76
    for module in _get_transport_modules():
79
77
        try:
80
78
            permutations = get_transport_test_permutations(
81
 
                reduce(getattr, (module).split('.')[1:], __import__(module)))
 
79
                pyutils.get_named_object(module))
82
80
            for (klass, server_factory) in permutations:
83
81
                scenario = ('%s,%s' % (klass.__name__, server_factory.__name__),
84
82
                    {"transport_class":klass,
102
100
 
103
101
    def setUp(self):
104
102
        super(TransportTests, self).setUp()
105
 
        self._captureVar('BZR_NO_SMART_VFS', None)
 
103
        self.overrideEnv('BZR_NO_SMART_VFS', None)
106
104
 
107
105
    def check_transport_contents(self, content, transport, relpath):
108
 
        """Check that transport.get(relpath).read() == content."""
109
 
        self.assertEqualDiff(content, transport.get(relpath).read())
 
106
        """Check that transport.get_bytes(relpath) == content."""
 
107
        self.assertEqualDiff(content, transport.get_bytes(relpath))
110
108
 
111
109
    def test_ensure_base_missing(self):
112
110
        """.ensure_base() should create the directory if it doesn't exist"""
211
209
                    ]
212
210
        self.build_tree(files, transport=t, line_endings='binary')
213
211
        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']))
 
212
        def iterate_and_close(func, *args):
 
213
            for f in func(*args):
 
214
                # We call f.read() here because things like paramiko actually
 
215
                # spawn a thread to prefetch the content, which we want to
 
216
                # consume before we close the handle.
 
217
                content = f.read()
 
218
                f.close()
 
219
        self.assertRaises(NoSuchFile, iterate_and_close,
 
220
                          t.get_multi, ['a', 'b', 'c'])
 
221
        self.assertRaises(NoSuchFile, iterate_and_close,
 
222
                          t.get_multi, iter(['a', 'b', 'c']))
216
223
 
217
224
    def test_get_directory_read_gives_ReadError(self):
218
225
        """consistent errors for read() on a file returned by get()."""
251
258
 
252
259
    def test_get_bytes_unknown_file(self):
253
260
        t = self.get_transport()
254
 
 
255
261
        self.assertRaises(NoSuchFile, t.get_bytes, 'c')
256
262
 
257
263
    def test_get_with_open_write_stream_sees_all_content(self):
261
267
        handle = t.open_write_stream('foo')
262
268
        try:
263
269
            handle.write('b')
264
 
            self.assertEqual('b', t.get('foo').read())
 
270
            self.assertEqual('b', t.get_bytes('foo'))
265
271
        finally:
266
272
            handle.close()
267
273
 
273
279
        try:
274
280
            handle.write('b')
275
281
            self.assertEqual('b', t.get_bytes('foo'))
276
 
            self.assertEqual('b', t.get('foo').read())
 
282
            f = t.get('foo')
 
283
            try:
 
284
                self.assertEqual('b', f.read())
 
285
            finally:
 
286
                f.close()
277
287
        finally:
278
288
            handle.close()
279
289
 
286
296
            return
287
297
 
288
298
        t.put_bytes('a', 'some text for a\n')
289
 
        self.failUnless(t.has('a'))
 
299
        self.assertTrue(t.has('a'))
290
300
        self.check_transport_contents('some text for a\n', t, 'a')
291
301
 
292
302
        # The contents should be overwritten
304
314
                    t.put_bytes_non_atomic, 'a', 'some text for a\n')
305
315
            return
306
316
 
307
 
        self.failIf(t.has('a'))
 
317
        self.assertFalse(t.has('a'))
308
318
        t.put_bytes_non_atomic('a', 'some text for a\n')
309
 
        self.failUnless(t.has('a'))
 
319
        self.assertTrue(t.has('a'))
310
320
        self.check_transport_contents('some text for a\n', t, 'a')
311
321
        # Put also replaces contents
312
322
        t.put_bytes_non_atomic('a', 'new\ncontents for\na\n')
324
334
        # Now test the create_parent flag
325
335
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'dir/a',
326
336
                                       'contents\n')
327
 
        self.failIf(t.has('dir/a'))
 
337
        self.assertFalse(t.has('dir/a'))
328
338
        t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',
329
339
                               create_parent_dir=True)
330
340
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
402
412
        result = t.put_file('a', StringIO('some text for a\n'))
403
413
        # put_file returns the length of the data written
404
414
        self.assertEqual(16, result)
405
 
        self.failUnless(t.has('a'))
 
415
        self.assertTrue(t.has('a'))
406
416
        self.check_transport_contents('some text for a\n', t, 'a')
407
417
        # Put also replaces contents
408
418
        result = t.put_file('a', StringIO('new\ncontents for\na\n'))
420
430
                    t.put_file_non_atomic, 'a', StringIO('some text for a\n'))
421
431
            return
422
432
 
423
 
        self.failIf(t.has('a'))
 
433
        self.assertFalse(t.has('a'))
424
434
        t.put_file_non_atomic('a', StringIO('some text for a\n'))
425
 
        self.failUnless(t.has('a'))
 
435
        self.assertTrue(t.has('a'))
426
436
        self.check_transport_contents('some text for a\n', t, 'a')
427
437
        # Put also replaces contents
428
438
        t.put_file_non_atomic('a', StringIO('new\ncontents for\na\n'))
440
450
        # Now test the create_parent flag
441
451
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'dir/a',
442
452
                                       StringIO('contents\n'))
443
 
        self.failIf(t.has('dir/a'))
 
453
        self.assertFalse(t.has('dir/a'))
444
454
        t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
445
455
                              create_parent_dir=True)
446
456
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
645
655
            self.build_tree(files, transport=transport_from)
646
656
            self.assertEqual(4, transport_from.copy_to(files, transport_to))
647
657
            for f in files:
648
 
                self.check_transport_contents(transport_to.get(f).read(),
 
658
                self.check_transport_contents(transport_to.get_bytes(f),
649
659
                                              transport_from, f)
650
660
 
651
661
        t = self.get_transport()
674
684
        files = ['a', 'b', 'c', 'd']
675
685
        t.copy_to(iter(files), temp_transport)
676
686
        for f in files:
677
 
            self.check_transport_contents(temp_transport.get(f).read(),
 
687
            self.check_transport_contents(temp_transport.get_bytes(f),
678
688
                                          t, f)
679
689
        del temp_transport
680
690
 
823
833
            return
824
834
 
825
835
        t.put_bytes('a', 'a little bit of text\n')
826
 
        self.failUnless(t.has('a'))
 
836
        self.assertTrue(t.has('a'))
827
837
        t.delete('a')
828
 
        self.failIf(t.has('a'))
 
838
        self.assertFalse(t.has('a'))
829
839
 
830
840
        self.assertRaises(NoSuchFile, t.delete, 'a')
831
841
 
837
847
        t.delete_multi(['a', 'c'])
838
848
        self.assertEqual([False, True, False],
839
849
                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'))
 
850
        self.assertFalse(t.has('a'))
 
851
        self.assertTrue(t.has('b'))
 
852
        self.assertFalse(t.has('c'))
843
853
 
844
854
        self.assertRaises(NoSuchFile,
845
855
                t.delete_multi, ['a', 'b', 'c'])
906
916
        t.mkdir('foo-baz')
907
917
        t.rmdir('foo')
908
918
        self.assertRaises((NoSuchFile, PathError), t.rmdir, 'foo')
909
 
        self.failUnless(t.has('foo-bar'))
 
919
        self.assertTrue(t.has('foo-bar'))
910
920
 
911
921
    def test_rename_dir_succeeds(self):
912
922
        t = self.get_transport()
995
1005
        self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
996
1006
 
997
1007
        t.move('a', 'b')
998
 
        self.failUnless(t.has('b'))
999
 
        self.failIf(t.has('a'))
 
1008
        self.assertTrue(t.has('b'))
 
1009
        self.assertFalse(t.has('a'))
1000
1010
 
1001
1011
        self.check_transport_contents('a first file\n', t, 'b')
1002
1012
        self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
1004
1014
        # Overwrite a file
1005
1015
        t.put_bytes('c', 'c this file\n')
1006
1016
        t.move('c', 'b')
1007
 
        self.failIf(t.has('c'))
 
1017
        self.assertFalse(t.has('c'))
1008
1018
        self.check_transport_contents('c this file\n', t, 'b')
1009
1019
 
1010
1020
        # TODO: Try to write a test for atomicity
1042
1052
        except NotImplementedError:
1043
1053
            raise TestSkipped("Transport %s has no bogus URL support." %
1044
1054
                              self._server.__class__)
1045
 
        t = get_transport(url)
 
1055
        t = _mod_transport.get_transport_from_url(url)
1046
1056
        self.assertRaises((ConnectionError, NoSuchFile), t.get, '.bzr/branch')
1047
1057
 
1048
1058
    def test_stat(self):
1064
1074
        for path, size in zip(paths, sizes):
1065
1075
            st = t.stat(path)
1066
1076
            if path.endswith('/'):
1067
 
                self.failUnless(S_ISDIR(st.st_mode))
 
1077
                self.assertTrue(S_ISDIR(st.st_mode))
1068
1078
                # directory sizes are meaningless
1069
1079
            else:
1070
 
                self.failUnless(S_ISREG(st.st_mode))
 
1080
                self.assertTrue(S_ISREG(st.st_mode))
1071
1081
                self.assertEqual(size, st.st_size)
1072
1082
 
1073
1083
        remote_stats = list(t.stat_multi(paths))
1080
1090
        self.assertListRaises(NoSuchFile, t.stat_multi, iter(['a', 'c', 'd']))
1081
1091
        self.build_tree(['subdir/', 'subdir/file'], transport=t)
1082
1092
        subdir = t.clone('subdir')
1083
 
        subdir.stat('./file')
1084
 
        subdir.stat('.')
 
1093
        st = subdir.stat('./file')
 
1094
        st = subdir.stat('.')
1085
1095
 
1086
1096
    def test_hardlink(self):
1087
1097
        from stat import ST_NLINK
1096
1106
        try:
1097
1107
            t.hardlink(source_name, link_name)
1098
1108
 
1099
 
            self.failUnless(t.has(source_name))
1100
 
            self.failUnless(t.has(link_name))
 
1109
            self.assertTrue(t.has(source_name))
 
1110
            self.assertTrue(t.has(link_name))
1101
1111
 
1102
1112
            st = t.stat(link_name)
1103
 
            self.failUnlessEqual(st[ST_NLINK], 2)
 
1113
            self.assertEqual(st[ST_NLINK], 2)
1104
1114
        except TransportNotPossible:
1105
1115
            raise TestSkipped("Transport %s does not support hardlinks." %
1106
1116
                              self._server.__class__)
1118
1128
        try:
1119
1129
            t.symlink(source_name, link_name)
1120
1130
 
1121
 
            self.failUnless(t.has(source_name))
1122
 
            self.failUnless(t.has(link_name))
 
1131
            self.assertTrue(t.has(source_name))
 
1132
            self.assertTrue(t.has(link_name))
1123
1133
 
1124
1134
            st = t.stat(link_name)
1125
 
            self.failUnless(S_ISLNK(st.st_mode))
 
1135
            self.assertTrue(S_ISLNK(st.st_mode),
 
1136
                "expected symlink, got mode %o" % st.st_mode)
1126
1137
        except TransportNotPossible:
1127
1138
            raise TestSkipped("Transport %s does not support symlinks." %
1128
1139
                              self._server.__class__)
1129
1140
        except IOError:
1130
 
            raise tests.KnownFailure("Paramiko fails to create symlinks during tests")
 
1141
            self.knownFailure("Paramiko fails to create symlinks during tests")
1131
1142
 
1132
1143
    def test_list_dir(self):
1133
1144
        # TODO: Test list_dir, just try once, and if it throws, stop testing
1197
1208
            raise TestSkipped("not a connected transport")
1198
1209
 
1199
1210
        t2 = t1.clone('subdir')
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)
 
1211
        self.assertEquals(t1._parsed_url.scheme, t2._parsed_url.scheme)
 
1212
        self.assertEquals(t1._parsed_url.user, t2._parsed_url.user)
 
1213
        self.assertEquals(t1._parsed_url.password, t2._parsed_url.password)
 
1214
        self.assertEquals(t1._parsed_url.host, t2._parsed_url.host)
 
1215
        self.assertEquals(t1._parsed_url.port, t2._parsed_url.port)
1205
1216
 
1206
1217
    def test__reuse_for(self):
1207
1218
        t = self.get_transport()
1214
1225
 
1215
1226
            Only the parameters different from None will be changed.
1216
1227
            """
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)
 
1228
            if scheme   is None: scheme   = t._parsed_url.scheme
 
1229
            if user     is None: user     = t._parsed_url.user
 
1230
            if password is None: password = t._parsed_url.password
 
1231
            if user     is None: user     = t._parsed_url.user
 
1232
            if host     is None: host     = t._parsed_url.host
 
1233
            if port     is None: port     = t._parsed_url.port
 
1234
            if path     is None: path     = t._parsed_url.path
 
1235
            return str(urlutils.URL(scheme, user, password, host, port, path))
1225
1236
 
1226
 
        if t._scheme == 'ftp':
 
1237
        if t._parsed_url.scheme == 'ftp':
1227
1238
            scheme = 'sftp'
1228
1239
        else:
1229
1240
            scheme = 'ftp'
1230
1241
        self.assertIsNot(t, t._reuse_for(new_url(scheme=scheme)))
1231
 
        if t._user == 'me':
 
1242
        if t._parsed_url.user == 'me':
1232
1243
            user = 'you'
1233
1244
        else:
1234
1245
            user = 'me'
1245
1256
        #   (they may be typed by the user when prompted for example)
1246
1257
        self.assertIs(t, t._reuse_for(new_url(password='from space')))
1247
1258
        # We will not connect, we can use a invalid host
1248
 
        self.assertIsNot(t, t._reuse_for(new_url(host=t._host + 'bar')))
1249
 
        if t._port == 1234:
 
1259
        self.assertIsNot(t, t._reuse_for(new_url(host=t._parsed_url.host + 'bar')))
 
1260
        if t._parsed_url.port == 1234:
1250
1261
            port = 4321
1251
1262
        else:
1252
1263
            port = 1234
1265
1276
        self.assertIs(t._get_connection(), c._get_connection())
1266
1277
 
1267
1278
        # Temporary failure, we need to create a new dummy connection
1268
 
        new_connection = object()
 
1279
        new_connection = None
1269
1280
        t._set_connection(new_connection)
1270
1281
        # Check that both transports use the same connection
1271
1282
        self.assertIs(new_connection, t._get_connection())
1293
1304
 
1294
1305
        self.build_tree(['a', 'b/', 'b/c'], transport=t1)
1295
1306
 
1296
 
        self.failUnless(t1.has('a'))
1297
 
        self.failUnless(t1.has('b/c'))
1298
 
        self.failIf(t1.has('c'))
 
1307
        self.assertTrue(t1.has('a'))
 
1308
        self.assertTrue(t1.has('b/c'))
 
1309
        self.assertFalse(t1.has('c'))
1299
1310
 
1300
1311
        t2 = t1.clone('b')
1301
1312
        self.assertEqual(t1.base + 'b/', t2.base)
1302
1313
 
1303
 
        self.failUnless(t2.has('c'))
1304
 
        self.failIf(t2.has('a'))
 
1314
        self.assertTrue(t2.has('c'))
 
1315
        self.assertFalse(t2.has('a'))
1305
1316
 
1306
1317
        t3 = t2.clone('..')
1307
 
        self.failUnless(t3.has('a'))
1308
 
        self.failIf(t3.has('c'))
 
1318
        self.assertTrue(t3.has('a'))
 
1319
        self.assertFalse(t3.has('c'))
1309
1320
 
1310
 
        self.failIf(t1.has('b/d'))
1311
 
        self.failIf(t2.has('d'))
1312
 
        self.failIf(t3.has('b/d'))
 
1321
        self.assertFalse(t1.has('b/d'))
 
1322
        self.assertFalse(t2.has('d'))
 
1323
        self.assertFalse(t3.has('b/d'))
1313
1324
 
1314
1325
        if t1.is_readonly():
1315
1326
            self.build_tree_contents([('b/d', 'newfile\n')])
1316
1327
        else:
1317
1328
            t2.put_bytes('d', 'newfile\n')
1318
1329
 
1319
 
        self.failUnless(t1.has('b/d'))
1320
 
        self.failUnless(t2.has('d'))
1321
 
        self.failUnless(t3.has('b/d'))
 
1330
        self.assertTrue(t1.has('b/d'))
 
1331
        self.assertTrue(t2.has('d'))
 
1332
        self.assertTrue(t3.has('b/d'))
1322
1333
 
1323
1334
    def test_clone_to_root(self):
1324
1335
        orig_transport = self.get_transport()
1398
1409
        self.assertEqual(transport.clone("/").abspath('foo'),
1399
1410
                         transport.abspath("/foo"))
1400
1411
 
 
1412
    # GZ 2011-01-26: Test in per_transport but not using self.get_transport?
1401
1413
    def test_win32_abspath(self):
1402
1414
        # Note: we tried to set sys.platform='win32' so we could test on
1403
1415
        # other platforms too, but then osutils does platform specific
1408
1420
 
1409
1421
        # smoke test for abspath on win32.
1410
1422
        # a transport based on 'file:///' never fully qualifies the drive.
1411
 
        transport = get_transport("file:///")
1412
 
        self.failUnlessEqual(transport.abspath("/"), "file:///")
 
1423
        transport = _mod_transport.get_transport_from_url("file:///")
 
1424
        self.assertEqual(transport.abspath("/"), "file:///")
1413
1425
 
1414
1426
        # but a transport that starts with a drive spec must keep it.
1415
 
        transport = get_transport("file:///C:/")
1416
 
        self.failUnlessEqual(transport.abspath("/"), "file:///C:/")
 
1427
        transport = _mod_transport.get_transport_from_url("file:///C:/")
 
1428
        self.assertEqual(transport.abspath("/"), "file:///C:/")
1417
1429
 
1418
1430
    def test_local_abspath(self):
1419
1431
        transport = self.get_transport()
1545
1557
 
1546
1558
        no_unicode_support = getattr(self._server, 'no_unicode_support', False)
1547
1559
        if no_unicode_support:
1548
 
            raise tests.KnownFailure("test server cannot handle unicode paths")
 
1560
            self.knownFailure("test server cannot handle unicode paths")
1549
1561
 
1550
1562
        try:
1551
1563
            self.build_tree(files, transport=t, line_endings='binary')
1616
1628
    def test_readv(self):
1617
1629
        transport = self.get_transport()
1618
1630
        if transport.is_readonly():
1619
 
            file('a', 'w').write('0123456789')
 
1631
            with file('a', 'w') as f: f.write('0123456789')
1620
1632
        else:
1621
1633
            transport.put_bytes('a', '0123456789')
1622
1634
 
1632
1644
    def test_readv_out_of_order(self):
1633
1645
        transport = self.get_transport()
1634
1646
        if transport.is_readonly():
1635
 
            file('a', 'w').write('0123456789')
 
1647
            with file('a', 'w') as f: f.write('0123456789')
1636
1648
        else:
1637
1649
            transport.put_bytes('a', '01234567890')
1638
1650
 
1710
1722
        transport = self.get_transport()
1711
1723
        # test from observed failure case.
1712
1724
        if transport.is_readonly():
1713
 
            file('a', 'w').write('a'*1024*1024)
 
1725
            with file('a', 'w') as f: f.write('a'*1024*1024)
1714
1726
        else:
1715
1727
            transport.put_bytes('a', 'a'*1024*1024)
1716
1728
        broken_vector = [(465219, 800), (225221, 800), (445548, 800),
1750
1762
    def test_readv_short_read(self):
1751
1763
        transport = self.get_transport()
1752
1764
        if transport.is_readonly():
1753
 
            file('a', 'w').write('0123456789')
 
1765
            with file('a', 'w') as f: f.write('0123456789')
1754
1766
        else:
1755
1767
            transport.put_bytes('a', '01234567890')
1756
1768
 
1765
1777
        # also raise a special error
1766
1778
        self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),
1767
1779
                              transport.readv, 'a', [(12,2)])
 
1780
 
 
1781
    def test_no_segment_parameters(self):
 
1782
        """Segment parameters should be stripped and stored in
 
1783
        transport.segment_parameters."""
 
1784
        transport = self.get_transport("foo")
 
1785
        self.assertEquals({}, transport.get_segment_parameters())
 
1786
 
 
1787
    def test_segment_parameters(self):
 
1788
        """Segment parameters should be stripped and stored in
 
1789
        transport.get_segment_parameters()."""
 
1790
        base_url = self._server.get_url()
 
1791
        parameters = {"key1": "val1", "key2": "val2"}
 
1792
        url = urlutils.join_segment_parameters(base_url, parameters)
 
1793
        transport = _mod_transport.get_transport_from_url(url)
 
1794
        self.assertEquals(parameters, transport.get_segment_parameters())
 
1795
 
 
1796
    def test_set_segment_parameters(self):
 
1797
        """Segment parameters can be set and show up in base."""
 
1798
        transport = self.get_transport("foo")
 
1799
        orig_base = transport.base
 
1800
        transport.set_segment_parameter("arm", "board")
 
1801
        self.assertEquals("%s,arm=board" % orig_base, transport.base)
 
1802
        self.assertEquals({"arm": "board"}, transport.get_segment_parameters())
 
1803
        transport.set_segment_parameter("arm", None)
 
1804
        transport.set_segment_parameter("nonexistant", None)
 
1805
        self.assertEquals({}, transport.get_segment_parameters())
 
1806
        self.assertEquals(orig_base, transport.base)
 
1807
 
 
1808
    def test_stat_symlink(self):
 
1809
        # if a transport points directly to a symlink (and supports symlinks
 
1810
        # at all) you can tell this.  helps with bug 32669.
 
1811
        t = self.get_transport()
 
1812
        try:
 
1813
            t.symlink('target', 'link')
 
1814
        except TransportNotPossible:
 
1815
            raise TestSkipped("symlinks not supported")
 
1816
        t2 = t.clone('link')
 
1817
        st = t2.stat('')
 
1818
        self.assertTrue(stat.S_ISLNK(st.st_mode))
 
1819
 
 
1820
    def test_abspath_url_unquote_unreserved(self):
 
1821
        """URLs from abspath should have unreserved characters unquoted
 
1822
        
 
1823
        Need consistent quoting notably for tildes, see lp:842223 for more.
 
1824
        """
 
1825
        t = self.get_transport()
 
1826
        needlessly_escaped_dir = "%2D%2E%30%39%41%5A%5F%61%7A%7E/"
 
1827
        self.assertEqual(t.base + "-.09AZ_az~",
 
1828
            t.abspath(needlessly_escaped_dir))
 
1829
 
 
1830
    def test_clone_url_unquote_unreserved(self):
 
1831
        """Base URL of a cloned branch needs unreserved characters unquoted
 
1832
        
 
1833
        Cloned transports should be prefix comparable for things like the
 
1834
        isolation checking of tests, see lp:842223 for more.
 
1835
        """
 
1836
        t1 = self.get_transport()
 
1837
        needlessly_escaped_dir = "%2D%2E%30%39%41%5A%5F%61%7A%7E/"
 
1838
        self.build_tree([needlessly_escaped_dir], transport=t1)
 
1839
        t2 = t1.clone(needlessly_escaped_dir)
 
1840
        self.assertEqual(t1.base + "-.09AZ_az~/", t2.base)
 
1841
 
 
1842
    def test_hook_post_connection_one(self):
 
1843
        """Fire post_connect hook after a ConnectedTransport is first used"""
 
1844
        log = []
 
1845
        Transport.hooks.install_named_hook("post_connect", log.append, None)
 
1846
        t = self.get_transport()
 
1847
        self.assertEqual([], log)
 
1848
        t.has("non-existant")
 
1849
        if isinstance(t, RemoteTransport):
 
1850
            self.assertEqual([t.get_smart_medium()], log)
 
1851
        elif isinstance(t, ConnectedTransport):
 
1852
            self.assertEqual([t], log)
 
1853
        else:
 
1854
            self.assertEqual([], log)
 
1855
 
 
1856
    def test_hook_post_connection_multi(self):
 
1857
        """Fire post_connect hook once per unshared underlying connection"""
 
1858
        log = []
 
1859
        Transport.hooks.install_named_hook("post_connect", log.append, None)
 
1860
        t1 = self.get_transport()
 
1861
        t2 = t1.clone(".")
 
1862
        t3 = self.get_transport()
 
1863
        self.assertEqual([], log)
 
1864
        t1.has("x")
 
1865
        t2.has("x")
 
1866
        t3.has("x")
 
1867
        if isinstance(t1, RemoteTransport):
 
1868
            self.assertEqual([t.get_smart_medium() for t in [t1, t3]], log)
 
1869
        elif isinstance(t1, ConnectedTransport):
 
1870
            self.assertEqual([t1, t3], log)
 
1871
        else:
 
1872
            self.assertEqual([], log)