/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: Andrew Bennetts
  • Date: 2010-01-12 03:53:21 UTC
  • mfrom: (4948 +trunk)
  • mto: This revision was merged to the branch mainline in revision 4964.
  • Revision ID: andrew.bennetts@canonical.com-20100112035321-hofpz5p10224ryj3
Merge lp:bzr, resolving conflicts.

Show diffs side-by-side

added added

removed removed

Lines of Context:
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 getpass
18
19
import sys
19
20
 
20
21
from bzrlib import (
24
25
    ui,
25
26
    )
26
27
 
 
28
from bzrlib.tests import ftp_server
 
29
 
27
30
 
28
31
class TestCaseWithFTPServer(tests.TestCaseWithTransport):
29
32
 
30
 
    _test_needs_features = [tests.FTPServerFeature]
 
33
    _test_needs_features = [ftp_server.FTPServerFeature]
31
34
 
32
35
    def setUp(self):
33
 
        from bzrlib.tests import ftp_server
34
 
        self.transport_server = ftp_server.FTPServer
 
36
        self.transport_server = ftp_server.FTPTestServer
35
37
        super(TestCaseWithFTPServer, self).setUp()
36
38
 
37
39
 
47
49
        self.assertEqual('aftp://host/path', t.abspath(''))
48
50
 
49
51
 
50
 
class TestFTPServer(TestCaseWithFTPServer):
 
52
class TestFTPTestServer(TestCaseWithFTPServer):
51
53
 
52
54
    def test_basic_exists(self):
53
55
        url = self.get_url()
66
68
        self.assertEqual('test more bytes\n', t.get_bytes('foo'))
67
69
 
68
70
 
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
 
71
class TestFTPTestServerUI(TestCaseWithFTPServer):
 
72
 
 
73
    def setUp(self):
 
74
        super(TestFTPTestServerUI, self).setUp()
 
75
        self.user = 'joe'
 
76
        self.password = 'secret'
 
77
        self.get_server().add_user(self.user, self.password)
 
78
 
 
79
    def get_url(self, relpath=None):
 
80
        """Overrides get_url to inject our user."""
 
81
        base = super(TestFTPTestServerUI, self).get_url(relpath)
 
82
        (scheme, user, password,
 
83
         host, port, path) = transport.ConnectedTransport._split_url(base)
 
84
        url = transport.ConnectedTransport._unsplit_url(
 
85
            scheme, self.user, self.password, host, port, path)
 
86
        return url
 
87
 
 
88
    def test_no_prompt_for_username(self):
 
89
        """ensure getpass.getuser() is used if there's no username in the 
 
90
        configuration.""",
 
91
        self.get_server().add_user(getpass.getuser(), self.password)
 
92
        t = self.get_transport()
 
93
        ui.ui_factory = ui.CannedInputUIFactory([self.password])
 
94
        # Issue a request to the server to connect
 
95
        t.put_bytes('foo', 'test bytes\n')
 
96
        self.assertEqual('test bytes\n', t.get_bytes('foo'))
 
97
        # Only the password should've been read
 
98
        ui.ui_factory.assert_all_input_consumed()
78
99
 
79
100
    def test_prompt_for_password(self):
80
101
        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)
 
102
        ui.ui_factory = ui.CannedInputUIFactory([self.password])
91
103
        # Issue a request to the server to connect
92
104
        t.has('whatever/not/existing')
93
105
        # stdin should be empty (the provided password have been consumed)
94
 
        self.assertEqual('', ui.ui_factory.stdin.readline())
 
106
        ui.ui_factory.assert_all_input_consumed()
95
107
 
96
108
    def test_no_prompt_for_password_when_using_auth_config(self):
97
109
        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
 
 
 
110
        ui.ui_factory = ui.CannedInputUIFactory([])
107
111
        # Create a config file with the right password
108
112
        conf = config.AuthenticationConfig()
109
113
        conf._get_config().update({'ftptest': {'scheme': 'ftp',
110
 
                                               'user': t._user,
111
 
                                               'password': password}})
 
114
                                               'user': self.user,
 
115
                                               'password': self.password}})
112
116
        conf._save()
113
117
        # Issue a request to the server to connect
114
118
        t.put_bytes('foo', 'test bytes\n')
115
119
        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())
 
120
 
 
121
    def test_empty_password(self):
 
122
        # Override the default user/password from setUp
 
123
        self.user = 'jim'
 
124
        self.password = ''
 
125
        self.get_server().add_user(self.user, self.password)
 
126
        t = self.get_transport()
 
127
        ui.ui_factory = ui.CannedInputUIFactory([self.password])
 
128
        # Issue a request to the server to connect
 
129
        t.has('whatever/not/existing')
 
130
        # stdin should be empty (the provided password have been consumed),
 
131
        # even if the password is empty, it's followed by a newline.
 
132
        ui.ui_factory.assert_all_input_consumed()