Newer
Older
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
from loomenv import loom_env, LOOM_TESTPROG, LOOM_TEST_DATA_DIR # noqa
import loom.client.tasks as tasks # noqa
from loom.client import TaskFailed
import pytest
loom_env # silence flake8
def test_py_call(loom_env):
def f(a, b):
return "{}, {}, {}, {}".format(
str(a.read(), encoding="ascii"), a.size(),
str(b.read(), encoding="ascii"), b.size())
def g():
return "Test"
loom_env.start(1)
c = tasks.const("ABC")
d = tasks.const("12345")
p = tasks.py_call(f, (c, d))
q = tasks.py_call(g)
result1, result2 = loom_env.submit((p, q))
assert result1 == b"ABC, 3, 12345, 5"
assert result2 == b"Test"
def test_py_fail_too_many_args(loom_env):
def g():
return "Test"
loom_env.start(1)
c = tasks.const("ABC")
a = tasks.py_call(g, (c,))
with pytest.raises(TaskFailed):
loom_env.submit(a)
def test_py_fail_too_few_args(loom_env):
def f(a):
return "ABC"
loom_env.start(1)
a = tasks.py_call(f, ())
with pytest.raises(TaskFailed):
loom_env.submit(a)
def test_py_fail_invalid_result(loom_env):
def f():
return 42.0
loom_env.start(1)
a = tasks.py_call(f, ())
with pytest.raises(TaskFailed):
loom_env.submit(a)