111 lines
3.8 KiB
Python
111 lines
3.8 KiB
Python
import sys
|
|
import os.path
|
|
import tempfile
|
|
from unittest import mock
|
|
from subprocess import CompletedProcess
|
|
|
|
from jeeves.core.objects import Task
|
|
|
|
MOCK_DEFINITION = {
|
|
"type": f"jeeves.core.actions.shell:ScriptAction",
|
|
"name": "Say hello world in bash",
|
|
"parameters": {"script": "#!/bin/bash\necho Hello World"},
|
|
}
|
|
|
|
|
|
def get_completed_process(returncode=0, stdout=b"", **kwargs):
|
|
"""Mocks a :class:`subprocess.CompletedProcess` object"""
|
|
return CompletedProcess("", returncode=returncode, stdout=stdout, **kwargs)
|
|
|
|
|
|
@mock.patch("subprocess.run", mock.MagicMock(return_value=get_completed_process()))
|
|
def test_script_bash_task_ok():
|
|
task = Task.parse_obj(MOCK_DEFINITION).action
|
|
result = task.execute()
|
|
assert result.success
|
|
|
|
|
|
@mock.patch(
|
|
"subprocess.run", mock.MagicMock(return_value=get_completed_process(returncode=1))
|
|
)
|
|
def test_script_bash_task_ko():
|
|
task = Task.parse_obj(MOCK_DEFINITION).action
|
|
result = task.execute()
|
|
assert not result.success
|
|
|
|
|
|
@mock.patch("subprocess.run", mock.MagicMock(return_value=get_completed_process()))
|
|
def test_script_no_shebang_defaults_to_bash_ok():
|
|
definition = MOCK_DEFINITION.copy()
|
|
definition["parameters"]["script"] = definition["parameters"]["script"].strip(
|
|
"#!/bin/bash"
|
|
)
|
|
task = Task.parse_obj(definition).action
|
|
assert task._get_script().startswith(task.DEFAULT_SHEBANG)
|
|
result = task.execute()
|
|
assert result.success
|
|
|
|
|
|
@mock.patch("subprocess.run", mock.MagicMock(return_value=get_completed_process()))
|
|
def test_script_with_other_shebang_ok():
|
|
py_interpreter = sys.executable
|
|
expected_output = "Hello world! (from python)"
|
|
definition = MOCK_DEFINITION.copy()
|
|
py_script = f"#!{py_interpreter}\nprint('{expected_output}')"
|
|
definition["parameters"]["script"] = py_script
|
|
task = Task.parse_obj(definition).action
|
|
assert task._get_script().startswith(f"#!{py_interpreter}")
|
|
result = task.execute()
|
|
assert result.success
|
|
|
|
|
|
def test_script_stdout_and_stderr_is_sent_to_result_ok():
|
|
"""
|
|
..warning:: This test actually calls ``subprocess.run``.
|
|
|
|
Not 100% sure this one is needed since we are just testing that subprocess.STDOUT works.
|
|
I'm leaving it for now since it's important to ensure we get the entire stdout/err in the
|
|
:any:`Result` object.
|
|
"""
|
|
script = "\n".join(
|
|
[
|
|
f"#!{sys.executable}",
|
|
"import sys",
|
|
"sys.stdout.write('Hello')",
|
|
"sys.stderr.write('World')",
|
|
]
|
|
)
|
|
definition = MOCK_DEFINITION.copy()
|
|
definition["parameters"]["script"] = script
|
|
task = Task.parse_obj(definition).action
|
|
result = task.execute()
|
|
assert "Hello" in result.output
|
|
assert "World" in result.output
|
|
|
|
|
|
@mock.patch("subprocess.run", mock.MagicMock(return_value=get_completed_process()))
|
|
def test_script_task_cleans_tempfile_ok():
|
|
"""Make sure that the script is removed from the system after execution"""
|
|
task = Task.parse_obj(MOCK_DEFINITION).action
|
|
temp = tempfile.NamedTemporaryFile(mode="w", delete=False)
|
|
with mock.patch(
|
|
"tempfile.NamedTemporaryFile", mock.MagicMock(return_value=temp)
|
|
) as mocked:
|
|
task.execute()
|
|
assert not os.path.isfile(mocked.return_value.name)
|
|
|
|
|
|
@mock.patch("subprocess.run", mock.MagicMock(return_value=get_completed_process()))
|
|
def test_script_task_sets_permissions_for_owner_only_ok():
|
|
"""Make sure that the script have only read and execution permissions for owner"""
|
|
task = Task.parse_obj(MOCK_DEFINITION).action
|
|
temp = tempfile.NamedTemporaryFile(mode="w", delete=False)
|
|
with mock.patch(
|
|
"tempfile.NamedTemporaryFile", mock.MagicMock(return_value=temp)
|
|
) as mocked:
|
|
with mock.patch("os.unlink"):
|
|
task.execute()
|
|
stat = os.stat(mocked.return_value.name)
|
|
assert oct(stat.st_mode).endswith("500")
|
|
os.unlink(mocked.return_value.name)
|