/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/per_transport.py

  • Committer: Vincent Ladeuil
  • Date: 2012-01-05 14:26:58 UTC
  • mto: This revision was merged to the branch mainline in revision 6468.
  • Revision ID: v.ladeuil+lp@free.fr-20120105142658-vek3v6pzlxb751s2
Tests passing for a first rough version of a cached branch config store. The changes here are too invasive and several parallel proposals have been made. 

@only_raises is evil and gave a hard time since any exception during
save_changes() was swallowed.

Possible improvements: 

- add some needs_write_lock decorators to crucial
  methods (_set_config_location ?) but keep locking the branch at higher levels

- decorate branch.unlock to call stack.save if last_lock() it True
  outside of @only_raises scope (evil decorator)

- add @needs_write_lock to stack.set and stack.remove (will probably get
  rid of most testing issues) we probably need a specialized decorator
  that can relay to the store and from there to the branch or whatever is
  needed. This will also helps bzr config to get it right. The
  get_mutable_section trick should not be needed anymore either.

- decorate branch.unlock to call stack.save if last_lock() it True outside
  of @only_raises scope (evil decorator)

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,
60
56
    _get_transport_modules,
61
57
    )
62
58
from bzrlib.transport.memory import MemoryTransport
78
74
    for module in _get_transport_modules():
79
75
        try:
80
76
            permutations = get_transport_test_permutations(
81
 
                reduce(getattr, (module).split('.')[1:], __import__(module)))
 
77
                pyutils.get_named_object(module))
82
78
            for (klass, server_factory) in permutations:
83
79
                scenario = ('%s,%s' % (klass.__name__, server_factory.__name__),
84
80
                    {"transport_class":klass,
102
98
 
103
99
    def setUp(self):
104
100
        super(TransportTests, self).setUp()
105
 
        self._captureVar('BZR_NO_SMART_VFS', None)
 
101
        self.overrideEnv('BZR_NO_SMART_VFS', None)
106
102
 
107
103
    def check_transport_contents(self, content, transport, relpath):
108
 
        """Check that transport.get(relpath).read() == content."""
109
 
        self.assertEqualDiff(content, transport.get(relpath).read())
 
104
        """Check that transport.get_bytes(relpath) == content."""
 
105
        self.assertEqualDiff(content, transport.get_bytes(relpath))
110
106
 
111
107
    def test_ensure_base_missing(self):
112
108
        """.ensure_base() should create the directory if it doesn't exist"""
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('.')
1085
1093
 
1086
1094
    def test_hardlink(self):
1087
1095
        from stat import ST_NLINK
1096
1104
        try:
1097
1105
            t.hardlink(source_name, link_name)
1098
1106
 
1099
 
            self.failUnless(t.has(source_name))
1100
 
            self.failUnless(t.has(link_name))
 
1107
            self.assertTrue(t.has(source_name))
 
1108
            self.assertTrue(t.has(link_name))
1101
1109
 
1102
1110
            st = t.stat(link_name)
1103
 
            self.failUnlessEqual(st[ST_NLINK], 2)
 
1111
            self.assertEqual(st[ST_NLINK], 2)
1104
1112
        except TransportNotPossible:
1105
1113
            raise TestSkipped("Transport %s does not support hardlinks." %
1106
1114
                              self._server.__class__)
1118
1126
        try:
1119
1127
            t.symlink(source_name, link_name)
1120
1128
 
1121
 
            self.failUnless(t.has(source_name))
1122
 
            self.failUnless(t.has(link_name))
 
1129
            self.assertTrue(t.has(source_name))
 
1130
            self.assertTrue(t.has(link_name))
1123
1131
 
1124
1132
            st = t.stat(link_name)
1125
 
            self.failUnless(S_ISLNK(st.st_mode))
 
1133
            self.assertTrue(S_ISLNK(st.st_mode),
 
1134
                "expected symlink, got mode %o" % st.st_mode)
1126
1135
        except TransportNotPossible:
1127
1136
            raise TestSkipped("Transport %s does not support symlinks." %
1128
1137
                              self._server.__class__)
1129
1138
        except IOError:
1130
 
            raise tests.KnownFailure("Paramiko fails to create symlinks during tests")
 
1139
            self.knownFailure("Paramiko fails to create symlinks during tests")
1131
1140
 
1132
1141
    def test_list_dir(self):
1133
1142
        # TODO: Test list_dir, just try once, and if it throws, stop testing
1197
1206
            raise TestSkipped("not a connected transport")
1198
1207
 
1199
1208
        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)
 
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)
1205
1214
 
1206
1215
    def test__reuse_for(self):
1207
1216
        t = self.get_transport()
1214
1223
 
1215
1224
            Only the parameters different from None will be changed.
1216
1225
            """
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)
 
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))
1225
1234
 
1226
 
        if t._scheme == 'ftp':
 
1235
        if t._parsed_url.scheme == 'ftp':
1227
1236
            scheme = 'sftp'
1228
1237
        else:
1229
1238
            scheme = 'ftp'
1230
1239
        self.assertIsNot(t, t._reuse_for(new_url(scheme=scheme)))
1231
 
        if t._user == 'me':
 
1240
        if t._parsed_url.user == 'me':
1232
1241
            user = 'you'
1233
1242
        else:
1234
1243
            user = 'me'
1245
1254
        #   (they may be typed by the user when prompted for example)
1246
1255
        self.assertIs(t, t._reuse_for(new_url(password='from space')))
1247
1256
        # 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:
 
1257
        self.assertIsNot(t, t._reuse_for(new_url(host=t._parsed_url.host + 'bar')))
 
1258
        if t._parsed_url.port == 1234:
1250
1259
            port = 4321
1251
1260
        else:
1252
1261
            port = 1234
1265
1274
        self.assertIs(t._get_connection(), c._get_connection())
1266
1275
 
1267
1276
        # Temporary failure, we need to create a new dummy connection
1268
 
        new_connection = object()
 
1277
        new_connection = None
1269
1278
        t._set_connection(new_connection)
1270
1279
        # Check that both transports use the same connection
1271
1280
        self.assertIs(new_connection, t._get_connection())
1293
1302
 
1294
1303
        self.build_tree(['a', 'b/', 'b/c'], transport=t1)
1295
1304
 
1296
 
        self.failUnless(t1.has('a'))
1297
 
        self.failUnless(t1.has('b/c'))
1298
 
        self.failIf(t1.has('c'))
 
1305
        self.assertTrue(t1.has('a'))
 
1306
        self.assertTrue(t1.has('b/c'))
 
1307
        self.assertFalse(t1.has('c'))
1299
1308
 
1300
1309
        t2 = t1.clone('b')
1301
1310
        self.assertEqual(t1.base + 'b/', t2.base)
1302
1311
 
1303
 
        self.failUnless(t2.has('c'))
1304
 
        self.failIf(t2.has('a'))
 
1312
        self.assertTrue(t2.has('c'))
 
1313
        self.assertFalse(t2.has('a'))
1305
1314
 
1306
1315
        t3 = t2.clone('..')
1307
 
        self.failUnless(t3.has('a'))
1308
 
        self.failIf(t3.has('c'))
 
1316
        self.assertTrue(t3.has('a'))
 
1317
        self.assertFalse(t3.has('c'))
1309
1318
 
1310
 
        self.failIf(t1.has('b/d'))
1311
 
        self.failIf(t2.has('d'))
1312
 
        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'))
1313
1322
 
1314
1323
        if t1.is_readonly():
1315
1324
            self.build_tree_contents([('b/d', 'newfile\n')])
1316
1325
        else:
1317
1326
            t2.put_bytes('d', 'newfile\n')
1318
1327
 
1319
 
        self.failUnless(t1.has('b/d'))
1320
 
        self.failUnless(t2.has('d'))
1321
 
        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'))
1322
1331
 
1323
1332
    def test_clone_to_root(self):
1324
1333
        orig_transport = self.get_transport()
1398
1407
        self.assertEqual(transport.clone("/").abspath('foo'),
1399
1408
                         transport.abspath("/foo"))
1400
1409
 
 
1410
    # GZ 2011-01-26: Test in per_transport but not using self.get_transport?
1401
1411
    def test_win32_abspath(self):
1402
1412
        # Note: we tried to set sys.platform='win32' so we could test on
1403
1413
        # other platforms too, but then osutils does platform specific
1408
1418
 
1409
1419
        # smoke test for abspath on win32.
1410
1420
        # a transport based on 'file:///' never fully qualifies the drive.
1411
 
        transport = get_transport("file:///")
1412
 
        self.failUnlessEqual(transport.abspath("/"), "file:///")
 
1421
        transport = _mod_transport.get_transport_from_url("file:///")
 
1422
        self.assertEqual(transport.abspath("/"), "file:///")
1413
1423
 
1414
1424
        # 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:/")
 
1425
        transport = _mod_transport.get_transport_from_url("file:///C:/")
 
1426
        self.assertEqual(transport.abspath("/"), "file:///C:/")
1417
1427
 
1418
1428
    def test_local_abspath(self):
1419
1429
        transport = self.get_transport()
1545
1555
 
1546
1556
        no_unicode_support = getattr(self._server, 'no_unicode_support', False)
1547
1557
        if no_unicode_support:
1548
 
            raise tests.KnownFailure("test server cannot handle unicode paths")
 
1558
            self.knownFailure("test server cannot handle unicode paths")
1549
1559
 
1550
1560
        try:
1551
1561
            self.build_tree(files, transport=t, line_endings='binary')
1616
1626
    def test_readv(self):
1617
1627
        transport = self.get_transport()
1618
1628
        if transport.is_readonly():
1619
 
            file('a', 'w').write('0123456789')
 
1629
            with file('a', 'w') as f: f.write('0123456789')
1620
1630
        else:
1621
1631
            transport.put_bytes('a', '0123456789')
1622
1632
 
1632
1642
    def test_readv_out_of_order(self):
1633
1643
        transport = self.get_transport()
1634
1644
        if transport.is_readonly():
1635
 
            file('a', 'w').write('0123456789')
 
1645
            with file('a', 'w') as f: f.write('0123456789')
1636
1646
        else:
1637
1647
            transport.put_bytes('a', '01234567890')
1638
1648
 
1710
1720
        transport = self.get_transport()
1711
1721
        # test from observed failure case.
1712
1722
        if transport.is_readonly():
1713
 
            file('a', 'w').write('a'*1024*1024)
 
1723
            with file('a', 'w') as f: f.write('a'*1024*1024)
1714
1724
        else:
1715
1725
            transport.put_bytes('a', 'a'*1024*1024)
1716
1726
        broken_vector = [(465219, 800), (225221, 800), (445548, 800),
1750
1760
    def test_readv_short_read(self):
1751
1761
        transport = self.get_transport()
1752
1762
        if transport.is_readonly():
1753
 
            file('a', 'w').write('0123456789')
 
1763
            with file('a', 'w') as f: f.write('0123456789')
1754
1764
        else:
1755
1765
            transport.put_bytes('a', '01234567890')
1756
1766
 
1765
1775
        # also raise a special error
1766
1776
        self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),
1767
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_set_segment_parameters(self):
 
1795
        """Segment parameters can be set and show up in base."""
 
1796
        transport = self.get_transport("foo")
 
1797
        orig_base = transport.base
 
1798
        transport.set_segment_parameter("arm", "board")
 
1799
        self.assertEquals("%s,arm=board" % orig_base, transport.base)
 
1800
        self.assertEquals({"arm": "board"}, transport.get_segment_parameters())
 
1801
        transport.set_segment_parameter("arm", None)
 
1802
        transport.set_segment_parameter("nonexistant", None)
 
1803
        self.assertEquals({}, transport.get_segment_parameters())
 
1804
        self.assertEquals(orig_base, transport.base)
 
1805
 
 
1806
    def test_stat_symlink(self):
 
1807
        # if a transport points directly to a symlink (and supports symlinks
 
1808
        # at all) you can tell this.  helps with bug 32669.
 
1809
        t = self.get_transport()
 
1810
        try:
 
1811
            t.symlink('target', 'link')
 
1812
        except TransportNotPossible:
 
1813
            raise TestSkipped("symlinks not supported")
 
1814
        t2 = t.clone('link')
 
1815
        st = t2.stat('')
 
1816
        self.assertTrue(stat.S_ISLNK(st.st_mode))
 
1817
 
 
1818
    def test_abspath_url_unquote_unreserved(self):
 
1819
        """URLs from abspath should have unreserved characters unquoted
 
1820
        
 
1821
        Need consistent quoting notably for tildes, see lp:842223 for more.
 
1822
        """
 
1823
        t = self.get_transport()
 
1824
        needlessly_escaped_dir = "%2D%2E%30%39%41%5A%5F%61%7A%7E/"
 
1825
        self.assertEqual(t.base + "-.09AZ_az~",
 
1826
            t.abspath(needlessly_escaped_dir))
 
1827
 
 
1828
    def test_clone_url_unquote_unreserved(self):
 
1829
        """Base URL of a cloned branch needs unreserved characters unquoted
 
1830
        
 
1831
        Cloned transports should be prefix comparable for things like the
 
1832
        isolation checking of tests, see lp:842223 for more.
 
1833
        """
 
1834
        t1 = self.get_transport()
 
1835
        needlessly_escaped_dir = "%2D%2E%30%39%41%5A%5F%61%7A%7E/"
 
1836
        self.build_tree([needlessly_escaped_dir], transport=t1)
 
1837
        t2 = t1.clone(needlessly_escaped_dir)
 
1838
        self.assertEqual(t1.base + "-.09AZ_az~/", t2.base)