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
|
import os
import time
from io import StringIO
from invoke import pty_size, CommandTimedOut
from invocations.environment import in_ci
from pytest import skip, raises
from pytest_relaxed import trap
from fabric import Connection, Config
# TODO: use pytest markers
def skip_outside_ci():
if not in_ci():
skip()
class Connection_:
class ssh_connections:
def open_method_generates_real_connection(self):
c = Connection("localhost")
c.open()
assert c.client.get_transport().active is True
assert c.is_connected is True
return c
def close_method_closes_connection(self):
# Handy shortcut - open things up, then return Connection for us to
# close
c = self.open_method_generates_real_connection()
c.close()
assert c.client.get_transport() is None
assert c.is_connected is False
class run:
def simple_command_on_host(self):
"""
Run command on localhost
"""
result = Connection("localhost").run("echo foo", hide=True)
assert result.stdout == "foo\n"
assert result.exited == 0
assert result.ok is True
def simple_command_with_pty(self):
"""
Run command under PTY on localhost
"""
# Most Unix systems should have stty, which asplodes when not run
# under a pty, and prints useful info otherwise
result = Connection("localhost").run(
"stty size", hide=True, pty=True
)
found = result.stdout.strip().split()
cols, rows = pty_size()
assert tuple(map(int, found)), rows == cols
# PTYs use \r\n, not \n, line separation
assert "\r\n" in result.stdout
assert result.pty is True
class shell:
@trap
def base_case(self):
result = Connection("localhost").shell(
# Some extra newlines to make sure it doesn't get split up by
# motd/prompt
in_stream=StringIO("\n\nexit\n")
)
assert result.command is None
assert "exit" in result.stdout
assert result.stderr == ""
assert result.exited == 0
assert result.pty is True
class local:
def wraps_invoke_run(self):
# NOTE: most of the interesting tests about this are in
# invoke.runners / invoke.integration.
cxn = Connection("localhost")
result = cxn.local("echo foo", hide=True)
assert result.stdout == "foo\n"
assert not cxn.is_connected # meh way of proving it didn't use SSH
def mixed_use_of_local_and_run(self):
"""
Run command truly locally, and over SSH via localhost
"""
cxn = Connection("localhost")
result = cxn.local("echo foo", hide=True)
assert result.stdout == "foo\n"
assert not cxn.is_connected # meh way of proving it didn't use SSH yet
result = cxn.run("echo foo", hide=True)
assert cxn.is_connected # NOW it's using SSH
assert result.stdout == "foo\n"
class sudo:
def setup(self):
# NOTE: assumes a user configured for passworded (NOT
# passwordless)_sudo, whose password is 'mypass', is executing the
# test suite. I.e. our travis-ci setup.
config = Config(
{"sudo": {"password": "mypass"}, "run": {"hide": True}}
)
self.cxn = Connection("localhost", config=config)
def sudo_command(self):
"""
Run command via sudo on host localhost
"""
skip_outside_ci()
assert self.cxn.sudo("whoami").stdout.strip() == "root"
def mixed_sudo_and_normal_commands(self):
"""
Run command via sudo, and not via sudo, on localhost
"""
skip_outside_ci()
logname = os.environ["LOGNAME"]
assert self.cxn.run("whoami").stdout.strip() == logname
assert self.cxn.sudo("whoami").stdout.strip() == "root"
def large_remote_commands_finish_cleanly(self):
# Guards against e.g. cleanup finishing before actually reading all
# data from the remote end. Which is largely an issue in Invoke-level
# code but one that only really manifests when doing stuff over the
# network. Yay computers!
path = "/usr/share/dict/words"
cxn = Connection("localhost")
with open(path) as fd:
words = [x.strip() for x in fd.readlines()]
stdout = cxn.run("cat {}".format(path), hide=True).stdout
lines = [x.strip() for x in stdout.splitlines()]
# When bug present, # lines received is significantly fewer than the
# true count in the file (by thousands).
assert len(lines) == len(words)
class command_timeout:
def setup(self):
self.cxn = Connection("localhost")
def does_not_raise_exception_when_under_timeout(self):
assert self.cxn.run("sleep 1", timeout=3)
def raises_exception_when_over_timeout(self):
with raises(CommandTimedOut) as info:
start = time.time()
self.cxn.run("sleep 5", timeout=1)
elapsed = time.time() - start
assert info.value.timeout == 1
# Catch scenarios where we except but don't actually shut down
# early (w/ a bit of fudge time for overhead)
assert elapsed <= 2
|
