← 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 Pool
  • Date: 2010-06-02 05:03:31 UTC
  • mto: This revision was merged to the branch mainline in revision 5279.
  • Revision ID: mbp@canonical.com-20100602050331-n2p1qt8hfsahspnv
Correct more sloppy use of the term 'Linux'

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,
 
59
    get_transport,
56
60
    _get_transport_modules,
57
61
    )
58
62
from bzrlib.transport.memory import MemoryTransport
74
78
    for module in _get_transport_modules():
75
79
        try:
76
80
            permutations = get_transport_test_permutations(
77
 
                pyutils.get_named_object(module))
 
81
                reduce(getattr, (module).split('.')[1:], __import__(module)))
78
82
            for (klass, server_factory) in permutations:
79
83
                scenario = ('%s,%s' % (klass.__name__, server_factory.__name__),
80
84
                    {"transport_class":klass,
98
102
 
99
103
    def setUp(self):
100
104
        super(TransportTests, self).setUp()
101
 
        self.overrideEnv('BZR_NO_SMART_VFS', None)
 
105
        self._captureVar('BZR_NO_SMART_VFS', None)
102
106
 
103
107
    def check_transport_contents(self, content, transport, relpath):
104
 
        """Check that transport.get_bytes(relpath) == content."""
105
 
        self.assertEqualDiff(content, transport.get_bytes(relpath))
 
108
        """Check that transport.get(relpath).read() == content."""
 
109
        self.assertEqualDiff(content, transport.get(relpath).read())
106
110
 
107
111
    def test_ensure_base_missing(self):
108
112
        """.ensure_base() should create the directory if it doesn't exist"""
247
251
 
248
252
    def test_get_bytes_unknown_file(self):
249
253
        t = self.get_transport()
 
254
 
250
255
        self.assertRaises(NoSuchFile, t.get_bytes, 'c')
251
256
 
252
257
    def test_get_with_open_write_stream_sees_all_content(self):
256
261
        handle = t.open_write_stream('foo')
257
262
        try:
258
263
            handle.write('b')
259
 
            self.assertEqual('b', t.get_bytes('foo'))
 
264
            self.assertEqual('b', t.get('foo').read())
260
265
        finally:
261
266
            handle.close()
262
267
 
268
273
        try:
269
274
            handle.write('b')
270
275
            self.assertEqual('b', t.get_bytes('foo'))
271
 
            f = t.get('foo')
272
 
            try:
273
 
                self.assertEqual('b', f.read())
274
 
            finally:
275
 
                f.close()
 
276
            self.assertEqual('b', t.get('foo').read())
276
277
        finally:
277
278
            handle.close()
278
279
 
644
645
            self.build_tree(files, transport=transport_from)
645
646
            self.assertEqual(4, transport_from.copy_to(files, transport_to))
646
647
            for f in files:
647
 
                self.check_transport_contents(transport_to.get_bytes(f),
 
648
                self.check_transport_contents(transport_to.get(f).read(),
648
649
                                              transport_from, f)
649
650
 
650
651
        t = self.get_transport()
673
674
        files = ['a', 'b', 'c', 'd']
674
675
        t.copy_to(iter(files), temp_transport)
675
676
        for f in files:
676
 
            self.check_transport_contents(temp_transport.get_bytes(f),
 
677
            self.check_transport_contents(temp_transport.get(f).read(),
677
678
                                          t, f)
678
679
        del temp_transport
679
680
 
1041
1042
        except NotImplementedError:
1042
1043
            raise TestSkipped("Transport %s has no bogus URL support." %
1043
1044
                              self._server.__class__)
1044
 
        t = _mod_transport.get_transport(url)
 
1045
        t = get_transport(url)
1045
1046
        self.assertRaises((ConnectionError, NoSuchFile), t.get, '.bzr/branch')
1046
1047
 
1047
1048
    def test_stat(self):
1121
1122
            self.failUnless(t.has(link_name))
1122
1123
 
1123
1124
            st = t.stat(link_name)
1124
 
            self.failUnless(S_ISLNK(st.st_mode),
1125
 
                "expected symlink, got mode %o" % st.st_mode)
 
1125
            self.failUnless(S_ISLNK(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 = None
 
1268
        new_connection = object()
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?
1402
1401
    def test_win32_abspath(self):
1403
1402
        # Note: we tried to set sys.platform='win32' so we could test on
1404
1403
        # other platforms too, but then osutils does platform specific
1409
1408
 
1410
1409
        # smoke test for abspath on win32.
1411
1410
        # a transport based on 'file:///' never fully qualifies the drive.
1412
 
        transport = _mod_transport.get_transport("file:///")
 
1411
        transport = get_transport("file:///")
1413
1412
        self.failUnlessEqual(transport.abspath("/"), "file:///")
1414
1413
 
1415
1414
        # but a transport that starts with a drive spec must keep it.
1416
 
        transport = _mod_transport.get_transport("file:///C:/")
 
1415
        transport = get_transport("file:///C:/")
1417
1416
        self.failUnlessEqual(transport.abspath("/"), "file:///C:/")
1418
1417
 
1419
1418
    def test_local_abspath(self):
1766
1765
        # also raise a special error
1767
1766
        self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),
1768
1767
                              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))