1
"""Tests for remote access to juju machines."""
27
class TestRemote(tests.FakeHomeTestCase):
29
precise_status_output = """\
38
public-address: 10.55.60.1
41
win2012hvr2_status_output = """\
50
public-address: 10.55.60.2
53
def test_remote_from_unit(self):
54
env = JujuData("an-env", {"type": "nonlocal"})
55
client = EnvJujuClient(env, None, None)
57
with patch.object(client, "get_status", autospec=True) as st:
58
st.return_value = Status.from_text(self.precise_status_output)
59
remote = remote_from_unit(client, unit)
62
"<SSHRemote env='an-env' unit='a-service/0'>")
63
self.assertIs(False, remote.is_windows())
65
def test_remote_from_unit_with_series(self):
66
env = JujuData("an-env", {"type": "nonlocal"})
67
client = EnvJujuClient(env, None, None)
69
remote = remote_from_unit(client, unit, series="trusty")
72
"<SSHRemote env='an-env' unit='a-service/0'>")
73
self.assertIs(False, remote.is_windows())
75
def test_remote_from_unit_with_status(self):
76
env = JujuData("an-env", {"type": "nonlocal"})
77
client = EnvJujuClient(env, None, None)
79
status = Status.from_text(self.win2012hvr2_status_output)
80
remote = remote_from_unit(client, unit, status=status)
83
"<WinRmRemote env='an-env' unit='a-service/0' addr='10.55.60.2'>")
84
self.assertIs(True, remote.is_windows())
86
def test_remote_from_address(self):
87
remote = remote_from_address("10.55.60.1")
88
self.assertEqual(repr(remote), "<SSHRemote addr='10.55.60.1'>")
89
self.assertIs(None, remote.is_windows())
91
def test_remote_from_address_and_series(self):
92
remote = remote_from_address("10.55.60.2", series="trusty")
93
self.assertEqual(repr(remote), "<SSHRemote addr='10.55.60.2'>")
94
self.assertIs(False, remote.is_windows())
96
def test_remote_from_address_and_win_series(self):
97
remote = remote_from_address("10.55.60.3", series="win2012hvr2")
98
self.assertEqual(repr(remote), "<WinRmRemote addr='10.55.60.3'>")
99
self.assertIs(True, remote.is_windows())
101
def test_run_with_unit(self):
102
env = JujuData("an-env", {"type": "nonlocal"})
103
client = EnvJujuClient(env, None, None)
105
remote = remote_from_unit(client, unit, series="trusty")
106
with patch.object(client, "get_juju_output") as mock_cmd:
107
mock_cmd.return_value = "contents of /a/file"
108
output = remote.run("cat /a/file")
109
self.assertEqual(output, "contents of /a/file")
110
mock_cmd.assert_called_once_with("ssh", unit, "cat /a/file",
113
def test_run_with_unit_fallback(self):
114
env = JujuData("an-env", {"type": "nonlocal"})
115
client = EnvJujuClient(env, None, None)
117
with patch.object(client, "get_status") as st:
118
st.return_value = Status.from_text(self.precise_status_output)
119
remote = remote_from_unit(client, unit)
120
with patch.object(client, "get_juju_output") as mock_gjo:
121
mock_gjo.side_effect = subprocess.CalledProcessError(1, "ssh",
123
with patch.object(remote, "_run_subprocess") as mock_run:
124
mock_run.return_value = "contents of /a/file"
125
output = remote.run("cat /a/file")
126
self.assertEqual(output, "contents of /a/file")
127
mock_gjo.assert_called_once_with("ssh", unit, "cat /a/file",
129
mock_run.assert_called_once_with([
132
"-o", "UserKnownHostsFile /dev/null",
133
"-o", "StrictHostKeyChecking no",
134
"-o", "PasswordAuthentication no",
138
self.assertRegexpMatches(
139
self.log_stream.getvalue(),
140
"(?m)^WARNING juju ssh to 'a-service/0' failed, .*")
142
def test_run_with_address(self):
143
remote = remote_from_address("10.55.60.1")
144
with patch.object(remote, "_run_subprocess") as mock_run:
145
mock_run.return_value = "contents of /a/file"
146
output = remote.run("cat /a/file")
147
self.assertEqual(output, "contents of /a/file")
148
mock_run.assert_called_once_with([
151
"-o", "UserKnownHostsFile /dev/null",
152
"-o", "StrictHostKeyChecking no",
153
"-o", "PasswordAuthentication no",
159
remote = remote_from_address("10.55.60.1")
160
with patch.object(remote, "_run_subprocess") as mock_run:
161
remote.cat("/a/file")
162
mock_run.assert_called_once_with([
165
"-o", "UserKnownHostsFile /dev/null",
166
"-o", "StrictHostKeyChecking no",
167
"-o", "PasswordAuthentication no",
172
def test_cat_on_windows(self):
173
env = JujuData("an-env", {"type": "nonlocal"})
174
client = EnvJujuClient(env, None, None)
176
with patch.object(client, "get_status", autospec=True) as st:
177
st.return_value = Status.from_text(self.win2012hvr2_status_output)
178
response = winrm.Response(("contents of /a/file", "", 0))
179
remote = remote_from_unit(client, unit)
180
with patch.object(remote.session, "run_cmd", autospec=True,
181
return_value=response) as mock_run:
182
output = remote.cat("/a/file")
183
self.assertEqual(output, "contents of /a/file")
184
st.assert_called_once_with()
185
mock_run.assert_called_once_with("type", ["/a/file"])
188
remote = remote_from_address("10.55.60.1")
190
with patch.object(remote, "_run_subprocess") as mock_run:
191
remote.copy(dest, ["/var/log/*", "~/.config"])
192
mock_run.assert_called_once_with([
196
"-o", "UserKnownHostsFile /dev/null",
197
"-o", "StrictHostKeyChecking no",
198
"-o", "PasswordAuthentication no",
199
"10.55.60.1:/var/log/*",
200
"10.55.60.1:~/.config",
204
def test_copy_on_windows(self):
205
env = JujuData("an-env", {"type": "nonlocal"})
206
client = EnvJujuClient(env, None, None)
209
with patch.object(client, "get_status", autospec=True) as st:
210
st.return_value = Status.from_text(self.win2012hvr2_status_output)
211
response = winrm.Response(("fake output", "", 0))
212
remote = remote_from_unit(client, unit)
213
with patch.object(remote.session, "run_ps", autospec=True,
214
return_value=response) as mock_run:
215
with patch.object(remote, "_encoded_copy_to_dir",
216
autospec=True) as mock_cpdir:
217
remote.copy(dest, ["C:\\logs\\*", "%APPDATA%\\*.log"])
218
mock_cpdir.assert_called_once_with(dest, "fake output")
219
st.assert_called_once_with()
220
self.assertEquals(mock_run.call_count, 1)
221
self.assertRegexpMatches(
222
mock_run.call_args[0][0],
223
r'.*"C:\\logs\\[*]","%APPDATA%\\[*].log".*')
225
def test_copy_ipv6(self):
226
remote = remote_from_address("2001:db8::34")
227
self.assertEqual(remote.address, "2001:db8::34")
229
with patch.object(remote, "_run_subprocess") as mock_run:
230
remote.copy(dest, ["/var/log/*", "~/.config"])
231
mock_run.assert_called_once_with([
235
"-o", "UserKnownHostsFile /dev/null",
236
"-o", "StrictHostKeyChecking no",
237
"-o", "PasswordAuthentication no",
238
"[2001:db8::34]:/var/log/*",
239
"[2001:db8::34]:~/.config",
243
def test_run_cmd(self):
244
env = JujuData("an-env", {"type": "nonlocal"})
245
client = EnvJujuClient(env, None, None)
247
with patch.object(client, "get_status", autospec=True) as st:
248
st.return_value = Status.from_text(self.win2012hvr2_status_output)
249
response = winrm.Response(("some out", "some err", 0))
250
remote = remote_from_unit(client, unit)
251
with patch.object(remote.session, "run_cmd", autospec=True,
252
return_value=response) as mock_run:
253
output = remote.run_cmd(
254
["C:\\Program Files\\bin.exe", "/IN", "Bob's Stuff"])
255
self.assertEqual(output, response)
256
st.assert_called_once_with()
257
mock_run.assert_called_once_with(
258
'"C:\\Program Files\\bin.exe"', ['/IN "Bob\'s Stuff"'])
260
def test_run_subprocess_timeout(self):
261
remote = remote_from_address("10.55.60.1")
263
with patch("subprocess.check_output", autospec=True) as mock_co:
264
remote.cat("/a/file")
265
mock_co.assert_called_once_with((
272
"-o", "UserKnownHostsFile /dev/null",
273
"-o", "StrictHostKeyChecking no",
274
"-o", "PasswordAuthentication no",
278
stdin=subprocess.PIPE,
281
def test_encoded_copy_to_dir_one(self):
282
output = "testfile|K0ktLuECAA==\r\n"
283
with temp_dir() as dest:
284
WinRmRemote._encoded_copy_to_dir(dest, output)
285
with open(os.path.join(dest, "testfile")) as f:
286
self.assertEqual(f.read(), "test\n")
288
def test_encoded_copy_to_dir_many(self):
289
output = "test one|K0ktLuECAA==\r\ntest two|K0ktLuECAA==\r\n\r\n"
290
with temp_dir() as dest:
291
WinRmRemote._encoded_copy_to_dir(dest, output)
292
for name in ("test one", "test two"):
293
with open(os.path.join(dest, name)) as f:
294
self.assertEqual(f.read(), "test\n")
296
def test_encoded_copy_traversal_guard(self):
297
output = "../../../etc/passwd|K0ktLuECAA==\r\n"
298
with temp_dir() as dest:
299
with self.assertRaises(ValueError):
300
WinRmRemote._encoded_copy_to_dir(dest, output)
added
removed
tests/test_remote.py
