7017582421

Committed 28 Nov 2023 10:42AM UTC coverage: 91.699% (+4.4%) from 87.286%

Build # 7017582421

Build Type

push

github

Committed by

uwefladrich

Commit Message

Fix handling of stderr in case command fails

Run Details

15 of 15 new or added lines in 2 files covered. (100.0%)

21 existing lines in 4 files now uncovered.

1900 of 2072 relevant lines covered (91.7%)

0.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.12

/src/scriptengine/tasks/base/include.py

"""Include task for ScriptEngine

   The Include task reads a YAML file, parses the content to create a
   ScriptEngine script and lets the active ScriptEngine instance execute it.
"""

import copy
from pathlib import Path

from scriptengine.context import Context
from scriptengine.exceptions import ScriptEngineTaskRunError
from scriptengine.tasks.core import Task, timed_runner
from scriptengine.yaml.parser import parse_file


def _file_in_paths(file, paths=None):
    if file.is_absolute():
        if file.is_file():
            return file
        raise FileNotFoundError

    searched_paths = []
    for p in paths or []:
        if p not in searched_paths:
            searched_paths.append(p)
            fpath = p / file
            if fpath.is_file():
                return fpath
    raise FileNotFoundError


class Include(Task):

    _required_arguments = ("src",)

    def __init__(self, arguments):
        Include.check_arguments(arguments)
        super().__init__(arguments)

    def __str__(self):
        return f"Include: {self.src}"

    @timed_runner
    def run(self, context):

        src = Path(self.getarg("src", context))
        self.log_info(f"Include script from {src}")

        # Search for include file in '.', old cwd, and script_path
        cwd = Path(".")
        ocwd = Path(context["se"]["cli"]["cwd"])
        script_path = (Path(p) for p in context["se"]["cli"]["script_path"])
        search_path = (cwd, ocwd, *script_path)
        self.log_debug(
            f"Include file search path: {tuple(str(p) for p in search_path)}"
        )

        try:
            inc_file = _file_in_paths(src, search_path)
        except FileNotFoundError:
            if self.getarg("ignore_not_found", default=False):
                self.log_warning(f"Include file not found: {src}")
                return
            self.log_error(f"Include file not found: {src}")
            raise ScriptEngineTaskRunError
        self.log_debug(f"Include file found: {str(inc_file)}")

        script = parse_file(inc_file)

        self.log_debug(f"Execute include script: {inc_file}")
        local_context = Context(copy.deepcopy(context))
        context_update = local_context["se"]["instance"].run(script, local_context)
        self.log_debug(f"Finished executing include script: {inc_file}")

        return context_update or None

1	"""Include task for ScriptEngine	1✔
2
3	The Include task reads a YAML file, parses the content to create a
4	ScriptEngine script and lets the active ScriptEngine instance execute it.
5	"""
6
7	import copy	1✔
8	from pathlib import Path	1✔
9
10	from scriptengine.context import Context	1✔
11	from scriptengine.exceptions import ScriptEngineTaskRunError	1✔
12	from scriptengine.tasks.core import Task, timed_runner	1✔
13	from scriptengine.yaml.parser import parse_file	1✔
14
15
16	def _file_in_paths(file, paths=None):	1✔
17	if file.is_absolute():	1✔
18	if file.is_file():	×
19	return file	×
UNCOV 20	raise FileNotFoundError	×
21
22	searched_paths = []	1✔
23	for p in paths or []:	1✔
24	if p not in searched_paths:	1✔
25	searched_paths.append(p)	1✔
26	fpath = p / file	1✔
27	if fpath.is_file():	1✔
28	return fpath	1✔
29	raise FileNotFoundError	1✔
30
31
32	class Include(Task):	1✔
33
34	_required_arguments = ("src",)	1✔
35
36	def __init__(self, arguments):	1✔
37	Include.check_arguments(arguments)	1✔
38	super().__init__(arguments)	1✔
39
40	def __str__(self):	1✔
41	return f"Include: {self.src}"	1✔
42
43	@timed_runner	1✔
44	def run(self, context):	1✔
45
46	src = Path(self.getarg("src", context))	1✔
47	self.log_info(f"Include script from {src}")	1✔
48
49	# Search for include file in '.', old cwd, and script_path
50	cwd = Path(".")	1✔
51	ocwd = Path(context["se"]["cli"]["cwd"])	1✔
52	script_path = (Path(p) for p in context["se"]["cli"]["script_path"])	1✔
53	search_path = (cwd, ocwd, *script_path)	1✔
54	self.log_debug(	1✔
55	f"Include file search path: {tuple(str(p) for p in search_path)}"
56	)
57
58	try:	1✔
59	inc_file = _file_in_paths(src, search_path)	1✔
60	except FileNotFoundError:	1✔
61	if self.getarg("ignore_not_found", default=False):	1✔
62	self.log_warning(f"Include file not found: {src}")	1✔
63	return	1✔
64	self.log_error(f"Include file not found: {src}")	1✔
65	raise ScriptEngineTaskRunError	1✔
66	self.log_debug(f"Include file found: {str(inc_file)}")	1✔
67
68	script = parse_file(inc_file)	1✔
69
70	self.log_debug(f"Execute include script: {inc_file}")	1✔
71	local_context = Context(copy.deepcopy(context))	1✔
72	context_update = local_context["se"]["instance"].run(script, local_context)	1✔
73	self.log_debug(f"Finished executing include script: {inc_file}")	1✔
74
75	return context_update or None	1✔

uwefladrich / scriptengine / 7017582421

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous