~bzr-pqm/bzr/bzr.dev

4763.2.4 by John Arbash Meinel
merge bzr.2.1 in preparation for NEWS entry.
1
# Copyright (C) 2007-2010 Canonical Ltd
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
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
4183.7.1 by Sabin Iacob
update FSF mailing address
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
16
17
from bzrlib.tests import TestCase
18
from bzrlib.errors import SSHVendorNotFound, UnknownSSH
19
from bzrlib.transport.ssh import (
20
    OpenSSHSubprocessVendor,
2221.5.16 by Dmitry Vasiliev
Added comments for test_cached_vendor
21
    PLinkSubprocessVendor,
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
22
    SSHCorpSubprocessVendor,
5444.2.2 by Matthew Gordon
Added tests for GNU lsh support.
23
    LSHSubprocessVendor,
2221.5.16 by Dmitry Vasiliev
Added comments for test_cached_vendor
24
    SSHVendorManager,
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
25
    )
26
27
28
class TestSSHVendorManager(SSHVendorManager):
29
30
    _ssh_version_string = ""
31
32
    def set_ssh_version_string(self, version):
33
        self._ssh_version_string = version
34
35
    def _get_ssh_version_string(self, args):
36
        return self._ssh_version_string
37
38
39
class SSHVendorManagerTests(TestCase):
40
41
    def test_register_vendor(self):
42
        manager = TestSSHVendorManager()
43
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
2221.5.6 by Dmitry Vasiliev
Changed tests to make sure the vendor manager returns the same object as was registered
44
        vendor = object()
45
        manager.register_vendor("vendor", vendor)
46
        self.assertIs(manager.get_vendor({"BZR_SSH": "vendor"}), vendor)
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
47
48
    def test_default_vendor(self):
49
        manager = TestSSHVendorManager()
50
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
2221.5.6 by Dmitry Vasiliev
Changed tests to make sure the vendor manager returns the same object as was registered
51
        vendor = object()
52
        manager.register_default_vendor(vendor)
53
        self.assertIs(manager.get_vendor({}), vendor)
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
54
55
    def test_get_vendor_by_environment(self):
56
        manager = TestSSHVendorManager()
57
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
58
        self.assertRaises(UnknownSSH,
59
            manager.get_vendor, {"BZR_SSH": "vendor"})
2221.5.6 by Dmitry Vasiliev
Changed tests to make sure the vendor manager returns the same object as was registered
60
        vendor = object()
61
        manager.register_vendor("vendor", vendor)
62
        self.assertIs(manager.get_vendor({"BZR_SSH": "vendor"}), vendor)
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
63
64
    def test_get_vendor_by_inspection_openssh(self):
65
        manager = TestSSHVendorManager()
66
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
67
        manager.set_ssh_version_string("OpenSSH")
68
        self.assertIsInstance(manager.get_vendor({}), OpenSSHSubprocessVendor)
69
70
    def test_get_vendor_by_inspection_sshcorp(self):
71
        manager = TestSSHVendorManager()
72
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
73
        manager.set_ssh_version_string("SSH Secure Shell")
74
        self.assertIsInstance(manager.get_vendor({}), SSHCorpSubprocessVendor)
75
5444.2.2 by Matthew Gordon
Added tests for GNU lsh support.
76
    def test_get_vendor_by_inspection_lsh(self):
77
        manager = TestSSHVendorManager()
78
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
79
        manager.set_ssh_version_string("lsh")
80
        self.assertIsInstance(manager.get_vendor({}), LSHSubprocessVendor)
81
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
82
    def test_get_vendor_by_inspection_plink(self):
83
        manager = TestSSHVendorManager()
84
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
85
        manager.set_ssh_version_string("plink")
4636.3.1 by Alexander Belchenko
Disabled auto-detection of plink SSH client to avoid many problems. On Windows paramiko wil be used as default. Nevertheless user can force usage of plink via env variable BZR_SSH=plink
86
        # Auto-detect of plink vendor disabled, on Windows recommended
87
        # default ssh-client is paramiko
88
        # see https://bugs.launchpad.net/bugs/414743
89
        #~self.assertIsInstance(manager.get_vendor({}), PLinkSubprocessVendor)
90
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
91
92
    def test_cached_vendor(self):
93
        manager = TestSSHVendorManager()
94
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
2221.5.6 by Dmitry Vasiliev
Changed tests to make sure the vendor manager returns the same object as was registered
95
        vendor = object()
96
        manager.register_vendor("vendor", vendor)
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
97
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
2221.5.16 by Dmitry Vasiliev
Added comments for test_cached_vendor
98
        # Once the vendor is found the result is cached (mainly because of the
99
        # 'get_vendor' sometimes can be an expensive operation) and later
100
        # invocations of the 'get_vendor' just returns the cached value.
2221.5.6 by Dmitry Vasiliev
Changed tests to make sure the vendor manager returns the same object as was registered
101
        self.assertIs(manager.get_vendor({"BZR_SSH": "vendor"}), vendor)
102
        self.assertIs(manager.get_vendor({}), vendor)
2221.5.16 by Dmitry Vasiliev
Added comments for test_cached_vendor
103
        # The cache can be cleared by the 'clear_cache' method
104
        manager.clear_cache()
105
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
106
2221.5.17 by Dmitry Vasiliev
Added comments for test_get_vendor_search_order
107
    def test_get_vendor_search_order(self):
108
        # The 'get_vendor' method search for SSH vendors as following:
109
        #
110
        #   1. Check previously cached value
111
        #   2. Check BZR_SSH environment variable
112
        #   3. Check the system for known SSH vendors
113
        #   4. Fall back to the default vendor if registered
114
        #
115
        # Let's now check the each check method in the reverse order
116
        # clearing the cache between each invocation:
117
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
118
        manager = TestSSHVendorManager()
2221.5.17 by Dmitry Vasiliev
Added comments for test_get_vendor_search_order
119
        # At first no vendors are found
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
120
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
121
2221.5.17 by Dmitry Vasiliev
Added comments for test_get_vendor_search_order
122
        # If the default vendor is registered it will be returned
2221.5.6 by Dmitry Vasiliev
Changed tests to make sure the vendor manager returns the same object as was registered
123
        default_vendor = object()
124
        manager.register_default_vendor(default_vendor)
125
        self.assertIs(manager.get_vendor({}), default_vendor)
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
126
2221.5.17 by Dmitry Vasiliev
Added comments for test_get_vendor_search_order
127
        # If the known vendor is found in the system it will be returned
2221.5.8 by Dmitry Vasiliev
Added SSHVendorManager.clear_cache() method
128
        manager.clear_cache()
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
129
        manager.set_ssh_version_string("OpenSSH")
130
        self.assertIsInstance(manager.get_vendor({}), OpenSSHSubprocessVendor)
131
2221.5.17 by Dmitry Vasiliev
Added comments for test_get_vendor_search_order
132
        # If the BZR_SSH environment variable is found it will be treated as
133
        # the vendor name
2221.5.8 by Dmitry Vasiliev
Added SSHVendorManager.clear_cache() method
134
        manager.clear_cache()
2221.5.6 by Dmitry Vasiliev
Changed tests to make sure the vendor manager returns the same object as was registered
135
        vendor = object()
136
        manager.register_vendor("vendor", vendor)
137
        self.assertIs(manager.get_vendor({"BZR_SSH": "vendor"}), vendor)
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
138
2221.5.17 by Dmitry Vasiliev
Added comments for test_get_vendor_search_order
139
        # Last cached value always checked first
2221.5.6 by Dmitry Vasiliev
Changed tests to make sure the vendor manager returns the same object as was registered
140
        self.assertIs(manager.get_vendor({}), vendor)
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
141
4595.17.1 by Martin
Add ability to give a path to a particular ssh client in BZR_SSH envvar
142
    def test_get_vendor_from_path_win32_plink(self):
143
        manager = TestSSHVendorManager()
144
        manager.set_ssh_version_string("plink: Release 0.60")
145
        plink_path = "C:/Program Files/PuTTY/plink.exe"
146
        vendor = manager.get_vendor({"BZR_SSH": plink_path})
147
        self.assertIsInstance(vendor, PLinkSubprocessVendor)
148
        args = vendor._get_vendor_specific_argv("user", "host", 22, ["bzr"])
149
        self.assertEqual(args[0], plink_path)
150
151
    def test_get_vendor_from_path_nix_openssh(self):
152
        manager = TestSSHVendorManager()
153
        manager.set_ssh_version_string(
154
            "OpenSSH_5.1p1 Debian-5, OpenSSL, 0.9.8g 19 Oct 2007")
155
        openssh_path = "/usr/bin/ssh"
156
        vendor = manager.get_vendor({"BZR_SSH": openssh_path})
157
        self.assertIsInstance(vendor, OpenSSHSubprocessVendor)
158
        args = vendor._get_vendor_specific_argv("user", "host", 22, ["bzr"])
159
        self.assertEqual(args[0], openssh_path)
160
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
161
162
class SubprocessVendorsTests(TestCase):
163
2221.5.3 by Dmitry Vasiliev
Fixed plink's arguments order. Added tests for such a case.
164
    def test_openssh_command_arguments(self):
165
        vendor = OpenSSHSubprocessVendor()
166
        self.assertEqual(
167
            vendor._get_vendor_specific_argv(
168
                "user", "host", 100, command=["bzr"]),
169
            ["ssh", "-oForwardX11=no", "-oForwardAgent=no",
5459.4.1 by Neil Martinsen-Burrell
dont force openssh to use protocol=2
170
                "-oClearAllForwardings=yes",
2221.5.3 by Dmitry Vasiliev
Fixed plink's arguments order. Added tests for such a case.
171
                "-oNoHostAuthenticationForLocalhost=yes",
172
                "-p", "100",
173
                "-l", "user",
174
                "host", "bzr"]
175
            )
176
177
    def test_openssh_subsystem_arguments(self):
178
        vendor = OpenSSHSubprocessVendor()
179
        self.assertEqual(
180
            vendor._get_vendor_specific_argv(
181
                "user", "host", 100, subsystem="sftp"),
182
            ["ssh", "-oForwardX11=no", "-oForwardAgent=no",
5459.4.1 by Neil Martinsen-Burrell
dont force openssh to use protocol=2
183
                "-oClearAllForwardings=yes",
2221.5.3 by Dmitry Vasiliev
Fixed plink's arguments order. Added tests for such a case.
184
                "-oNoHostAuthenticationForLocalhost=yes",
185
                "-p", "100",
186
                "-l", "user",
187
                "-s", "host", "sftp"]
188
            )
189
190
    def test_sshcorp_command_arguments(self):
191
        vendor = SSHCorpSubprocessVendor()
192
        self.assertEqual(
193
            vendor._get_vendor_specific_argv(
194
                "user", "host", 100, command=["bzr"]),
195
            ["ssh", "-x",
196
                "-p", "100",
197
                "-l", "user",
198
                "host", "bzr"]
199
            )
200
201
    def test_sshcorp_subsystem_arguments(self):
202
        vendor = SSHCorpSubprocessVendor()
203
        self.assertEqual(
204
            vendor._get_vendor_specific_argv(
205
                "user", "host", 100, subsystem="sftp"),
206
            ["ssh", "-x",
207
                "-p", "100",
208
                "-l", "user",
209
                "-s", "sftp", "host"]
210
            )
211
5444.2.2 by Matthew Gordon
Added tests for GNU lsh support.
212
    def test_lsh_command_arguments(self):
213
        vendor = LSHSubprocessVendor()
214
        self.assertEqual(
215
            vendor._get_vendor_specific_argv(
216
                "user", "host", 100, command=["bzr"]),
217
            ["lsh",
218
                "-p", "100",
219
                "-l", "user",
220
                "host", "bzr"]
221
            )
222
223
    def test_lsh_subsystem_arguments(self):
224
        vendor = LSHSubprocessVendor()
225
        self.assertEqual(
226
            vendor._get_vendor_specific_argv(
227
                "user", "host", 100, subsystem="sftp"),
228
            ["lsh",
229
                "-p", "100",
230
                "-l", "user",
231
                "--subsystem", "sftp", "host"]
232
            )
233
2221.5.3 by Dmitry Vasiliev
Fixed plink's arguments order. Added tests for such a case.
234
    def test_plink_command_arguments(self):
235
        vendor = PLinkSubprocessVendor()
236
        self.assertEqual(
237
            vendor._get_vendor_specific_argv(
238
                "user", "host", 100, command=["bzr"]),
3220.1.2 by Dmitry Vasiliev
Re-enabled auto-detection of plink vendor and fixed tests
239
            ["plink", "-x", "-a", "-ssh", "-2", "-batch",
2221.5.3 by Dmitry Vasiliev
Fixed plink's arguments order. Added tests for such a case.
240
                "-P", "100",
241
                "-l", "user",
242
                "host", "bzr"]
243
            )
244
245
    def test_plink_subsystem_arguments(self):
246
        vendor = PLinkSubprocessVendor()
247
        self.assertEqual(
248
            vendor._get_vendor_specific_argv(
249
                "user", "host", 100, subsystem="sftp"),
3220.1.2 by Dmitry Vasiliev
Re-enabled auto-detection of plink vendor and fixed tests
250
            ["plink", "-x", "-a", "-ssh", "-2", "-batch",
2221.5.3 by Dmitry Vasiliev
Fixed plink's arguments order. Added tests for such a case.
251
                "-P", "100",
252
                "-l", "user",
253
                "-s", "host", "sftp"]
2221.5.1 by Dmitry Vasiliev
Added support for Putty's SSH implementation
254
            )