/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: Michael Ellerman
  • Date: 2006-03-09 00:24:48 UTC
  • mto: (1610.1.8 bzr.mbp.integration)
  • mto: This revision was merged to the branch mainline in revision 1616.
  • Revision ID: michael@ellerman.id.au-20060309002448-70cce15e3d605130
Make the "ignore line" in the commit message editor the "right" width, so
that if you make your message that wide it won't wrap in bzr log output.
Just as a visual aid.

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
from bzrlib.osutils import pathjoin
 
26
from bzrlib.trace import mutter
 
27
 
 
28
 
 
29
class StubServer (ServerInterface):
 
30
    def __init__(self, test_case):
 
31
        ServerInterface.__init__(self)
 
32
        self._test_case = test_case
 
33
 
 
34
    def check_auth_password(self, username, password):
 
35
        # all are allowed
 
36
        self._test_case.log('sftpserver - authorizing: %s' % (username,))
 
37
        return AUTH_SUCCESSFUL
 
38
 
 
39
    def check_channel_request(self, kind, chanid):
 
40
        self._test_case.log('sftpserver - channel request: %s, %s' % (kind, chanid))
 
41
        return OPEN_SUCCEEDED
 
42
 
 
43
 
 
44
class StubSFTPHandle (SFTPHandle):
 
45
    def stat(self):
 
46
        try:
 
47
            return SFTPAttributes.from_stat(os.fstat(self.readfile.fileno()))
 
48
        except OSError, e:
 
49
            return SFTPServer.convert_errno(e.errno)
 
50
 
 
51
    def chattr(self, attr):
 
52
        # python doesn't have equivalents to fchown or fchmod, so we have to
 
53
        # use the stored filename
 
54
        mutter('Changing permissions on %s to %s', self.filename, attr)
 
55
        try:
 
56
            SFTPServer.set_file_attr(self.filename, attr)
 
57
        except OSError, e:
 
58
            return SFTPServer.convert_errno(e.errno)
 
59
 
 
60
 
 
61
class StubSFTPServer (SFTPServerInterface):
 
62
    def __init__(self, server, root, home=None):
 
63
        SFTPServerInterface.__init__(self, server)
 
64
        self.root = root
 
65
        if home is None:
 
66
            self.home = self.root
 
67
        else:
 
68
            self.home = home[len(self.root):]
 
69
        if (len(self.home) > 0) and (self.home[0] == '/'):
 
70
            self.home = self.home[1:]
 
71
        server._test_case.log('sftpserver - new connection')
 
72
 
 
73
    def _realpath(self, path):
 
74
        return self.root + self.canonicalize(path)
 
75
 
 
76
    def canonicalize(self, path):
 
77
        if os.path.isabs(path):
 
78
            return os.path.normpath(path)
 
79
        else:
 
80
            return os.path.normpath('/' + os.path.join(self.home, path))
 
81
 
 
82
    def chattr(self, path, attr):
 
83
        try:
 
84
            SFTPServer.set_file_attr(path, attr)
 
85
        except OSError, e:
 
86
            return SFTPServer.convert_errno(e.errno)
 
87
        return SFTP_OK
 
88
 
 
89
    def list_folder(self, path):
 
90
        path = self._realpath(path)
 
91
        try:
 
92
            out = [ ]
 
93
            flist = os.listdir(path)
 
94
            for fname in flist:
 
95
                attr = SFTPAttributes.from_stat(os.stat(pathjoin(path, fname)))
 
96
                attr.filename = fname
 
97
                out.append(attr)
 
98
            return out
 
99
        except OSError, e:
 
100
            return SFTPServer.convert_errno(e.errno)
 
101
 
 
102
    def stat(self, path):
 
103
        path = self._realpath(path)
 
104
        try:
 
105
            return SFTPAttributes.from_stat(os.stat(path))
 
106
        except OSError, e:
 
107
            return SFTPServer.convert_errno(e.errno)
 
108
 
 
109
    def lstat(self, path):
 
110
        path = self._realpath(path)
 
111
        try:
 
112
            return SFTPAttributes.from_stat(os.lstat(path))
 
113
        except OSError, e:
 
114
            return SFTPServer.convert_errno(e.errno)
 
115
 
 
116
    def open(self, path, flags, attr):
 
117
        path = self._realpath(path)
 
118
        try:
 
119
            if hasattr(os, 'O_BINARY'):
 
120
                flags |= os.O_BINARY
 
121
            if getattr(attr, 'st_mode', None):
 
122
                fd = os.open(path, flags, attr.st_mode)
 
123
            else:
 
124
                fd = os.open(path, flags)
 
125
        except OSError, e:
 
126
            return SFTPServer.convert_errno(e.errno)
 
127
        if (flags & os.O_CREAT) and (attr is not None):
 
128
            attr._flags &= ~attr.FLAG_PERMISSIONS
 
129
            SFTPServer.set_file_attr(path, attr)
 
130
        if flags & os.O_WRONLY:
 
131
            fstr = 'wb'
 
132
        elif flags & os.O_RDWR:
 
133
            fstr = 'rb+'
 
134
        else:
 
135
            # O_RDONLY (== 0)
 
136
            fstr = 'rb'
 
137
        try:
 
138
            f = os.fdopen(fd, fstr)
 
139
        except OSError, e:
 
140
            return SFTPServer.convert_errno(e.errno)
 
141
        fobj = StubSFTPHandle()
 
142
        fobj.filename = path
 
143
        fobj.readfile = f
 
144
        fobj.writefile = f
 
145
        return fobj
 
146
 
 
147
    def remove(self, path):
 
148
        path = self._realpath(path)
 
149
        try:
 
150
            os.remove(path)
 
151
        except OSError, e:
 
152
            return SFTPServer.convert_errno(e.errno)
 
153
        return SFTP_OK
 
154
 
 
155
    def rename(self, oldpath, newpath):
 
156
        oldpath = self._realpath(oldpath)
 
157
        newpath = self._realpath(newpath)
 
158
        try:
 
159
            os.rename(oldpath, newpath)
 
160
        except OSError, e:
 
161
            return SFTPServer.convert_errno(e.errno)
 
162
        return SFTP_OK
 
163
 
 
164
    def mkdir(self, path, attr):
 
165
        path = self._realpath(path)
 
166
        try:
 
167
            # Using getattr() in case st_mode is None or 0
 
168
            # both evaluate to False
 
169
            if getattr(attr, 'st_mode', None):
 
170
                os.mkdir(path, attr.st_mode)
 
171
            else:
 
172
                os.mkdir(path)
 
173
            if attr is not None:
 
174
                attr._flags &= ~attr.FLAG_PERMISSIONS
 
175
                SFTPServer.set_file_attr(path, attr)
 
176
        except OSError, e:
 
177
            return SFTPServer.convert_errno(e.errno)
 
178
        return SFTP_OK
 
179
 
 
180
    def rmdir(self, path):
 
181
        path = self._realpath(path)
 
182
        try:
 
183
            os.rmdir(path)
 
184
        except OSError, e:
 
185
            return SFTPServer.convert_errno(e.errno)
 
186
        return SFTP_OK
 
187
 
 
188
    # removed: chattr, symlink, readlink
 
189
    # (nothing in bzr's sftp transport uses those)