Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"bundlewrap",
{
'full_name': "Blöck Wart",
'gid': 2345,
'groups': ["group1", "group2"],
'home': "/home/bundlewrap",
'password_hash': "secret",
'shell': "/bin/bash",
'uid': 1123,
},
)
passwd_grep_result = RunResult()
passwd_grep_result.return_code = 0
passwd_grep_result.stdout = "bundlewrap:x:1123:2345:Blöck Wart:/home/bundlewrap:/bin/bash\n"
shadow_grep_result = RunResult()
shadow_grep_result.return_code = 0
shadow_grep_result.stdout = "bundlewrap:topsecret:::::::"
results = [shadow_grep_result, passwd_grep_result]
def pop_result(*args, **kwargs):
return results.pop()
bundle.node.run.side_effect = pop_result
status = user.get_status()
self.assertFalse(status.correct)
def test_short_mode(self):
node = MagicMock()
run_result = RunResult()
run_result.stdout = "user:group:666:4321"
node.run.return_value = run_result
stat_result = remote.stat(node, "/dev/null")
self.assertEqual(stat_result, {
'owner': "user",
'group': "group",
'mode': "0666",
'size': 4321,
})
{
'full_name': "Blöck Wart",
'gid': 2345,
'groups': ["group1", "group2"],
'home': "/home/bundlewrap",
'password_hash': "topsecret",
'shell': "/bin/bash",
'uid': 1123,
'use_shadow': False,
},
)
passwd_grep_result = RunResult()
passwd_grep_result.return_code = 0
passwd_grep_result.stdout = "bundlewrap:x:666:666:Blöck Wart:/home/bundlewrap:/bin/bash\n"
shadow_grep_result = RunResult()
shadow_grep_result.return_code = 0
shadow_grep_result.stdout = "bundlewrap:secret:::::::"
results = [shadow_grep_result, passwd_grep_result]
def pop_result(*args, **kwargs):
return results.pop()
bundle.node.run.side_effect = pop_result
status = user.get_status()
self.assertFalse(status.correct)
def test_installed(self):
runresult = RunResult()
runresult.return_code = 0
runresult.stdout = "Status: install ok installed\n"
node = MagicMock()
node.run.return_value = runresult
self.assertTrue(pkg_pip.pkg_installed(node, "foo"))
def test_not_installed(self):
runresult = RunResult()
runresult.return_code = 1
runresult.stdout = (
"Package `foo' is not installed and no info is available.\n"
"Use dpkg --info (= dpkg-deb --info) to examine archive files,\n"
"and dpkg --contents (= dpkg-deb --contents) to list their contents.\n"
)
node = MagicMock()
node.run.return_value = runresult
self.assertFalse(pkg_apt.pkg_installed(node, "foo"))
def test_not_running(self):
runresult = RunResult()
runresult.return_code = 3
runresult.stdout = "whatever, does not matter"
node = MagicMock()
node.run.return_value = runresult
self.assertFalse(svc_systemd.svc_running(node, "foo"))
def test_not_installed(self):
runresult = RunResult()
runresult.return_code = 1
runresult.stdout = "Error: No matching Packages to list\n"
node = MagicMock()
node.run.return_value = runresult
self.assertFalse(pkg_yum.pkg_installed(node, "foo"))
def test_groups(self):
node = MagicMock()
result1 = RunResult()
result1.stdout = "group1 group2\n"
result2 = RunResult()
result2.stdout = "group1\n"
results = [result2, result1]
def get_result(*args, **kwargs):
return results.pop()
node.run.side_effect = get_result
groups = users._groups_for_user(node, "jdoe")
self.assertEqual(groups, ["group2"])
#
# Luckily stdout is a somewhat simpler affair: we can just close
# the writing end of the pipe, causing the reader thread to
# shut down as it sees the EOF.
quit_event.set()
close(stdout_fd_w)
stdout_thread.join()
stderr_thread.join()
stdout_lb.close()
stderr_lb.close()
for fd in (stdout_fd_r, stderr_fd_r, stderr_fd_w):
close(fd)
io.debug("command finished with return code {}".format(ssh_process.returncode))
result = RunResult()
result.stdout = stdout_lb.record.getvalue()
result.stderr = stderr_lb.record.getvalue()
result.return_code = ssh_process.returncode
if result.return_code != 0 and (not ignore_failure or result.return_code == 255):
raise RemoteException(_(
"Non-zero return code ({rcode}) running '{command}' on '{host}':\n\n{result}"
).format(
command=command,
host=hostname,
rcode=result.return_code,
result=force_text(result.stdout) + force_text(result.stderr),
))
return result