← Back to branch summary

~mbp/bzr/whoami

« back to all changes in this revision

Viewing changes to bzrlib/tests/per_transport.py

  • Committer: Martin Pool
  • Date: 2011-04-01 03:07:34 UTC
  • mfrom: (5050.303.295 2.3)
  • Revision ID: mbp@canonical.com-20110401030734-wip8a66uf8aphgud
merge up to bzr 2.3

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"""
251
247
 
252
248
    def test_get_bytes_unknown_file(self):
253
249
        t = self.get_transport()
254
 
 
255
250
        self.assertRaises(NoSuchFile, t.get_bytes, 'c')
256
251
 
257
252
    def test_get_with_open_write_stream_sees_all_content(self):
261
256
        handle = t.open_write_stream('foo')
262
257
        try:
263
258
            handle.write('b')
264
 
            self.assertEqual('b', t.get('foo').read())
 
259
            self.assertEqual('b', t.get_bytes('foo'))
265
260
        finally:
266
261
            handle.close()
267
262
 
273
268
        try:
274
269
            handle.write('b')
275
270
            self.assertEqual('b', t.get_bytes('foo'))
276
 
            self.assertEqual('b', t.get('foo').read())
 
271
            f = t.get('foo')
 
272
            try:
 
273
                self.assertEqual('b', f.read())
 
274
            finally:
 
275
                f.close()
277
276
        finally:
278
277
            handle.close()
279
278
 
645
644
            self.build_tree(files, transport=transport_from)
646
645
            self.assertEqual(4, transport_from.copy_to(files, transport_to))
647
646
            for f in files:
648
 
                self.check_transport_contents(transport_to.get(f).read(),
 
647
                self.check_transport_contents(transport_to.get_bytes(f),
649
648
                                              transport_from, f)
650
649
 
651
650
        t = self.get_transport()
674
673
        files = ['a', 'b', 'c', 'd']
675
674
        t.copy_to(iter(files), temp_transport)
676
675
        for f in files:
677
 
            self.check_transport_contents(temp_transport.get(f).read(),
 
676
            self.check_transport_contents(temp_transport.get_bytes(f),
678
677
                                          t, f)
679
678
        del temp_transport
680
679
 
1042
1041
        except NotImplementedError:
1043
1042
            raise TestSkipped("Transport %s has no bogus URL support." %
1044
1043
                              self._server.__class__)
1045
 
        t = get_transport(url)
 
1044
        t = _mod_transport.get_transport(url)
1046
1045
        self.assertRaises((ConnectionError, NoSuchFile), t.get, '.bzr/branch')
1047
1046
 
1048
1047
    def test_stat(self):
1122
1121
            self.failUnless(t.has(link_name))
1123
1122
 
1124
1123
            st = t.stat(link_name)
1125
 
            self.failUnless(S_ISLNK(st.st_mode))
 
1124
            self.failUnless(S_ISLNK(st.st_mode),
 
1125
                "expected symlink, got mode %o" % st.st_mode)
1126
1126
        except TransportNotPossible:
1127
1127
            raise TestSkipped("Transport %s does not support symlinks." %
1128
1128
                              self._server.__class__)
1265
1265
        self.assertIs(t._get_connection(), c._get_connection())
1266
1266
 
1267
1267
        # Temporary failure, we need to create a new dummy connection
1268
 
        new_connection = object()
 
1268
        new_connection = None
1269
1269
        t._set_connection(new_connection)
1270
1270
        # Check that both transports use the same connection
1271
1271
        self.assertIs(new_connection, t._get_connection())
1398
1398
        self.assertEqual(transport.clone("/").abspath('foo'),
1399
1399
                         transport.abspath("/foo"))
1400
1400
 
 
1401
    # GZ 2011-01-26: Test in per_transport but not using self.get_transport?
1401
1402
    def test_win32_abspath(self):
1402
1403
        # Note: we tried to set sys.platform='win32' so we could test on
1403
1404
        # other platforms too, but then osutils does platform specific
1408
1409
 
1409
1410
        # smoke test for abspath on win32.
1410
1411
        # a transport based on 'file:///' never fully qualifies the drive.
1411
 
        transport = get_transport("file:///")
 
1412
        transport = _mod_transport.get_transport("file:///")
1412
1413
        self.failUnlessEqual(transport.abspath("/"), "file:///")
1413
1414
 
1414
1415
        # but a transport that starts with a drive spec must keep it.
1415
 
        transport = get_transport("file:///C:/")
 
1416
        transport = _mod_transport.get_transport("file:///C:/")
1416
1417
        self.failUnlessEqual(transport.abspath("/"), "file:///C:/")
1417
1418
 
1418
1419
    def test_local_abspath(self):
1765
1766
        # also raise a special error
1766
1767
        self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),
1767
1768
                              transport.readv, 'a', [(12,2)])
 
1769
 
 
1770
    def test_stat_symlink(self):
 
1771
        # if a transport points directly to a symlink (and supports symlinks
 
1772
        # at all) you can tell this.  helps with bug 32669.
 
1773
        t = self.get_transport()
 
1774
        try:
 
1775
            t.symlink('target', 'link')
 
1776
        except TransportNotPossible:
 
1777
            raise TestSkipped("symlinks not supported")
 
1778
        t2 = t.clone('link')
 
1779
        st = t2.stat('')
 
1780
        self.assertTrue(stat.S_ISLNK(st.st_mode))