← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/per_transport.py

merge trunk

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"""
260
256
        handle = t.open_write_stream('foo')
261
257
        try:
262
258
            handle.write('b')
263
 
            self.assertEqual('b', t.get('foo').read())
 
259
            self.assertEqual('b', t.get_bytes('foo'))
264
260
        finally:
265
261
            handle.close()
266
262
 
272
268
        try:
273
269
            handle.write('b')
274
270
            self.assertEqual('b', t.get_bytes('foo'))
275
 
            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()
276
276
        finally:
277
277
            handle.close()
278
278
 
644
644
            self.build_tree(files, transport=transport_from)
645
645
            self.assertEqual(4, transport_from.copy_to(files, transport_to))
646
646
            for f in files:
647
 
                self.check_transport_contents(transport_to.get(f).read(),
 
647
                self.check_transport_contents(transport_to.get_bytes(f),
648
648
                                              transport_from, f)
649
649
 
650
650
        t = self.get_transport()
673
673
        files = ['a', 'b', 'c', 'd']
674
674
        t.copy_to(iter(files), temp_transport)
675
675
        for f in files:
676
 
            self.check_transport_contents(temp_transport.get(f).read(),
 
676
            self.check_transport_contents(temp_transport.get_bytes(f),
677
677
                                          t, f)
678
678
        del temp_transport
679
679
 
1041
1041
        except NotImplementedError:
1042
1042
            raise TestSkipped("Transport %s has no bogus URL support." %
1043
1043
                              self._server.__class__)
1044
 
        t = get_transport(url)
 
1044
        t = _mod_transport.get_transport(url)
1045
1045
        self.assertRaises((ConnectionError, NoSuchFile), t.get, '.bzr/branch')
1046
1046
 
1047
1047
    def test_stat(self):
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):