← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/stub_sftp.py

  • Committer: Martin Pool
  • Date: 2005-09-06 02:26:28 UTC
  • Revision ID: mbp@sourcefrog.net-20050906022628-66d65f0feb4a9e80
- implement version 5 xml storage, and tests

  This stores files identified by the version that introduced the 
  text, and the version that introduced the name.  Entry kinds are
  given by the xml tag not an explicit kind field.

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