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

  • Committer: John Arbash Meinel
  • Date: 2008-08-18 22:34:21 UTC
  • mto: (3606.5.6 1.6)
  • mto: This revision was merged to the branch mainline in revision 3641.
  • Revision ID: john@arbash-meinel.com-20080818223421-todjny24vj4faj4t
Add tests for the fetching behavior.

The proper parameter passed is 'unordered' add an assert for it, and
fix callers that were passing 'unsorted' instead.
Add tests that we make the right get_record_stream call based
on the value of _fetch_uses_deltas.
Fix the fetch request for signatures.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/stub_sftp.py
 
1
# Copyright (C) 2005 Robey Pointer <robey@lag.net>, Canonical Ltd
 
2
#
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
#
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
#
 
13
# You should have received a copy of the GNU General Public License
 
14
# along with this program; if not, write to the Free Software
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
16
 
 
17
"""
 
18
A stub SFTP server for loopback SFTP testing.
 
19
Adapted from the one in paramiko's unit tests.
 
20
"""
 
21
 
 
22
import os
 
23
from paramiko import ServerInterface, SFTPServerInterface, SFTPServer, SFTPAttributes, \
 
24
    SFTPHandle, SFTP_OK, AUTH_SUCCESSFUL, OPEN_SUCCEEDED
 
25
import sys
 
26
 
 
27
from bzrlib.osutils import pathjoin
 
28
from bzrlib.trace import mutter
 
29
 
 
30
 
 
31
class StubServer (ServerInterface):
 
32
 
 
33
    def __init__(self, test_case):
 
34
        ServerInterface.__init__(self)
 
35
        self._test_case = test_case
 
36
 
 
37
    def check_auth_password(self, username, password):
 
38
        # all are allowed
 
39
        self._test_case.log('sftpserver - authorizing: %s' % (username,))
 
40
        return AUTH_SUCCESSFUL
 
41
 
 
42
    def check_channel_request(self, kind, chanid):
 
43
        self._test_case.log('sftpserver - channel request: %s, %s' % (kind, chanid))
 
44
        return OPEN_SUCCEEDED
 
45
 
 
46
 
 
47
class StubSFTPHandle (SFTPHandle):
 
48
    def stat(self):
 
49
        try:
 
50
            return SFTPAttributes.from_stat(os.fstat(self.readfile.fileno()))
 
51
        except OSError, e:
 
52
            return SFTPServer.convert_errno(e.errno)
 
53
 
 
54
    def chattr(self, attr):
 
55
        # python doesn't have equivalents to fchown or fchmod, so we have to
 
56
        # use the stored filename
 
57
        mutter('Changing permissions on %s to %s', self.filename, attr)
 
58
        try:
 
59
            SFTPServer.set_file_attr(self.filename, attr)
 
60
        except OSError, e:
 
61
            return SFTPServer.convert_errno(e.errno)
 
62
 
 
63
 
 
64
class StubSFTPServer (SFTPServerInterface):
 
65
 
 
66
    def __init__(self, server, root, home=None):
 
67
        SFTPServerInterface.__init__(self, server)
 
68
        # All paths are actually relative to 'root'.
 
69
        # this is like implementing chroot().
 
70
        self.root = root
 
71
        if home is None:
 
72
            self.home = ''
 
73
        else:
 
74
            if not home.startswith(self.root):
 
75
                raise AssertionError(
 
76
                    "home must be a subdirectory of root (%s vs %s)"
 
77
                    % (home, root))
 
78
            self.home = home[len(self.root):]
 
79
        if self.home.startswith('/'):
 
80
            self.home = self.home[1:]
 
81
        server._test_case.log('sftpserver - new connection')
 
82
 
 
83
    def _realpath(self, path):
 
84
        # paths returned from self.canonicalize() always start with
 
85
        # a path separator. So if 'root' is just '/', this would cause
 
86
        # a double slash at the beginning '//home/dir'. 
 
87
        if self.root == '/':
 
88
            return self.canonicalize(path)
 
89
        return self.root + self.canonicalize(path)
 
90
 
 
91
    if sys.platform == 'win32':
 
92
        def canonicalize(self, path):
 
93
            # Win32 sftp paths end up looking like
 
94
            #     sftp://host@foo/h:/foo/bar
 
95
            # which means absolute paths look like:
 
96
            #     /h:/foo/bar
 
97
            # and relative paths stay the same:
 
98
            #     foo/bar
 
99
            # win32 needs to use the Unicode APIs. so we require the 
 
100
            # paths to be utf8 (Linux just uses bytestreams)
 
101
            thispath = path.decode('utf8')
 
102
            if path.startswith('/'):
 
103
                # Abspath H:/foo/bar
 
104
                return os.path.normpath(thispath[1:])
 
105
            else:
 
106
                return os.path.normpath(os.path.join(self.home, thispath))
 
107
    else:
 
108
        def canonicalize(self, path):
 
109
            if os.path.isabs(path):
 
110
                return os.path.normpath(path)
 
111
            else:
 
112
                return os.path.normpath('/' + os.path.join(self.home, path))
 
113
 
 
114
    def chattr(self, path, attr):
 
115
        try:
 
116
            SFTPServer.set_file_attr(path, attr)
 
117
        except OSError, e:
 
118
            return SFTPServer.convert_errno(e.errno)
 
119
        return SFTP_OK
 
120
 
 
121
    def list_folder(self, path):
 
122
        path = self._realpath(path)
 
123
        try:
 
124
            out = [ ]
 
125
            # TODO: win32 incorrectly lists paths with non-ascii if path is not
 
126
            # unicode. However on Linux the server should only deal with
 
127
            # bytestreams and posix.listdir does the right thing 
 
128
            if sys.platform == 'win32':
 
129
                flist = [f.encode('utf8') for f in os.listdir(path)]
 
130
            else:
 
131
                flist = os.listdir(path)
 
132
            for fname in flist:
 
133
                attr = SFTPAttributes.from_stat(os.stat(pathjoin(path, fname)))
 
134
                attr.filename = fname
 
135
                out.append(attr)
 
136
            return out
 
137
        except OSError, e:
 
138
            return SFTPServer.convert_errno(e.errno)
 
139
 
 
140
    def stat(self, path):
 
141
        path = self._realpath(path)
 
142
        try:
 
143
            return SFTPAttributes.from_stat(os.stat(path))
 
144
        except OSError, e:
 
145
            return SFTPServer.convert_errno(e.errno)
 
146
 
 
147
    def lstat(self, path):
 
148
        path = self._realpath(path)
 
149
        try:
 
150
            return SFTPAttributes.from_stat(os.lstat(path))
 
151
        except OSError, e:
 
152
            return SFTPServer.convert_errno(e.errno)
 
153
 
 
154
    def open(self, path, flags, attr):
 
155
        path = self._realpath(path)
 
156
        try:
 
157
            flags |= getattr(os, 'O_BINARY', 0)
 
158
            if getattr(attr, 'st_mode', None):
 
159
                fd = os.open(path, flags, attr.st_mode)
 
160
            else:
 
161
                # os.open() defaults to 0777 which is
 
162
                # an odd default mode for files
 
163
                fd = os.open(path, flags, 0666)
 
164
        except OSError, e:
 
165
            return SFTPServer.convert_errno(e.errno)
 
166
 
 
167
        if (flags & os.O_CREAT) and (attr is not None):
 
168
            attr._flags &= ~attr.FLAG_PERMISSIONS
 
169
            SFTPServer.set_file_attr(path, attr)
 
170
        if flags & os.O_WRONLY:
 
171
            fstr = 'wb'
 
172
        elif flags & os.O_RDWR:
 
173
            fstr = 'rb+'
 
174
        else:
 
175
            # O_RDONLY (== 0)
 
176
            fstr = 'rb'
 
177
        try:
 
178
            f = os.fdopen(fd, fstr)
 
179
        except (IOError, OSError), e:
 
180
            return SFTPServer.convert_errno(e.errno)
 
181
        fobj = StubSFTPHandle()
 
182
        fobj.filename = path
 
183
        fobj.readfile = f
 
184
        fobj.writefile = f
 
185
        return fobj
 
186
 
 
187
    def remove(self, path):
 
188
        path = self._realpath(path)
 
189
        try:
 
190
            os.remove(path)
 
191
        except OSError, e:
 
192
            return SFTPServer.convert_errno(e.errno)
 
193
        return SFTP_OK
 
194
 
 
195
    def rename(self, oldpath, newpath):
 
196
        oldpath = self._realpath(oldpath)
 
197
        newpath = self._realpath(newpath)
 
198
        try:
 
199
            os.rename(oldpath, newpath)
 
200
        except OSError, e:
 
201
            return SFTPServer.convert_errno(e.errno)
 
202
        return SFTP_OK
 
203
 
 
204
    def mkdir(self, path, attr):
 
205
        path = self._realpath(path)
 
206
        try:
 
207
            # Using getattr() in case st_mode is None or 0
 
208
            # both evaluate to False
 
209
            if getattr(attr, 'st_mode', None):
 
210
                os.mkdir(path, attr.st_mode)
 
211
            else:
 
212
                os.mkdir(path)
 
213
            if attr is not None:
 
214
                attr._flags &= ~attr.FLAG_PERMISSIONS
 
215
                SFTPServer.set_file_attr(path, attr)
 
216
        except OSError, e:
 
217
            return SFTPServer.convert_errno(e.errno)
 
218
        return SFTP_OK
 
219
 
 
220
    def rmdir(self, path):
 
221
        path = self._realpath(path)
 
222
        try:
 
223
            os.rmdir(path)
 
224
        except OSError, e:
 
225
            return SFTPServer.convert_errno(e.errno)
 
226
        return SFTP_OK
 
227
 
 
228
    # removed: chattr, symlink, readlink
 
229
    # (nothing in bzr's sftp transport uses those)