Commit 53b5dbea authored by Wout De Nolf's avatar Wout De Nolf
Browse files

rearrange tests

parent 9260caf5
from esrftaskgraph.utils import qualname
from esrftaskgraph.task import Task
def mymethod1(a=0, b=0):
return a + b
def test_method_task1(varinfo):
task = Task.instantiate(
"MethodExecutorTask",
inputs={"method": qualname(mymethod1), "a": 3, "b": 5},
varinfo=varinfo,
)
task.execute()
assert task.done
assert task.output_values == {"return_value": 8}
def mymethod2(*args):
return sum(args)
def test_method_task2(varinfo):
task = Task.instantiate(
"MethodExecutorTask",
inputs={"method": qualname(mymethod2), 0: 3, 1: 5},
varinfo=varinfo,
)
task.execute()
assert task.done
assert task.output_values == {"return_value": 8}
def mymethod3(a, *args, b=None, c=3, **kw):
print("a:", a)
print("b:", b)
print("c:", c)
print("args:", args)
print("kwargs:", kw)
return a + sum(args) + b + c + sum(kw.values())
def test_method_task3(varinfo):
task = Task.instantiate(
"MethodExecutorTask",
inputs={"method": qualname(mymethod3), 0: 2, 1: 4, "b": 7, "d": 10},
varinfo=varinfo,
)
task.execute()
assert task.done
assert task.output_values == {"return_value": 26}
def myppfmethod(a=0, b=0, **kw):
return {"a": a + b}
def test_ppfmethod_task(varinfo):
task = Task.instantiate(
"PpfMethodExecutorTask",
inputs={"method": qualname(myppfmethod), "a": 3, "b": 5},
varinfo=varinfo,
)
task.execute()
assert task.done
assert task.output_values == {"ppfdict": {"a": 8, "b": 5}}
import os
from esrftaskgraph.task import Task
pyscript = r"""
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--a", type=int, default=0)
args = parser.parse_args()
print("input a =", args.a)
assert args.a == 10
"""
def test_python_script_task(tmpdir, varinfo, capsys):
pyscriptname = tmpdir / "test.py"
with open(pyscriptname, mode="w") as f:
f.writelines(pyscript)
task = Task.instantiate(
"ScriptExecutorTask",
inputs={"a": 10, "script": str(pyscriptname)},
varinfo=varinfo,
)
task.execute()
assert task.done
assert task.outputs.returncode == 0
captured = capsys.readouterr()
# assert captured.out == "10\n"
assert captured.err == ""
shellscript = r"""a=0
while getopts u:a:f: flag
do
case "${flag}" in
a) a=${OPTARG};;
esac
done
echo "input a = "$a
if [[ $a == "10" ]]; then
exit 0
else
exit 1
fi
"""
def test_shell_script_task(tmpdir, varinfo, capsys):
shellscriptname = tmpdir / "test.sh"
with open(shellscriptname, mode="w") as f:
f.writelines(shellscript)
os.chmod(shellscriptname, 0o755)
task = Task.instantiate(
"ScriptExecutorTask",
inputs={"a": 10, "script": str(shellscriptname)},
varinfo=varinfo,
)
task.execute()
assert task.done
assert task.outputs.returncode == 0
captured = capsys.readouterr()
# assert captured.out == "10\n"
assert captured.err == ""
import os
import pytest
import json
from esrftaskgraph.utils import qualname
from esrftaskgraph.task import Task
from tasklib.tasks import SumTask
def mymethod1(a=0, b=0):
return a + b
def mymethod2(*args):
return sum(args)
def myppfmethod(a=0, b=0, **kw):
return {"a": a + b}
def assert_storage(tmpdir, expected):
lst = []
for fileobj in tmpdir.listdir():
......@@ -125,99 +111,3 @@ def test_task_storage(tmpdir, varinfo):
assert task.outputs.result == 13
expected += [{"result": str(task.output_variables["result"].uhash)}, 13]
assert_storage(tmpdir, expected)
def test_method_task(varinfo):
task = Task.instantiate(
"MethodExecutorTask",
inputs={"method": qualname(mymethod1), "a": 3, "b": 5},
varinfo=varinfo,
)
task.execute()
assert task.done
assert task.output_values == {"return_value": 8}
task = Task.instantiate(
"MethodExecutorTask",
inputs={"method": qualname(mymethod2), 0: 3, 1: 5},
varinfo=varinfo,
)
task.execute()
assert task.done
assert task.output_values == {"return_value": 8}
def test_ppfmethod_task(varinfo):
task = Task.instantiate(
"PpfMethodExecutorTask",
inputs={"method": qualname(myppfmethod), "a": 3, "b": 5},
varinfo=varinfo,
)
task.execute()
assert task.done
assert task.output_values == {"ppfdict": {"a": 8, "b": 5}}
shellscript = r"""a=0
while getopts u:a:f: flag
do
case "${flag}" in
a) a=${OPTARG};;
esac
done
echo "input a = "$a
if [[ $a == "10" ]]; then
exit 0
else
exit 1
fi
"""
pyscript = r"""
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--a", type=int, default=0)
args = parser.parse_args()
print("input a =", args.a)
assert args.a == 10
"""
def test_python_script_task(tmpdir, varinfo, capsys):
pyscriptname = tmpdir / "test.py"
with open(pyscriptname, mode="w") as f:
f.writelines(pyscript)
task = Task.instantiate(
"ScriptExecutorTask",
inputs={"a": 10, "script": str(pyscriptname)},
varinfo=varinfo,
)
task.execute()
assert task.done
assert task.outputs.returncode == 0
captured = capsys.readouterr()
# assert captured.out == "10\n"
assert captured.err == ""
def test_shell_script_task(tmpdir, varinfo, capsys):
shellscriptname = tmpdir / "test.sh"
with open(shellscriptname, mode="w") as f:
f.writelines(shellscript)
os.chmod(shellscriptname, 0o755)
task = Task.instantiate(
"ScriptExecutorTask",
inputs={"a": 10, "script": str(shellscriptname)},
varinfo=varinfo,
)
task.execute()
assert task.done
assert task.outputs.returncode == 0
captured = capsys.readouterr()
# assert captured.out == "10\n"
assert captured.err == ""
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment