← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_ftp_transport.py

merge bzr.dev r4171

Show diffs side-by-side

added added

removed removed

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