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

  • Committer: Martin von Gagern
  • Date: 2010-04-20 08:47:38 UTC
  • mfrom: (5167 +trunk)
  • mto: This revision was merged to the branch mainline in revision 5195.
  • Revision ID: martin.vgagern@gmx.net-20100420084738-ygymnqmdllzrhpfn
merge trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_ftp_transport.py
1
 
# Copyright (C) 2006 Canonical Ltd
 
1
# Copyright (C) 2006, 2010 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
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
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
17
from cStringIO import StringIO
 
18
import ftplib
 
19
import getpass
18
20
import sys
19
21
 
20
22
from bzrlib import (
21
23
    config,
 
24
    errors,
22
25
    tests,
23
26
    transport,
24
27
    ui,
25
28
    )
26
29
 
 
30
from bzrlib.transport import ftp
 
31
 
 
32
from bzrlib.tests import ftp_server
 
33
 
27
34
 
28
35
class TestCaseWithFTPServer(tests.TestCaseWithTransport):
29
36
 
30
 
    _test_needs_features = [tests.FTPServerFeature]
 
37
    _test_needs_features = [ftp_server.FTPServerFeature]
31
38
 
32
39
    def setUp(self):
33
 
        from bzrlib.tests import ftp_server
34
 
        self.transport_server = ftp_server.FTPServer
 
40
        self.transport_server = ftp_server.FTPTestServer
35
41
        super(TestCaseWithFTPServer, self).setUp()
36
42
 
37
43
 
47
53
        self.assertEqual('aftp://host/path', t.abspath(''))
48
54
 
49
55
 
50
 
class TestFTPServer(TestCaseWithFTPServer):
 
56
class TestFTPTestServer(TestCaseWithFTPServer):
51
57
 
52
58
    def test_basic_exists(self):
53
59
        url = self.get_url()
66
72
        self.assertEqual('test more bytes\n', t.get_bytes('foo'))
67
73
 
68
74
 
69
 
class TestFTPServerUI(TestCaseWithFTPServer):
70
 
 
71
 
    def _add_authorized_user(self, user, password):
72
 
        server = self.get_server()
73
 
        # FIXME: There should be a better way to declare authorized users and
74
 
        # passwords to the server
75
 
        authorizer = server._ftp_server.authorizer
76
 
        authorizer.secured_user = user
77
 
        authorizer.secured_password = password
 
75
class TestFTPTestServerUI(TestCaseWithFTPServer):
 
76
 
 
77
    def setUp(self):
 
78
        super(TestFTPTestServerUI, self).setUp()
 
79
        self.user = 'joe'
 
80
        self.password = 'secret'
 
81
        self.get_server().add_user(self.user, self.password)
 
82
 
 
83
    def get_url(self, relpath=None):
 
84
        """Overrides get_url to inject our user."""
 
85
        base = super(TestFTPTestServerUI, self).get_url(relpath)
 
86
        (scheme, user, password,
 
87
         host, port, path) = transport.ConnectedTransport._split_url(base)
 
88
        url = transport.ConnectedTransport._unsplit_url(
 
89
            scheme, self.user, self.password, host, port, path)
 
90
        return url
 
91
 
 
92
    def test_no_prompt_for_username(self):
 
93
        """ensure getpass.getuser() is used if there's no username in the 
 
94
        configuration.""",
 
95
        self.get_server().add_user(getpass.getuser(), self.password)
 
96
        t = self.get_transport()
 
97
        ui.ui_factory = ui.CannedInputUIFactory([self.password])
 
98
        # Issue a request to the server to connect
 
99
        t.put_bytes('foo', 'test bytes\n')
 
100
        self.assertEqual('test bytes\n', t.get_bytes('foo'))
 
101
        # Only the password should've been read
 
102
        ui.ui_factory.assert_all_input_consumed()
78
103
 
79
104
    def test_prompt_for_password(self):
80
105
        t = self.get_transport()
81
 
        # Ensure that the test framework set the password
82
 
        self.assertIsNot(t._password, None)
83
 
        # Reset the password (get_url set the password to 'bar' so we
84
 
        # reset it to None in the transport before the connection).
85
 
        password = t._password
86
 
        t._password = None
87
 
        ui.ui_factory = tests.TestUIFactory(stdin=password+'\n',
88
 
                                            stdout=tests.StringIOWrapper())
89
 
        # Ask the server to check the password
90
 
        self._add_authorized_user(t._user, password)
 
106
        ui.ui_factory = ui.CannedInputUIFactory([self.password])
91
107
        # Issue a request to the server to connect
92
108
        t.has('whatever/not/existing')
93
109
        # stdin should be empty (the provided password have been consumed)
94
 
        self.assertEqual('', ui.ui_factory.stdin.readline())
 
110
        ui.ui_factory.assert_all_input_consumed()
95
111
 
96
112
    def test_no_prompt_for_password_when_using_auth_config(self):
97
113
        t = self.get_transport()
98
 
        # Reset the password (get_url set the password to 'bar' so we
99
 
        # reset it to None in the transport before the connection).
100
 
        password = t._password
101
 
        t._password = None
102
 
        ui.ui_factory = tests.TestUIFactory(stdin='precious\n',
103
 
                                            stdout=tests.StringIOWrapper())
104
 
        # Ask the server to check the password
105
 
        self._add_authorized_user(t._user, password)
106
 
 
 
114
        ui.ui_factory = ui.CannedInputUIFactory([])
107
115
        # Create a config file with the right password
108
116
        conf = config.AuthenticationConfig()
109
117
        conf._get_config().update({'ftptest': {'scheme': 'ftp',
110
 
                                               'user': t._user,
111
 
                                               'password': password}})
 
118
                                               'user': self.user,
 
119
                                               'password': self.password}})
112
120
        conf._save()
113
121
        # Issue a request to the server to connect
114
122
        t.put_bytes('foo', 'test bytes\n')
115
123
        self.assertEqual('test bytes\n', t.get_bytes('foo'))
116
 
        # stdin should have  been left untouched
117
 
        self.assertEqual('precious\n', ui.ui_factory.stdin.readline())
 
124
 
 
125
    def test_empty_password(self):
 
126
        # Override the default user/password from setUp
 
127
        self.user = 'jim'
 
128
        self.password = ''
 
129
        self.get_server().add_user(self.user, self.password)
 
130
        t = self.get_transport()
 
131
        ui.ui_factory = ui.CannedInputUIFactory([self.password])
 
132
        # Issue a request to the server to connect
 
133
        t.has('whatever/not/existing')
 
134
        # stdin should be empty (the provided password have been consumed),
 
135
        # even if the password is empty, it's followed by a newline.
 
136
        ui.ui_factory.assert_all_input_consumed()
 
137
 
 
138
 
 
139
class TestFTPErrorTranslation(tests.TestCase):
 
140
 
 
141
    def test_translate_directory_not_empty(self):
 
142
        # https://bugs.launchpad.net/bugs/528722
 
143
        
 
144
        t = ftp.FtpTransport("ftp://none/")
 
145
 
 
146
        try:
 
147
            raise ftplib.error_temp("Rename/move failure: Directory not empty")
 
148
        except Exception, e:
 
149
            e = self.assertRaises(errors.DirectoryNotEmpty,
 
150
                t._translate_ftp_error, e, "/path")