7021324889

Committed 28 Nov 2023 03:55PM UTC coverage: 91.371% (-0.3%) from 91.671%

Build # 7021324889

Build Type

Pull #100

github

Committed by

uwefladrich

Commit Message

Update CHANGES

Pull Request Pull Request #100: Better Jinja2 error messages

Run Details

23 of 38 new or added lines in 4 files covered. (60.53%)

1 existing line in 1 file now uncovered.

1906 of 2086 relevant lines covered (91.37%)

0.91 hits per line

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

89.01

/src/scriptengine/tasks/core/task.py

"""ScriptEngine task module.

Provides the base class for all tasks.
"""

import logging
import uuid

import yaml

import scriptengine.jinja
import scriptengine.yaml
from scriptengine.exceptions import (
    ScriptEngineParseJinjaError,
    ScriptEngineTaskArgumentInvalidError,
    ScriptEngineTaskArgumentMissingError,
)
from scriptengine.yaml.noparse_strings import NoParseJinjaString, NoParseYamlString

_SENTINEL = object()


class Task:
    _reg_name = None
    _invalid_arguments = (
        "run",
        "id",
    )

    @classmethod
    def check_arguments(cls, arguments):
        """Checks arguments against class members '_invalid_arguments' and
        '_required_arguments', if present. Returns None or throws either
        ScriptEngineTaskArgumentInvalidError or
        ScriptEngineTaskArgumentMissingError.
        """
        for name in arguments:
            if name in getattr(cls, "_invalid_arguments", ()):
                logging.getLogger("se.task").error(
                    (
                        f'Invalid argument "{name}" found while '
                        f'trying to create "{cls.__name__}" task'
                    ),
                    extra={"id": "no id", "type": cls.__name__},
                )
                raise ScriptEngineTaskArgumentInvalidError

        for name in getattr(cls, "_required_arguments", ()):
            if name not in arguments:
                logging.getLogger("se.task").error(
                    (
                        f'Missing required argument "{name}" while '
                        f'trying to create "{cls.__name__}" task'
                    ),
                    extra={"id": "no id", "type": cls.__name__},
                )
                raise ScriptEngineTaskArgumentMissingError

    def __init__(self, arguments=None):
        self._identifier = uuid.uuid4()

        if arguments is not None:
            Task.check_arguments(arguments)
            for name, value in arguments.items():
                if hasattr(self, name):
                    self.log_error(f"Invalid (reserved name) task argument: {name}")
                    raise ScriptEngineTaskArgumentInvalidError
                setattr(self, name, value)
        self.log_debug(f"Created task: {self}")

    @classmethod
    def register_name(cls, name):
        cls._reg_name = name

    @property
    def reg_name(self):
        return (
            self._reg_name or f"{self.__class__.__module__}:{self.__class__.__name__}"
        )

    @property
    def id(self):
        return self._identifier

    @property
    def shortid(self):
        return self._identifier.hex[:10]

    def __repr__(self):
        params = {
            key: val
            for key, val in self.__dict__.items()
            if not (isinstance(key, str) and key.startswith("_"))
        }
        params_list = f'{", ".join([f"{k}={params[k]}" for k in params])}'
        return f"{self.__class__.__name__}({params_list})"

    def run(self, context):
        raise NotImplementedError("Base class function Task.run() must not be called")

    def getarg(
        self, name, context={}, *, parse_jinja=True, parse_yaml=True, default=_SENTINEL
    ):
        """Returns the value of argument 'name'.
        The argument value is parsed with Jinja2 and the given context,
        unless 'parse_jinja' is False. The result is parsed once more with
        the YAML parser in order to get a correctly typed result. If
        parse_yaml is not True, the extra YAML parsing is skipped.
        Parsing with Jinja/YAML is also skipped, if the argument value is a
        string that starts with '_noparse_', '_noparsejinja_',
        '_noparseyaml_', respectively.
        If argument 'name' does not exist, the function raises an
        AttributeError, unless a 'default' value is given.
        """

        def parse(arg_):
            # Recursively parse list items
            if isinstance(arg_, list):
                return [parse(item) for item in arg_]

            # Recursively parse dict values (not keys!)
            if isinstance(arg_, dict):
                return dict((key, parse(val)) for key, val in arg_.items())

            if isinstance(arg_, str):
                if parse_jinja and not isinstance(arg_, NoParseJinjaString):
                    try:
                        # Make sure that a NoParseString is still a NoParseString
                        # after this!
                        arg_ = type(arg_)(scriptengine.jinja.render(arg_, context))
                    except ScriptEngineParseJinjaError as e:
                        self.log_error(e)
                        raise ScriptEngineTaskArgumentInvalidError

                if parse_yaml and not isinstance(arg_, NoParseYamlString):
                    try:
                        return yaml.full_load(arg_)
                    except (
                        yaml.scanner.ScannerError,
                        yaml.parser.ParserError,
                        yaml.constructor.ConstructorError,
                    ):
                        self.log_debug(f'Reparsing argument "{arg_}" with YAML failed')

                # Return plain strings, not NoParse*Strings
                return str(arg_)

            # If not a list, dict or string, just return
            return arg_

        try:
            arg = getattr(self, name)
        except AttributeError:
            if default is _SENTINEL:
                self.log_error(f"Trying to access missing task argument: {name}")
                raise ScriptEngineTaskArgumentMissingError
            arg = default
        return parse(arg)

    def _log(self, level, msg):
        logger = logging.getLogger("se.task")
        logger.log(level, msg, extra={"type": self.reg_name, "id": self.shortid})

    def log_debug(self, msg):
        self._log(logging.DEBUG, msg)

    def log_info(self, msg):
        self._log(logging.INFO, msg)

    def log_warning(self, msg):
        self._log(logging.WARNING, msg)

    def log_error(self, msg):
        self._log(logging.ERROR, msg)

    def log_critical(self, msg):
        self._log(logging.CRITICAL, msg)

1	"""ScriptEngine task module.	1✔
2
3	Provides the base class for all tasks.
4	"""
5
6	import logging	1✔
7	import uuid	1✔
8
9	import yaml	1✔
10
11	import scriptengine.jinja	1✔
12	import scriptengine.yaml	1✔
13	from scriptengine.exceptions import (	1✔
14	ScriptEngineParseJinjaError,
15	ScriptEngineTaskArgumentInvalidError,
16	ScriptEngineTaskArgumentMissingError,
17	)
18	from scriptengine.yaml.noparse_strings import NoParseJinjaString, NoParseYamlString	1✔
19
20	_SENTINEL = object()	1✔
21
22
23	class Task:	1✔
24	_reg_name = None	1✔
25	_invalid_arguments = (	1✔
26	"run",
27	"id",
28	)
29
30	@classmethod	1✔
31	def check_arguments(cls, arguments):	1✔
32	"""Checks arguments against class members '_invalid_arguments' and
33	'_required_arguments', if present. Returns None or throws either
34	ScriptEngineTaskArgumentInvalidError or
35	ScriptEngineTaskArgumentMissingError.
36	"""
37	for name in arguments:	1✔
38	if name in getattr(cls, "_invalid_arguments", ()):	1✔
39	logging.getLogger("se.task").error(	1✔
40	(
41	f'Invalid argument "{name}" found while '
42	f'trying to create "{cls.__name__}" task'
43	),
44	extra={"id": "no id", "type": cls.__name__},
45	)
46	raise ScriptEngineTaskArgumentInvalidError	1✔
47
48	for name in getattr(cls, "_required_arguments", ()):	1✔
49	if name not in arguments:	1✔
50	logging.getLogger("se.task").error(	1✔
51	(
52	f'Missing required argument "{name}" while '
53	f'trying to create "{cls.__name__}" task'
54	),
55	extra={"id": "no id", "type": cls.__name__},
56	)
57	raise ScriptEngineTaskArgumentMissingError	1✔
58
59	def __init__(self, arguments=None):	1✔
60	self._identifier = uuid.uuid4()	1✔
61
62	if arguments is not None:	1✔
63	Task.check_arguments(arguments)	1✔
64	for name, value in arguments.items():	1✔
65	if hasattr(self, name):	1✔
NEW 66	self.log_error(f"Invalid (reserved name) task argument: {name}")	×
UNCOV 67	raise ScriptEngineTaskArgumentInvalidError	×
68	setattr(self, name, value)	1✔
69	self.log_debug(f"Created task: {self}")	1✔
70
71	@classmethod	1✔
72	def register_name(cls, name):	1✔
73	cls._reg_name = name	1✔
74
75	@property	1✔
76	def reg_name(self):	1✔
77	return (	1✔
78	self._reg_name or f"{self.__class__.__module__}:{self.__class__.__name__}"
79	)
80
81	@property	1✔
82	def id(self):	1✔
83	return self._identifier	1✔
84
85	@property	1✔
86	def shortid(self):	1✔
87	return self._identifier.hex[:10]	1✔
88
89	def __repr__(self):	1✔
90	params = {	1✔
91	key: val
92	for key, val in self.__dict__.items()
93	if not (isinstance(key, str) and key.startswith("_"))
94	}
95	params_list = f'{", ".join([f"{k}={params[k]}" for k in params])}'	1✔
96	return f"{self.__class__.__name__}({params_list})"	1✔
97
98	def run(self, context):	1✔
NEW 99	raise NotImplementedError("Base class function Task.run() must not be called")	×
100
101	def getarg(	1✔
102	self, name, context={}, *, parse_jinja=True, parse_yaml=True, default=_SENTINEL
103	):
104	"""Returns the value of argument 'name'.
105	The argument value is parsed with Jinja2 and the given context,
106	unless 'parse_jinja' is False. The result is parsed once more with
107	the YAML parser in order to get a correctly typed result. If
108	parse_yaml is not True, the extra YAML parsing is skipped.
109	Parsing with Jinja/YAML is also skipped, if the argument value is a
110	string that starts with '_noparse_', '_noparsejinja_',
111	'_noparseyaml_', respectively.
112	If argument 'name' does not exist, the function raises an
113	AttributeError, unless a 'default' value is given.
114	"""
115
116	def parse(arg_):	1✔
117	# Recursively parse list items
118	if isinstance(arg_, list):	1✔
119	return [parse(item) for item in arg_]	1✔
120
121	# Recursively parse dict values (not keys!)
122	if isinstance(arg_, dict):	1✔
123	return dict((key, parse(val)) for key, val in arg_.items())	1✔
124
125	if isinstance(arg_, str):	1✔
126	if parse_jinja and not isinstance(arg_, NoParseJinjaString):	1✔
127	try:	1✔
128	# Make sure that a NoParseString is still a NoParseString
129	# after this!
130	arg_ = type(arg_)(scriptengine.jinja.render(arg_, context))	1✔
NEW 131	except ScriptEngineParseJinjaError as e:	×
NEW 132	self.log_error(e)	×
NEW 133	raise ScriptEngineTaskArgumentInvalidError	×
134
135	if parse_yaml and not isinstance(arg_, NoParseYamlString):	1✔
136	try:	1✔
137	return yaml.full_load(arg_)	1✔
138	except (	×
139	yaml.scanner.ScannerError,
140	yaml.parser.ParserError,
141	yaml.constructor.ConstructorError,
142	):
NEW 143	self.log_debug(f'Reparsing argument "{arg_}" with YAML failed')	×
144
145	# Return plain strings, not NoParse*Strings
146	return str(arg_)	×
147
148	# If not a list, dict or string, just return
149	return arg_	1✔
150
151	try:	1✔
152	arg = getattr(self, name)	1✔
153	except AttributeError:	1✔
154	if default is _SENTINEL:	1✔
155	self.log_error(f"Trying to access missing task argument: {name}")	1✔
156	raise ScriptEngineTaskArgumentMissingError	1✔
157	arg = default	1✔
158	return parse(arg)	1✔
159
160	def _log(self, level, msg):	1✔
161	logger = logging.getLogger("se.task")	1✔
162	logger.log(level, msg, extra={"type": self.reg_name, "id": self.shortid})	1✔
163
164	def log_debug(self, msg):	1✔
165	self._log(logging.DEBUG, msg)	1✔
166
167	def log_info(self, msg):	1✔
168	self._log(logging.INFO, msg)	1✔
169
170	def log_warning(self, msg):	1✔
171	self._log(logging.WARNING, msg)	1✔
172
173	def log_error(self, msg):	1✔
174	self._log(logging.ERROR, msg)	1✔
175
176	def log_critical(self, msg):	1✔
177	self._log(logging.CRITICAL, msg)	×

uwefladrich / scriptengine / 7021324889

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