← 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: Kit Randel
  • Date: 2014-12-15 20:24:42 UTC
  • mto: This revision was merged to the branch mainline in revision 6602.
  • Revision ID: kit.randel@canonical.com-20141215202442-usf2ixhypqg8yh6q
added a note for bug-1400567 to the 2.7b release notes

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()."""
260
267
        handle = t.open_write_stream('foo')
261
268
        try:
262
269
            handle.write('b')
263
 
            self.assertEqual('b', t.get('foo').read())
 
270
            self.assertEqual('b', t.get_bytes('foo'))
264
271
        finally:
265
272
            handle.close()
266
273
 
272
279
        try:
273
280
            handle.write('b')
274
281
            self.assertEqual('b', t.get_bytes('foo'))
275
 
            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()
276
287
        finally:
277
288
            handle.close()
278
289
 
285
296
            return
286
297
 
287
298
        t.put_bytes('a', 'some text for a\n')
288
 
        self.failUnless(t.has('a'))
 
299
        self.assertTrue(t.has('a'))
289
300
        self.check_transport_contents('some text for a\n', t, 'a')
290
301
 
291
302
        # The contents should be overwritten
303
314
                    t.put_bytes_non_atomic, 'a', 'some text for a\n')
304
315
            return
305
316
 
306
 
        self.failIf(t.has('a'))
 
317
        self.assertFalse(t.has('a'))
307
318
        t.put_bytes_non_atomic('a', 'some text for a\n')
308
 
        self.failUnless(t.has('a'))
 
319
        self.assertTrue(t.has('a'))
309
320
        self.check_transport_contents('some text for a\n', t, 'a')
310
321
        # Put also replaces contents
311
322
        t.put_bytes_non_atomic('a', 'new\ncontents for\na\n')
323
334
        # Now test the create_parent flag
324
335
        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'dir/a',
325
336
                                       'contents\n')
326
 
        self.failIf(t.has('dir/a'))
 
337
        self.assertFalse(t.has('dir/a'))
327
338
        t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',
328
339
                               create_parent_dir=True)
329
340
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
401
412
        result = t.put_file('a', StringIO('some text for a\n'))
402
413
        # put_file returns the length of the data written
403
414
        self.assertEqual(16, result)
404
 
        self.failUnless(t.has('a'))
 
415
        self.assertTrue(t.has('a'))
405
416
        self.check_transport_contents('some text for a\n', t, 'a')
406
417
        # Put also replaces contents
407
418
        result = t.put_file('a', StringIO('new\ncontents for\na\n'))
419
430
                    t.put_file_non_atomic, 'a', StringIO('some text for a\n'))
420
431
            return
421
432
 
422
 
        self.failIf(t.has('a'))
 
433
        self.assertFalse(t.has('a'))
423
434
        t.put_file_non_atomic('a', StringIO('some text for a\n'))
424
 
        self.failUnless(t.has('a'))
 
435
        self.assertTrue(t.has('a'))
425
436
        self.check_transport_contents('some text for a\n', t, 'a')
426
437
        # Put also replaces contents
427
438
        t.put_file_non_atomic('a', StringIO('new\ncontents for\na\n'))
439
450
        # Now test the create_parent flag
440
451
        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'dir/a',
441
452
                                       StringIO('contents\n'))
442
 
        self.failIf(t.has('dir/a'))
 
453
        self.assertFalse(t.has('dir/a'))
443
454
        t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),
444
455
                              create_parent_dir=True)
445
456
        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')
644
655
            self.build_tree(files, transport=transport_from)
645
656
            self.assertEqual(4, transport_from.copy_to(files, transport_to))
646
657
            for f in files:
647
 
                self.check_transport_contents(transport_to.get(f).read(),
 
658
                self.check_transport_contents(transport_to.get_bytes(f),
648
659
                                              transport_from, f)
649
660
 
650
661
        t = self.get_transport()
673
684
        files = ['a', 'b', 'c', 'd']
674
685
        t.copy_to(iter(files), temp_transport)
675
686
        for f in files:
676
 
            self.check_transport_contents(temp_transport.get(f).read(),
 
687
            self.check_transport_contents(temp_transport.get_bytes(f),
677
688
                                          t, f)
678
689
        del temp_transport
679
690
 
822
833
            return
823
834
 
824
835
        t.put_bytes('a', 'a little bit of text\n')
825
 
        self.failUnless(t.has('a'))
 
836
        self.assertTrue(t.has('a'))
826
837
        t.delete('a')
827
 
        self.failIf(t.has('a'))
 
838
        self.assertFalse(t.has('a'))
828
839
 
829
840
        self.assertRaises(NoSuchFile, t.delete, 'a')
830
841
 
836
847
        t.delete_multi(['a', 'c'])
837
848
        self.assertEqual([False, True, False],
838
849
                list(t.has_multi(['a', 'b', 'c'])))
839
 
        self.failIf(t.has('a'))
840
 
        self.failUnless(t.has('b'))
841
 
        self.failIf(t.has('c'))
 
850
        self.assertFalse(t.has('a'))
 
851
        self.assertTrue(t.has('b'))
 
852
        self.assertFalse(t.has('c'))
842
853
 
843
854
        self.assertRaises(NoSuchFile,
844
855
                t.delete_multi, ['a', 'b', 'c'])
905
916
        t.mkdir('foo-baz')
906
917
        t.rmdir('foo')
907
918
        self.assertRaises((NoSuchFile, PathError), t.rmdir, 'foo')
908
 
        self.failUnless(t.has('foo-bar'))
 
919
        self.assertTrue(t.has('foo-bar'))
909
920
 
910
921
    def test_rename_dir_succeeds(self):
911
922
        t = self.get_transport()
994
1005
        self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))
995
1006
 
996
1007
        t.move('a', 'b')
997
 
        self.failUnless(t.has('b'))
998
 
        self.failIf(t.has('a'))
 
1008
        self.assertTrue(t.has('b'))
 
1009
        self.assertFalse(t.has('a'))
999
1010
 
1000
1011
        self.check_transport_contents('a first file\n', t, 'b')
1001
1012
        self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))
1003
1014
        # Overwrite a file
1004
1015
        t.put_bytes('c', 'c this file\n')
1005
1016
        t.move('c', 'b')
1006
 
        self.failIf(t.has('c'))
 
1017
        self.assertFalse(t.has('c'))
1007
1018
        self.check_transport_contents('c this file\n', t, 'b')
1008
1019
 
1009
1020
        # TODO: Try to write a test for atomicity
1041
1052
        except NotImplementedError:
1042
1053
            raise TestSkipped("Transport %s has no bogus URL support." %
1043
1054
                              self._server.__class__)
1044
 
        t = get_transport(url)
 
1055
        t = _mod_transport.get_transport_from_url(url)
1045
1056
        self.assertRaises((ConnectionError, NoSuchFile), t.get, '.bzr/branch')
1046
1057
 
1047
1058
    def test_stat(self):
1063
1074
        for path, size in zip(paths, sizes):
1064
1075
            st = t.stat(path)
1065
1076
            if path.endswith('/'):
1066
 
                self.failUnless(S_ISDIR(st.st_mode))
 
1077
                self.assertTrue(S_ISDIR(st.st_mode))
1067
1078
                # directory sizes are meaningless
1068
1079
            else:
1069
 
                self.failUnless(S_ISREG(st.st_mode))
 
1080
                self.assertTrue(S_ISREG(st.st_mode))
1070
1081
                self.assertEqual(size, st.st_size)
1071
1082
 
1072
1083
        remote_stats = list(t.stat_multi(paths))
1079
1090
        self.assertListRaises(NoSuchFile, t.stat_multi, iter(['a', 'c', 'd']))
1080
1091
        self.build_tree(['subdir/', 'subdir/file'], transport=t)
1081
1092
        subdir = t.clone('subdir')
1082
 
        subdir.stat('./file')
1083
 
        subdir.stat('.')
 
1093
        st = subdir.stat('./file')
 
1094
        st = subdir.stat('.')
1084
1095
 
1085
1096
    def test_hardlink(self):
1086
1097
        from stat import ST_NLINK
1095
1106
        try:
1096
1107
            t.hardlink(source_name, link_name)
1097
1108
 
1098
 
            self.failUnless(t.has(source_name))
1099
 
            self.failUnless(t.has(link_name))
 
1109
            self.assertTrue(t.has(source_name))
 
1110
            self.assertTrue(t.has(link_name))
1100
1111
 
1101
1112
            st = t.stat(link_name)
1102
 
            self.failUnlessEqual(st[ST_NLINK], 2)
 
1113
            self.assertEqual(st[ST_NLINK], 2)
1103
1114
        except TransportNotPossible:
1104
1115
            raise TestSkipped("Transport %s does not support hardlinks." %
1105
1116
                              self._server.__class__)
1117
1128
        try:
1118
1129
            t.symlink(source_name, link_name)
1119
1130
 
1120
 
            self.failUnless(t.has(source_name))
1121
 
            self.failUnless(t.has(link_name))
 
1131
            self.assertTrue(t.has(source_name))
 
1132
            self.assertTrue(t.has(link_name))
1122
1133
 
1123
1134
            st = t.stat(link_name)
1124
 
            self.failUnless(S_ISLNK(st.st_mode),
 
1135
            self.assertTrue(S_ISLNK(st.st_mode),
1125
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
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
 
1766
1778
        self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),
1767
1779
                              transport.readv, 'a', [(12,2)])
1768
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
 
1769
1808
    def test_stat_symlink(self):
1770
1809
        # if a transport points directly to a symlink (and supports symlinks
1771
1810
        # at all) you can tell this.  helps with bug 32669.
1777
1816
        t2 = t.clone('link')
1778
1817
        st = t2.stat('')
1779
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)