1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
|
from __future__ import unicode_literals
import io
import socket
import spur
import spur.ssh
from .assertions import assert_equal, assert_raises
from .testing import create_ssh_shell, HOSTNAME, PORT, PASSWORD, USERNAME
from .process_test_set import ProcessTestSet
from .open_test_set import OpenTestSet
class SshTestMixin(object):
def create_shell(self):
return create_ssh_shell()
class SshOpenTests(OpenTestSet, SshTestMixin):
pass
class SshProcessTests(ProcessTestSet, SshTestMixin):
pass
def test_attempting_to_connect_to_wrong_port_raises_connection_error():
def try_connection():
shell = _create_shell_with_wrong_port()
shell.run(["echo", "hello"])
assert_raises(spur.ssh.ConnectionError, try_connection)
def test_connection_error_contains_original_error():
try:
shell = _create_shell_with_wrong_port()
shell.run(["true"])
# Expected error
assert False
except spur.ssh.ConnectionError as error:
assert isinstance(error.original_error, IOError)
def test_connection_error_contains_traceback_for_original_error():
try:
shell = _create_shell_with_wrong_port()
shell.run(["true"])
# Expected error
assert False
except spur.ssh.ConnectionError as error:
assert "Traceback (most recent call last):" in error.original_traceback
def test_missing_host_key_set_to_accept_allows_connection_with_missing_host_key():
with create_ssh_shell(missing_host_key=spur.ssh.MissingHostKey.accept) as shell:
shell.run(["true"])
def test_missing_host_key_set_to_warn_allows_connection_with_missing_host_key():
with create_ssh_shell(missing_host_key=spur.ssh.MissingHostKey.warn) as shell:
shell.run(["true"])
def test_missing_host_key_set_to_raise_error_raises_error_when_missing_host_key():
with create_ssh_shell(missing_host_key=spur.ssh.MissingHostKey.raise_error) as shell:
assert_raises(spur.ssh.ConnectionError, lambda: shell.run(["true"]))
def test_trying_to_use_ssh_shell_after_exit_results_in_error():
with create_ssh_shell() as shell:
pass
assert_raises(Exception, lambda: shell.run(["true"]))
def test_an_open_socket_can_be_used_for_ssh_connection_with_sock_argument():
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((HOSTNAME, PORT))
with _create_shell_with_wrong_port(sock=sock) as shell:
result = shell.run(["echo", "hello"])
assert_equal(b"hello\n", result.output)
def _create_shell_with_wrong_port(**kwargs):
return spur.SshShell(
username=USERNAME,
password=PASSWORD,
hostname=HOSTNAME,
port=54321,
missing_host_key=spur.ssh.MissingHostKey.accept,
**kwargs
)
class MinimalSshTestMixin(object):
def create_shell(self):
return create_ssh_shell(shell_type=spur.ssh.ShellTypes.minimal)
class MinimalSshOpenTests(OpenTestSet, MinimalSshTestMixin):
pass
class MinimalSshProcessTests(ProcessTestSet, MinimalSshTestMixin):
test_spawning_command_that_uses_path_env_variable_asks_if_command_is_installed = None
test_spawning_non_existent_command_raises_specific_no_such_command_exception = None
test_can_get_process_id_of_process_if_store_pid_is_true = None
test_can_send_signal_to_process_if_store_pid_is_set = None
# cwd is not supported when using a minimal shell
test_using_non_existent_cwd_raises_could_not_change_directory_error = None
test_attempting_to_change_directory_without_permissions_raises_cannot_change_directory_error = None
test_using_non_existent_cwd_and_command_raises_could_not_change_directory_error = None
test_using_non_existent_command_and_correct_cwd_raises_no_such_command_exception = None
test_can_find_command_in_cwd = None
def test_cannot_store_pid(self):
self._assert_unsupported_feature(store_pid=True)
test_cwd_of_run_can_be_set = None
def test_cannot_set_cwd(self):
self._assert_unsupported_feature(cwd="/")
test_environment_variables_can_be_added_for_run = None
def test_update_env_can_be_empty(self):
self._assert_supported_feature(update_env={})
def test_cannot_update_env(self):
self._assert_unsupported_feature(update_env={"x": "one"})
def test_cannot_set_new_process_group(self):
self._assert_unsupported_feature(new_process_group=True)
def _assert_supported_feature(self, **kwargs):
with self.create_shell() as shell:
result = shell.run(["echo", "hello"], **kwargs)
assert_equal(b"hello\n", result.output)
def _assert_unsupported_feature(self, **kwargs):
name, = kwargs.keys()
try:
with self.create_shell() as shell:
shell.run(["echo", "hello"], **kwargs)
assert False, "Expected error"
except spur.ssh.UnsupportedArgumentError as error:
assert_equal("'{0}' is not supported when using a minimal shell".format(name), str(error))
class ReadInitializationLineTests(object):
def test_reading_initialization_line_returns_int_from_line_of_file(self):
assert_equal(42, spur.ssh._read_int_initialization_line(io.StringIO("42\n")))
def test_blank_lines_are_skipped(self):
assert_equal(42, spur.ssh._read_int_initialization_line(io.StringIO("\n \n\t\t\n42\n")))
def test_error_if_non_blank_line_is_not_integer(self):
try:
spur.ssh._read_int_initialization_line(io.StringIO("x\n"))
assert False, "Expected error"
except spur.CommandInitializationError as error:
assert "Failed to parse line 'x' as integer" in str(error)
|
