← 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: Martin
  • Date: 2010-05-03 20:57:39 UTC
  • mto: This revision was merged to the branch mainline in revision 5204.
  • Revision ID: gzlist@googlemail.com-20100503205739-n326zdvevv0rmruh
Retain original stack and error message when translating to ValueError in bencode

Show diffs side-by-side

added added

removed removed

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