/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: Canonical.com Patch Queue Manager
  • Date: 2011-01-14 17:18:23 UTC
  • mfrom: (5536.2.9 fetch-dev-docs)
  • Revision ID: pqm@pqm.ubuntu.com-20110114171823-5gx64sero62ag6r4
(jelmer) Add a developer doc on the topic of fetch. (Andrew Bennetts)

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
35
    urlutils,
36
36
    )
37
37
from bzrlib.errors import (ConnectionError,
38
 
                           DirectoryNotEmpty,
39
38
                           FileExists,
40
39
                           InvalidURL,
41
 
                           LockError,
42
40
                           NoSuchFile,
43
 
                           NotLocalUrl,
44
41
                           PathError,
45
42
                           TransportNotPossible,
46
43
                           )
47
44
from bzrlib.osutils import getcwd
48
45
from bzrlib.smart import medium
49
46
from bzrlib.tests import (
50
 
    TestCaseInTempDir,
51
47
    TestSkipped,
52
48
    TestNotApplicable,
53
49
    multiply_tests,
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
 
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())
1765
1765
        # also raise a special error
1766
1766
        self.assertListRaises((errors.ShortReadvError, errors.InvalidRange),
1767
1767
                              transport.readv, 'a', [(12,2)])
 
1768
 
 
1769
    def test_stat_symlink(self):
 
1770
        # if a transport points directly to a symlink (and supports symlinks
 
1771
        # at all) you can tell this.  helps with bug 32669.
 
1772
        t = self.get_transport()
 
1773
        try:
 
1774
            t.symlink('target', 'link')
 
1775
        except TransportNotPossible:
 
1776
            raise TestSkipped("symlinks not supported")
 
1777
        t2 = t.clone('link')
 
1778
        st = t2.stat('')
 
1779
        self.assertTrue(stat.S_ISLNK(st.st_mode))