/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

Merge from bzr.dev

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_ftp_transport.py
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
 
17
from cStringIO import StringIO
17
18
import sys
18
19
 
19
20
from bzrlib import (
 
21
    config,
20
22
    tests,
21
23
    transport,
22
24
    ui,
23
25
    )
24
26
 
25
27
 
26
 
class _MedusaFeature(tests.Feature):
27
 
    """Some tests want an FTP Server, check if one is available.
28
 
 
29
 
    Right now, the only way this is available is if 'medusa' is installed.
30
 
    """
31
 
 
32
 
    def _probe(self):
33
 
        try:
34
 
            import medusa
35
 
            import medusa.filesys
36
 
            import medusa.ftp_server
37
 
            return True
38
 
        except ImportError:
39
 
            return False
40
 
 
41
 
    def feature_name(self):
42
 
        return 'medusa'
43
 
 
44
 
MedusaFeature = _MedusaFeature()
45
 
 
46
 
 
47
28
class TestCaseWithFTPServer(tests.TestCaseWithTransport):
48
29
 
49
 
    _test_needs_features = [MedusaFeature]
 
30
    _test_needs_features = [tests.FTPServerFeature]
50
31
 
51
32
    def setUp(self):
52
 
        from bzrlib.transport.ftp import FtpServer
53
 
        self.transport_server = FtpServer
 
33
        from bzrlib.tests import ftp_server
 
34
        self.transport_server = ftp_server.FTPServer
54
35
        super(TestCaseWithFTPServer, self).setUp()
55
36
 
56
37
 
57
 
 
58
38
class TestCaseAFTP(tests.TestCaseWithTransport):
59
39
    """Test aftp transport."""
60
40
 
77
57
        t.put_bytes('foo', 'test bytes\n')
78
58
        self.assertEqual('test bytes\n', t.get_bytes('foo'))
79
59
 
 
60
    def test__reconnect(self):
 
61
        t = self.get_transport()
 
62
        t.put_bytes('foo', 'test bytes\n')
 
63
        self.assertEqual('test bytes\n', t.get_bytes('foo'))
 
64
        t._reconnect()
 
65
        t.put_bytes('foo', 'test more bytes\n')
 
66
        self.assertEqual('test more bytes\n', t.get_bytes('foo'))
 
67
 
80
68
 
81
69
class TestFTPServerUI(TestCaseWithFTPServer):
82
70
 
83
 
    def setUp(self):
84
 
        super(TestFTPServerUI, self).setUp()
85
 
        self.old_factory = ui.ui_factory
86
 
        # The following has the unfortunate side-effect of hiding any ouput
87
 
        # during the tests (including pdb prompts). Feel free to comment them
88
 
        # for debugging purposes but leave them in place, there are needed to
89
 
        # run the tests without any console
90
 
        self.old_stdout = sys.stdout
91
 
        sys.stdout = tests.StringIOWrapper()
92
 
        self.addCleanup(self.restoreUIFactory)
93
 
 
94
 
    def restoreUIFactory(self):
95
 
        ui.ui_factory = self.old_factory
96
 
        sys.stdout = self.old_stdout
97
 
 
98
 
    def test_prompt_for_password(self):
99
 
        t = self.get_transport()
100
 
        # Ensure that the test framework set the password
101
 
        self.assertIsNot(t._password, None)
102
 
        # Reset the password (get_url set the password to 'bar' so we
103
 
        # reset it to None in the transport before the connection).
104
 
        password = t._password
105
 
        t._password = None
106
 
        ui.ui_factory = tests.TestUIFactory(stdin=password+'\n')
107
 
        # Ask the server to check the password
 
71
    def _add_authorized_user(self, user, password):
108
72
        server = self.get_server()
109
73
        # FIXME: There should be a better way to declare authorized users and
110
74
        # passwords to the server
111
75
        authorizer = server._ftp_server.authorizer
112
 
        authorizer.secured_user = t._user
 
76
        authorizer.secured_user = user
113
77
        authorizer.secured_password = password
 
78
 
 
79
    def test_prompt_for_password(self):
 
80
        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)
114
91
        # Issue a request to the server to connect
115
92
        t.has('whatever/not/existing')
116
93
        # stdin should be empty (the provided password have been consumed)
117
94
        self.assertEqual('', ui.ui_factory.stdin.readline())
 
95
 
 
96
    def test_no_prompt_for_password_when_using_auth_config(self):
 
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
        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
 
 
107
        # Create a config file with the right password
 
108
        conf = config.AuthenticationConfig()
 
109
        conf._get_config().update({'ftptest': {'scheme': 'ftp',
 
110
                                               'user': t._user,
 
111
                                               'password': password}})
 
112
        conf._save()
 
113
        # Issue a request to the server to connect
 
114
        t.put_bytes('foo', 'test bytes\n')
 
115
        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())