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
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
|
"""Tests monitoring records app name under various decoration patterns.
"""
import os
import time
import pytest
import parsl
from parsl.tests.configs.htex_local_alternate import fresh_config
@parsl.python_app
def regular_decorated_app():
return 5
@pytest.mark.local
def get_regular_decorated_app():
return regular_decorated_app
def for_decoration_later():
return 77
def get_for_decoration_later():
return parsl.python_app(for_decoration_later)
def get_decorated_closure():
r = 53
@parsl.python_app
def decorated_closure():
return r
return decorated_closure
@pytest.mark.local
@pytest.mark.parametrize("get_app,expected_name,expected_result",
[(get_regular_decorated_app, "regular_decorated_app", 5),
(get_for_decoration_later, "for_decoration_later", 77),
(get_decorated_closure, "decorated_closure", 53)
])
def test_app_name(get_app, expected_name, expected_result, tmpd_cwd):
# this is imported here rather than at module level because
# it isn't available in a plain parsl install, so this module
# would otherwise fail to import and break even a basic test
# run.
import sqlalchemy
c = fresh_config()
c.run_dir = tmpd_cwd
c.monitoring.logging_endpoint = f"sqlite:///{tmpd_cwd}/monitoring.db"
parsl.load(c)
app = get_app()
assert app().result() == expected_result
parsl.dfk().cleanup()
engine = sqlalchemy.create_engine(c.monitoring.logging_endpoint)
with engine.begin() as connection:
def count_rows(table: str):
result = connection.execute(sqlalchemy.text(f"SELECT COUNT(*) FROM {table}"))
(c, ) = result.first()
return c
# one workflow...
assert count_rows("workflow") == 1
# ... with one task ...
assert count_rows("task") == 1
# ... that was tried once ...
assert count_rows("try") == 1
# ... and has the expected name.
result = connection.execute(sqlalchemy.text("SELECT task_func_name FROM task"))
(c, ) = result.first()
assert c == expected_name
|
