/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 breezy/tests/test_ssh_transport.py

  • Committer: Jelmer Vernooij
  • Date: 2019-09-21 17:08:09 UTC
  • mfrom: (7389 work)
  • mto: This revision was merged to the branch mainline in revision 7390.
  • Revision ID: jelmer@jelmer.uk-20190921170809-ejewbeue585deajo
merge trunk.

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/tests/test_ssh_transport.py
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
 
from breezy.tests import TestCase
 
17
from breezy import config
 
18
from breezy.tests import TestCase, TestCaseWithTransport
18
19
from breezy.errors import SSHVendorNotFound, UnknownSSH
19
20
from breezy.transport.ssh import (
20
21
    OpenSSHSubprocessVendor,
37
38
        return self._ssh_version_string
38
39
 
39
40
 
40
 
class SSHVendorManagerTests(TestCase):
 
41
class SSHVendorManagerTests(TestCaseWithTransport):
41
42
 
42
43
    def test_register_vendor(self):
43
44
        manager = TestSSHVendorManager()
44
 
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
 
45
        self.overrideEnv('BRZ_SSH', None)
 
46
        self.assertRaises(SSHVendorNotFound, manager.get_vendor)
45
47
        vendor = object()
46
48
        manager.register_vendor("vendor", vendor)
47
 
        self.assertIs(manager.get_vendor({"BRZ_SSH": "vendor"}), vendor)
 
49
        self.overrideEnv('BRZ_SSH', 'vendor')
 
50
        self.assertIs(manager.get_vendor(), vendor)
48
51
 
49
52
    def test_default_vendor(self):
50
53
        manager = TestSSHVendorManager()
51
 
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
 
54
        self.overrideEnv('BRZ_SSH', None)
 
55
        self.assertRaises(SSHVendorNotFound, manager.get_vendor)
52
56
        vendor = object()
53
57
        manager.register_default_vendor(vendor)
54
 
        self.assertIs(manager.get_vendor({}), vendor)
 
58
        self.assertIs(manager.get_vendor(), vendor)
55
59
 
56
60
    def test_get_vendor_by_environment(self):
57
61
        manager = TestSSHVendorManager()
58
 
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
59
 
        self.assertRaises(UnknownSSH,
60
 
                          manager.get_vendor, {"BRZ_SSH": "vendor"})
61
 
        vendor = object()
62
 
        manager.register_vendor("vendor", vendor)
63
 
        self.assertIs(manager.get_vendor({"BRZ_SSH": "vendor"}), vendor)
 
62
        self.overrideEnv('BRZ_SSH', None)
 
63
        self.assertRaises(SSHVendorNotFound, manager.get_vendor)
 
64
        self.overrideEnv('BRZ_SSH', 'vendor')
 
65
        self.assertRaises(UnknownSSH, manager.get_vendor)
 
66
        vendor = object()
 
67
        manager.register_vendor("vendor", vendor)
 
68
        self.assertIs(manager.get_vendor(), vendor)
 
69
 
 
70
    def test_get_vendor_by_config(self):
 
71
        manager = TestSSHVendorManager()
 
72
        self.overrideEnv('BRZ_SSH', None)
 
73
        self.assertRaises(SSHVendorNotFound, manager.get_vendor)
 
74
        config.GlobalStack().set('ssh', 'vendor')
 
75
        self.assertRaises(UnknownSSH, manager.get_vendor)
 
76
        vendor = object()
 
77
        manager.register_vendor("vendor", vendor)
 
78
        self.assertIs(manager.get_vendor(), vendor)
64
79
 
65
80
    def test_get_vendor_by_inspection_openssh(self):
66
81
        manager = TestSSHVendorManager()
67
 
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
 
82
        self.overrideEnv('BRZ_SSH', None)
 
83
        self.assertRaises(SSHVendorNotFound, manager.get_vendor)
68
84
        manager.set_ssh_version_string("OpenSSH")
69
 
        self.assertIsInstance(manager.get_vendor({}), OpenSSHSubprocessVendor)
 
85
        self.assertIsInstance(manager.get_vendor(), OpenSSHSubprocessVendor)
70
86
 
71
87
    def test_get_vendor_by_inspection_sshcorp(self):
72
88
        manager = TestSSHVendorManager()
73
 
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
 
89
        self.overrideEnv('BRZ_SSH', None)
 
90
        self.assertRaises(SSHVendorNotFound, manager.get_vendor)
74
91
        manager.set_ssh_version_string("SSH Secure Shell")
75
 
        self.assertIsInstance(manager.get_vendor({}), SSHCorpSubprocessVendor)
 
92
        self.assertIsInstance(manager.get_vendor(), SSHCorpSubprocessVendor)
76
93
 
77
94
    def test_get_vendor_by_inspection_lsh(self):
78
95
        manager = TestSSHVendorManager()
79
 
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
 
96
        self.overrideEnv('BRZ_SSH', None)
 
97
        self.assertRaises(SSHVendorNotFound, manager.get_vendor)
80
98
        manager.set_ssh_version_string("lsh")
81
 
        self.assertIsInstance(manager.get_vendor({}), LSHSubprocessVendor)
 
99
        self.assertIsInstance(manager.get_vendor(), LSHSubprocessVendor)
82
100
 
83
101
    def test_get_vendor_by_inspection_plink(self):
84
102
        manager = TestSSHVendorManager()
85
 
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
 
103
        self.overrideEnv('BRZ_SSH', None)
 
104
        self.assertRaises(SSHVendorNotFound, manager.get_vendor)
86
105
        manager.set_ssh_version_string("plink")
87
106
        # Auto-detect of plink vendor disabled, on Windows recommended
88
107
        # default ssh-client is paramiko
89
108
        # see https://bugs.launchpad.net/bugs/414743
90
 
        #~self.assertIsInstance(manager.get_vendor({}), PLinkSubprocessVendor)
91
 
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
 
109
        #~self.assertIsInstance(manager.get_vendor(), PLinkSubprocessVendor)
 
110
        self.assertRaises(SSHVendorNotFound, manager.get_vendor)
92
111
 
93
112
    def test_cached_vendor(self):
94
113
        manager = TestSSHVendorManager()
95
 
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
 
114
        self.overrideEnv('BRZ_SSH', None)
 
115
        self.assertRaises(SSHVendorNotFound, manager.get_vendor)
96
116
        vendor = object()
97
117
        manager.register_vendor("vendor", vendor)
98
 
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
 
118
        self.assertRaises(SSHVendorNotFound, manager.get_vendor)
99
119
        # Once the vendor is found the result is cached (mainly because of the
100
120
        # 'get_vendor' sometimes can be an expensive operation) and later
101
121
        # invocations of the 'get_vendor' just returns the cached value.
102
 
        self.assertIs(manager.get_vendor({"BRZ_SSH": "vendor"}), vendor)
103
 
        self.assertIs(manager.get_vendor({}), vendor)
 
122
        self.overrideEnv('BRZ_SSH', 'vendor')
 
123
        self.assertIs(manager.get_vendor(), vendor)
 
124
        self.overrideEnv('BRZ_SSH', None)
 
125
        self.assertIs(manager.get_vendor(), vendor)
104
126
        # The cache can be cleared by the 'clear_cache' method
105
127
        manager.clear_cache()
106
 
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
 
128
        self.overrideEnv('BRZ_SSH', None)
 
129
        self.assertRaises(SSHVendorNotFound, manager.get_vendor)
107
130
 
108
131
    def test_get_vendor_search_order(self):
109
132
        # The 'get_vendor' method search for SSH vendors as following:
118
141
 
119
142
        manager = TestSSHVendorManager()
120
143
        # At first no vendors are found
121
 
        self.assertRaises(SSHVendorNotFound, manager.get_vendor, {})
 
144
        self.overrideEnv('BRZ_SSH', None)
 
145
        self.assertRaises(SSHVendorNotFound, manager.get_vendor)
122
146
 
123
147
        # If the default vendor is registered it will be returned
124
148
        default_vendor = object()
125
149
        manager.register_default_vendor(default_vendor)
126
 
        self.assertIs(manager.get_vendor({}), default_vendor)
 
150
        self.assertIs(manager.get_vendor(), default_vendor)
127
151
 
128
152
        # If the known vendor is found in the system it will be returned
129
153
        manager.clear_cache()
130
154
        manager.set_ssh_version_string("OpenSSH")
131
 
        self.assertIsInstance(manager.get_vendor({}), OpenSSHSubprocessVendor)
 
155
        self.assertIsInstance(manager.get_vendor(), OpenSSHSubprocessVendor)
132
156
 
133
157
        # If the BRZ_SSH environment variable is found it will be treated as
134
158
        # the vendor name
135
159
        manager.clear_cache()
136
160
        vendor = object()
137
161
        manager.register_vendor("vendor", vendor)
138
 
        self.assertIs(manager.get_vendor({"BRZ_SSH": "vendor"}), vendor)
 
162
        self.overrideEnv('BRZ_SSH', 'vendor')
 
163
        self.assertIs(manager.get_vendor(), vendor)
139
164
 
140
165
        # Last cached value always checked first
141
 
        self.assertIs(manager.get_vendor({}), vendor)
 
166
        self.overrideEnv('BRZ_SSH', 'vendor')
 
167
        self.assertIs(manager.get_vendor(), vendor)
142
168
 
143
169
    def test_get_vendor_from_path_win32_plink(self):
144
170
        manager = TestSSHVendorManager()
145
171
        manager.set_ssh_version_string("plink: Release 0.60")
146
172
        plink_path = "C:/Program Files/PuTTY/plink.exe"
147
 
        vendor = manager.get_vendor({"BRZ_SSH": plink_path})
 
173
        self.overrideEnv('BRZ_SSH', plink_path)
 
174
        vendor = manager.get_vendor()
148
175
        self.assertIsInstance(vendor, PLinkSubprocessVendor)
149
176
        args = vendor._get_vendor_specific_argv("user", "host", 22, ["bzr"])
150
177
        self.assertEqual(args[0], plink_path)
154
181
        manager.set_ssh_version_string(
155
182
            "OpenSSH_5.1p1 Debian-5, OpenSSL, 0.9.8g 19 Oct 2007")
156
183
        openssh_path = "/usr/bin/ssh"
157
 
        vendor = manager.get_vendor({"BRZ_SSH": openssh_path})
 
184
        self.overrideEnv('BRZ_SSH', openssh_path)
 
185
        vendor = manager.get_vendor()
158
186
        self.assertIsInstance(vendor, OpenSSHSubprocessVendor)
159
187
        args = vendor._get_vendor_specific_argv("user", "host", 22, ["bzr"])
160
188
        self.assertEqual(args[0], openssh_path)