/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/test_transport_implementations.py

  • Committer: Jan Balster
  • Date: 2006-08-15 12:39:42 UTC
  • mfrom: (1923 +trunk)
  • mto: This revision was merged to the branch mainline in revision 1928.
  • Revision ID: jan@merlinux.de-20060815123942-22c388c6e9a8ac91
merge bzr.dev 1923

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport_implementations.py
1
1
# Copyright (C) 2004, 2005, 2006 by 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
5
5
# the Free Software Foundation; either version 2 of the License, or
6
6
# (at your option) any later version.
7
 
 
 
7
#
8
8
# This program is distributed in the hope that it will be useful,
9
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
11
# GNU General Public License for more details.
12
 
 
 
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
25
25
import stat
26
26
import sys
27
27
 
 
28
from bzrlib import (
 
29
    osutils,
 
30
    urlutils,
 
31
    )
28
32
from bzrlib.errors import (DirectoryNotEmpty, NoSuchFile, FileExists,
29
33
                           LockError, PathError,
30
34
                           TransportNotPossible, ConnectionError,
31
35
                           InvalidURL)
32
36
from bzrlib.osutils import getcwd
33
37
from bzrlib.tests import TestCaseInTempDir, TestSkipped
 
38
from bzrlib.tests.test_transport import TestTransportImplementation
34
39
from bzrlib.transport import memory
35
40
import bzrlib.transport
36
 
import bzrlib.urlutils as urlutils
37
41
 
38
42
 
39
43
def _append(fn, txt):
45
49
        f.close()
46
50
 
47
51
 
48
 
class TestTransportImplementation(TestCaseInTempDir):
49
 
    """Implementation verification for transports.
50
 
    
51
 
    To verify a transport we need a server factory, which is a callable
52
 
    that accepts no parameters and returns an implementation of
53
 
    bzrlib.transport.Server.
54
 
    
55
 
    That Server is then used to construct transport instances and test
56
 
    the transport via loopback activity.
57
 
 
58
 
    Currently this assumes that the Transport object is connected to the 
59
 
    current working directory.  So that whatever is done 
60
 
    through the transport, should show up in the working 
61
 
    directory, and vice-versa. This is a bug, because its possible to have
62
 
    URL schemes which provide access to something that may not be 
63
 
    result in storage on the local disk, i.e. due to file system limits, or 
64
 
    due to it being a database or some other non-filesystem tool.
65
 
 
66
 
    This also tests to make sure that the functions work with both
67
 
    generators and lists (assuming iter(list) is effectively a generator)
68
 
    """
69
 
    
70
 
    def setUp(self):
71
 
        super(TestTransportImplementation, self).setUp()
72
 
        self._server = self.transport_server()
73
 
        self._server.setUp()
74
 
 
75
 
    def tearDown(self):
76
 
        super(TestTransportImplementation, self).tearDown()
77
 
        self._server.tearDown()
78
 
        
 
52
class TransportTests(TestTransportImplementation):
 
53
 
79
54
    def check_transport_contents(self, content, transport, relpath):
80
55
        """Check that transport.get(relpath).read() == content."""
81
56
        self.assertEqualDiff(content, transport.get(relpath).read())
82
57
 
83
 
    def get_transport(self):
84
 
        """Return a connected transport to the local directory."""
85
 
        base_url = self._server.get_url()
86
 
        t = bzrlib.transport.get_transport(base_url)
87
 
        if not isinstance(t, self.transport_class):
88
 
            # we want to make sure to construct one particular class, even if
89
 
            # there are several available implementations of this transport;
90
 
            # therefore construct it by hand rather than through the regular
91
 
            # get_transport method
92
 
            t = self.transport_class(base_url)
93
 
        return t
94
 
 
95
58
    def assertListRaises(self, excClass, func, *args, **kwargs):
96
59
        """Fail unless excClass is raised when the iterator from func is used.
97
60
        
186
149
 
187
150
        if t.is_readonly():
188
151
            return
 
152
        if not t._can_roundtrip_unix_modebits():
 
153
            # Can't roundtrip, so no need to run this test
 
154
            return
189
155
        t.put('mode644', StringIO('test text\n'), mode=0644)
190
156
        self.assertTransportMode(t, 'mode644', 0644)
191
157
        t.put('mode666', StringIO('test text\n'), mode=0666)
197
163
        self.assertTransportMode(t, 'mode400', 0400)
198
164
        t.put_multi([('mmode644', StringIO('text\n'))], mode=0644)
199
165
        self.assertTransportMode(t, 'mmode644', 0644)
 
166
 
 
167
        # The default permissions should be based on the current umask
 
168
        umask = osutils.get_umask()
 
169
        t.put('nomode', StringIO('test text\n'), mode=None)
 
170
        self.assertTransportMode(t, 'nomode', 0666 & ~umask)
200
171
        
201
172
    def test_mkdir(self):
202
173
        t = self.get_transport()
261
232
        self.assertTransportMode(t, 'dmode777', 0777)
262
233
        t.mkdir('dmode700', mode=0700)
263
234
        self.assertTransportMode(t, 'dmode700', 0700)
264
 
        # TODO: jam 20051215 test mkdir_multi with a mode
265
235
        t.mkdir_multi(['mdmode755'], mode=0755)
266
236
        self.assertTransportMode(t, 'mdmode755', 0755)
267
237
 
 
238
        # Default mode should be based on umask
 
239
        umask = osutils.get_umask()
 
240
        t.mkdir('dnomode', mode=None)
 
241
        self.assertTransportMode(t, 'dnomode', 0777 & ~umask)
 
242
 
268
243
    def test_copy_to(self):
269
244
        # FIXME: test:   same server to same server (partly done)
270
245
        # same protocol two servers
706
681
        except (ConnectionError, NoSuchFile), e:
707
682
            pass
708
683
        except (Exception), e:
709
 
            self.fail('Wrong exception thrown (%s): %s' 
710
 
                        % (e.__class__.__name__, e))
 
684
            self.fail('Wrong exception thrown (%s.%s): %s' 
 
685
                        % (e.__class__.__module__, e.__class__.__name__, e))
711
686
        else:
712
687
            self.fail('Did not get the expected ConnectionError or NoSuchFile.')
713
688
 
913
888
        """Test that we can read/write files with Unicode names."""
914
889
        t = self.get_transport()
915
890
 
916
 
        files = [u'\xe5', # a w/ circle iso-8859-1
917
 
                 u'\xe4', # a w/ dots iso-8859-1
 
891
        # With FAT32 and certain encodings on win32
 
892
        # '\xe5' and '\xe4' actually map to the same file
 
893
        # adding a suffix kicks in the 'preserving but insensitive'
 
894
        # route, and maintains the right files
 
895
        files = [u'\xe5.1', # a w/ circle iso-8859-1
 
896
                 u'\xe4.2', # a w/ dots iso-8859-1
918
897
                 u'\u017d', # Z with umlat iso-8859-2
919
898
                 u'\u062c', # Arabic j
920
899
                 u'\u0410', # Russian A
981
960
        if transport.is_readonly():
982
961
            file('a', 'w').write('0123456789')
983
962
        else:
984
 
            transport.put('a', StringIO('01234567890'))
 
963
            transport.put('a', StringIO('0123456789'))
985
964
 
986
965
        d = list(transport.readv('a', ((0, 1), (1, 1), (3, 2), (9, 1))))
987
966
        self.assertEqual(d[0], (0, '0'))
988
967
        self.assertEqual(d[1], (1, '1'))
989
968
        self.assertEqual(d[2], (3, '34'))
990
969
        self.assertEqual(d[3], (9, '9'))
 
970
 
 
971
    def test_readv_out_of_order(self):
 
972
        transport = self.get_transport()
 
973
        if transport.is_readonly():
 
974
            file('a', 'w').write('0123456789')
 
975
        else:
 
976
            transport.put('a', StringIO('01234567890'))
 
977
 
 
978
        d = list(transport.readv('a', ((1, 1), (9, 1), (0, 1), (3, 2))))
 
979
        self.assertEqual(d[0], (1, '1'))
 
980
        self.assertEqual(d[1], (9, '9'))
 
981
        self.assertEqual(d[2], (0, '0'))
 
982
        self.assertEqual(d[3], (3, '34'))