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
|
# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE.
# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""Tests involving the Cylc Subprocess Context Object
"""
from logging import DEBUG
from textwrap import dedent
async def test_log_xtrigger_stdout(
flow, scheduler, run_dir, start, log_filter
):
"""Output from xtriggers should appear in the scheduler log:
(As per the toy example in the Cylc Docs)
"""
# Setup a workflow:
id_ = flow({
'scheduler': {'allow implicit tasks': True},
'scheduling': {
'graph': {'R1': '@myxtrigger => foo'},
'xtriggers': {'myxtrigger': 'myxtrigger()'}
}
})
# Create an xtrigger:
xt_lib = run_dir / id_ / 'lib/python/myxtrigger.py'
xt_lib.parent.mkdir(parents=True, exist_ok=True)
xt_lib.write_text(dedent(r"""
from sys import stderr
def myxtrigger():
print('Hello World')
print('Hello Hades', file=stderr)
return True, {}
"""))
schd = scheduler(id_)
async with start(schd, level=DEBUG):
# Set off check for x-trigger:
task = schd.pool.get_tasks()[0]
schd.xtrigger_mgr.call_xtriggers_async(task)
# while not schd.xtrigger_mgr._get_xtrigs(task):
while schd.proc_pool.is_not_done():
schd.proc_pool.process()
# Assert that both stderr and out from the print statement
# in our xtrigger appear in the log.
for expected in ['Hello World', 'Hello Hades']:
assert log_filter(DEBUG, expected)
|
