28270112145

Committed 26 Jun 2026 11:03PM UTC coverage: 95.859% (-0.005%) from 95.864%

Build # 28270112145

Build Type

Pull #895

github

Committed by

web-flow

Commit Message

Merge 030a1b5a2 into dc6787fd9

Pull Request Pull Request #895: Fix input path mounting on retry

Coverage Stats

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

21 existing lines in 4 files now uncovered.

8797 of 9177 relevant lines covered (95.86%)

0.96 hits per line

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

91.1

/WDL/Error.py

from typing import (
    List,
    Optional,
    Union,
    Iterable,
    Generator,
    Callable,
    Any,
    Dict,
    NamedTuple,
)
import os
from functools import total_ordering
from contextlib import contextmanager

from . import Type


class SourcePosition(
    NamedTuple(
        "SourcePosition",
        [
            ("uri", str),
            ("abspath", str),
            ("line", int),
            ("column", int),
            ("end_line", int),
            ("end_column", int),
        ],
    )
):
    """
    Source position attached to AST nodes and exceptions; NamedTuple of ``uri`` the filename/URI
    passed to :func:`WDL.load` or a WDL import statement, which may be relative; ``abspath`` the
    absolute filename/URI; and one-based int positions ``line`` ``end_line`` ``column``
    ``end_column``
    """


class SyntaxError(Exception):
    """Failure to lex/parse a WDL document"""

    pos: SourcePosition
    wdl_version: str
    declared_wdl_version: Optional[str]

    def __init__(
        self, pos: SourcePosition, msg: str, wdl_version: str, declared_wdl_version: Optional[str]
    ) -> None:
        super().__init__(msg)
        self.pos = pos
        self.wdl_version = wdl_version
        self.declared_wdl_version = declared_wdl_version


class _BadCharacterEncoding(Exception):
    """"""

    # Invalid escape sequence in a string literal; this is used internally, eventually resurfaced
    # as a SyntaxError.
    pos: SourcePosition

    def __init__(self, pos: SourcePosition):
        self.pos = pos


class ImportError(Exception):
    """Failure to open/retrieve an imported WDL document

    The ``__cause__`` attribute may hold the inner error object."""

    pos: SourcePosition

    def __init__(self, pos: SourcePosition, import_uri: str, message: Optional[str] = None) -> None:
        msg = "Failed to import " + import_uri
        if message:
            msg = msg + ", " + message
        super().__init__(msg)
        self.pos = pos


@total_ordering
class SourceNode:
    """Base class for an AST node, recording the source position"""

    pos: SourcePosition
    """
    :type: SourcePosition

    Source position for this AST node
    """

    parent: Optional["SourceNode"] = None
    """
    :type: Optional[SourceNode]

    Parent node in the AST, if any
    """

    def __init__(self, pos: SourcePosition) -> None:
        self.pos = pos

    @property
    def source_dir(self) -> str:
        """
        Local directory containing this source node, with trailing "/", or "".

        Source-relative File/Directory paths need an explicit local source directory. Parsed
        buffers, remote URIs, and other non-local source locations are represented as an empty
        string so callers can produce context-specific errors only when relative path resolution is
        actually needed.

        The directory is normalized but symlinks are intentionally preserved (abspath, not
        realpath): if the user reached the WDL through a symlinked path, that's presumed deliberate
        and is kept in resolved File/Directory values. Symlink resolution is applied only where it's
        a security boundary (see ``path_really_within`` usage in source-relative path resolution).
        """
        source = self.pos.abspath
        if not source or source == "(buffer)" or not os.path.isabs(source):
            return ""
        return os.path.join(os.path.abspath(os.path.dirname(source)), "")

    def __lt__(self, rhs: "SourceNode") -> bool:
        return isinstance(rhs, SourceNode) and (
            self.pos.abspath,
            self.pos.line,
            self.pos.column,
            self.pos.end_line,
            self.pos.end_column,
        ) < (
            rhs.pos.abspath,
            rhs.pos.line,
            rhs.pos.column,
            rhs.pos.end_line,
            rhs.pos.end_column,
        )

    def __eq__(self, rhs: Any) -> bool:
        return isinstance(rhs, SourceNode) and self.pos == rhs.pos

    @property
    def children(self) -> Iterable["SourceNode"]:
        """
        :type: Iterable[SourceNode]

        Yield all child nodes
        """
        return []


class ValidationError(Exception):
    """
    Base class for a WDL validation error (when the document loads and parses, but fails typechecking or other static
    validity tests)
    """

    pos: SourcePosition
    """:type: SourcePosition"""

    node: Optional[SourceNode] = None
    """:type: Optional[SourceNode]"""

    source_text: Optional[str] = None
    """:type: Optional[str]

    The complete source text of the WDL document (if available)"""

    declared_wdl_version: Optional[str] = None

    def __init__(self, node: Union[SourceNode, SourcePosition], message: str) -> None:
        if isinstance(node, SourceNode):
            self.node = node
            self.pos = node.pos
        else:
            self.pos = node
        super().__init__(message)


class InvalidType(ValidationError):
    pass


class IndeterminateType(ValidationError):
    pass


class NoSuchTask(ValidationError):
    def __init__(self, node: Union[SourceNode, SourcePosition], name: str) -> None:
        super().__init__(node, "No such task/workflow: " + name)


class NoSuchCall(ValidationError):
    def __init__(self, node: Union[SourceNode, SourcePosition], name: str) -> None:
        super().__init__(node, "No such call in this workflow: " + name)


class NoSuchFunction(ValidationError):
    def __init__(self, node: Union[SourceNode, SourcePosition], name: str) -> None:
        super().__init__(node, "No such function: " + name)


class WrongArity(ValidationError):
    def __init__(self, node: Union[SourceNode, SourcePosition], expected: int) -> None:
        # avoiding circular dep:
        # assert isinstance(node, WDL.Expr.Apply)
        msg = "{} expects {} argument(s)".format(getattr(node, "function_name"), expected)
        super().__init__(node, msg)


class NotAnArray(ValidationError):
    def __init__(self, node: Union[SourceNode, SourcePosition]) -> None:
        super().__init__(node, "Not an array")


class NoSuchMember(ValidationError):
    def __init__(self, node: Union[SourceNode, SourcePosition], member: str) -> None:
        super().__init__(node, "No such member '{}'".format(member))


class StaticTypeMismatch(ValidationError):
    message: str

    def __init__(
        self, node: SourceNode, expected: Type.Base, actual: Type.Base, message: str = ""
    ) -> None:
        self.expected = expected
        self.actual = actual
        self.message = message
        super().__init__(node, message)

    def __str__(self) -> str:
        msg = f"Expected {self.expected} instead of {self.actual}"
        if self.message:
            msg += "; " + self.message
        elif isinstance(self.expected, Type.Int) and isinstance(self.actual, Type.Float):
            msg += "; perhaps try floor() or round()"
        elif str(self.actual).replace("?", "") == str(self.expected):
            msg += (
                " -- to coerce T? X into T, try select_first([X,defaultValue])"
                " or select_first([X]) (which might fail at runtime);"
                " to coerce Array[T?] X into Array[T], try select_all(X)"
            )
        return msg


class IncompatibleOperand(ValidationError):
    def __init__(self, node: SourceNode, message: str) -> None:
        super().__init__(node, message)


class UnknownIdentifier(ValidationError):
    def __init__(self, node: SourceNode, message: Optional[str] = None) -> None:
        # avoiding circular dep:
        # assert isinstance(node, WDL.Expr.Ident)
        if not message:
            message = "Unknown identifier " + str(node)
        super().__init__(node, message)


class NoSuchInput(ValidationError):
    def __init__(self, node: SourceNode, name: str) -> None:
        super().__init__(node, "No such input " + name)


class UncallableWorkflow(ValidationError):
    def __init__(self, node: SourceNode, name: str) -> None:
        super().__init__(
            node,
            (
                "Cannot call subworkflow {} because its own calls have missing required inputs, "
                "and/or it lacks an output section"
            ).format(name),
        )


class MultipleDefinitions(ValidationError):
    pass


class StrayInputDeclaration(ValidationError):
    pass


class CircularDependencies(ValidationError):
    def __init__(self, node: SourceNode) -> None:
        msg = "circular dependencies"
        nm = next(
            (getattr(node, attr) for attr in ("name", "workflow_node_id") if hasattr(node, attr)),
            None,
        )
        if nm:
            nm += " involving " + nm
        super().__init__(node, msg)


class MultipleValidationErrors(Exception):
    """Propagates several validation/typechecking errors"""

    exceptions: List[ValidationError]
    """:type: List[ValidationError]"""

    source_text: Optional[str] = None

    declared_wdl_version: Optional[str] = None

    def __init__(
        self, *exceptions: List[Union[ValidationError, "MultipleValidationErrors"]]
    ) -> None:
        super().__init__()
        self.exceptions = []
        for exn in exceptions:
            if isinstance(exn, ValidationError):
                self.exceptions.append(exn)
            elif isinstance(exn, MultipleValidationErrors):
                self.exceptions.extend(exn.exceptions)
            else:
                assert False
        assert self.exceptions
        self.exceptions = sorted(self.exceptions, key=lambda exn: getattr(exn, "pos"))


class _MultiContext:
    """"""

    _exceptions: List[Union[ValidationError, MultipleValidationErrors]]

    def __init__(self) -> None:
        self._exceptions = []

    def try1(self, fn: Callable[[], Any]) -> Optional[Any]:
        try:
            return fn()
        except (ValidationError, MultipleValidationErrors) as exn:
            self._exceptions.append(exn)
            return None

    def append(self, exn: Union[ValidationError, MultipleValidationErrors]) -> None:
        self._exceptions.append(exn)

    def maybe_raise(self) -> None:
        if len(self._exceptions) == 1:
            raise self._exceptions[0]
        if self._exceptions:
            raise MultipleValidationErrors(*self._exceptions) from self._exceptions[0]  # type: ignore


@contextmanager
def multi_context() -> Generator[_MultiContext, None, None]:
    """"""
    # Context manager to assist with catching and propagating multiple
    # validation/typechecking errors
    #
    # with WDL.Error.multi_context() as errors:
    #
    #    result = errors.try1(lambda: perform_validation())
    #    # Returns the result of invoking the lambda. If the lambda invocation
    #    # raises WDL.Error.ValidationError or
    #    # WDL.Error.MultipleValidationErrors, records the error and returns
    #    # None. (Other exceptions would halt execution and propagate
    #    # normally.)
    #
    #    errors.append(WDL.Error.NullValue())
    #    # errors.append() manually records one error.
    #
    # When the context closes, any exceptions recorded with errors.try1() or
    # errors.append() are raised at that point. Note that any exception raised
    # outside of errors.try1() will exit the context immediately and discard
    # any previously-recorded errors.
    #
    # Lastly, you can call errors.maybe_raise() to immediately propagate any
    # exceptions recorded so far, or if none, proceed with the remainder of
    # the context body.
    ctx = _MultiContext()
    yield ctx
    ctx.maybe_raise()


class RuntimeError(Exception):
    more_info: Dict[str, Any]
    """
    Backend-specific information about an error (for example, pointer to a centralized log system)
    """

    def __init__(self, *args, more_info: Optional[Dict[str, Any]] = None, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        self.more_info = more_info if more_info else {}


class EvalError(RuntimeError):
    """Error evaluating a WDL expression or declaration"""

    pos: SourcePosition
    """:type: SourcePosition"""

    node: Optional[SourceNode] = None
    """:type: Optional[SourceNode]"""

    def __init__(self, node: Union[SourceNode, SourcePosition], message: str) -> None:
        if isinstance(node, SourceNode):
            self.node = node
            self.pos = node.pos
        else:
            self.pos = node
        super().__init__(message)


class OutOfBounds(EvalError):
    pass


class EmptyArray(EvalError):
    def __init__(self, node: SourceNode) -> None:
        super().__init__(node, "Empty array for Array+ input/declaration")


class NullValue(EvalError):
    def __init__(self, node: Union[SourceNode, SourcePosition]) -> None:
        super().__init__(node, "Null value")


class InputError(RuntimeError):
    """Error reading an input value/file"""

    pass

1	from typing import (	1✔
2	List,
3	Optional,
4	Union,
5	Iterable,
6	Generator,
7	Callable,
8	Any,
9	Dict,
10	NamedTuple,
11	)
12	import os	1✔
13	from functools import total_ordering	1✔
14	from contextlib import contextmanager	1✔
15
16	from . import Type	1✔
17
18
19	class SourcePosition(	1✔
20	NamedTuple(
21	"SourcePosition",
22	[
23	("uri", str),
24	("abspath", str),
25	("line", int),
26	("column", int),
27	("end_line", int),
28	("end_column", int),
29	],
30	)
31	):
32	"""
33	Source position attached to AST nodes and exceptions; NamedTuple of ``uri`` the filename/URI
34	passed to :func:`WDL.load` or a WDL import statement, which may be relative; ``abspath`` the
35	absolute filename/URI; and one-based int positions ``line`` ``end_line`` ``column``
36	``end_column``
37	"""
38
39
40	class SyntaxError(Exception):	1✔
41	"""Failure to lex/parse a WDL document"""
42
43	pos: SourcePosition	1✔
44	wdl_version: str	1✔
45	declared_wdl_version: Optional[str]	1✔
46
47	def __init__(	1✔
48	self, pos: SourcePosition, msg: str, wdl_version: str, declared_wdl_version: Optional[str]
49	) -> None:
50	super().__init__(msg)	1✔
51	self.pos = pos	1✔
52	self.wdl_version = wdl_version	1✔
53	self.declared_wdl_version = declared_wdl_version	1✔
54
55
56	class _BadCharacterEncoding(Exception):	1✔
57	""""""
58
59	# Invalid escape sequence in a string literal; this is used internally, eventually resurfaced
60	# as a SyntaxError.
61	pos: SourcePosition	1✔
62
63	def __init__(self, pos: SourcePosition):	1✔
64	self.pos = pos	1✔
65
66
67	class ImportError(Exception):	1✔
68	"""Failure to open/retrieve an imported WDL document
69
70	The ``__cause__`` attribute may hold the inner error object."""
71
72	pos: SourcePosition	1✔
73
74	def __init__(self, pos: SourcePosition, import_uri: str, message: Optional[str] = None) -> None:	1✔
75	msg = "Failed to import " + import_uri	1✔
76	if message:	1✔
77	msg = msg + ", " + message	×
78	super().__init__(msg)	1✔
79	self.pos = pos	1✔
80
81
82	@total_ordering	1✔
83	class SourceNode:	1✔
84	"""Base class for an AST node, recording the source position"""
85
86	pos: SourcePosition	1✔
87	"""	×
88	:type: SourcePosition
89
90	Source position for this AST node
91	"""
92
93	parent: Optional["SourceNode"] = None	1✔
94	"""	×
95	:type: Optional[SourceNode]
96
97	Parent node in the AST, if any
98	"""
99
100	def __init__(self, pos: SourcePosition) -> None:	1✔
101	self.pos = pos	1✔
102
103	@property	1✔
104	def source_dir(self) -> str:	1✔
105	"""
106	Local directory containing this source node, with trailing "/", or "".
107
108	Source-relative File/Directory paths need an explicit local source directory. Parsed
109	buffers, remote URIs, and other non-local source locations are represented as an empty
110	string so callers can produce context-specific errors only when relative path resolution is
111	actually needed.
112
113	The directory is normalized but symlinks are intentionally preserved (abspath, not
114	realpath): if the user reached the WDL through a symlinked path, that's presumed deliberate
115	and is kept in resolved File/Directory values. Symlink resolution is applied only where it's
116	a security boundary (see ``path_really_within`` usage in source-relative path resolution).
117	"""
118	source = self.pos.abspath	1✔
119	if not source or source == "(buffer)" or not os.path.isabs(source):	1✔
120	return ""	1✔
121	return os.path.join(os.path.abspath(os.path.dirname(source)), "")	1✔
122
123	def __lt__(self, rhs: "SourceNode") -> bool:	1✔
UNCOV 124	return isinstance(rhs, SourceNode) and (	×
125	self.pos.abspath,
126	self.pos.line,
127	self.pos.column,
128	self.pos.end_line,
129	self.pos.end_column,
130	) < (
131	rhs.pos.abspath,
132	rhs.pos.line,
133	rhs.pos.column,
134	rhs.pos.end_line,
135	rhs.pos.end_column,
136	)
137
138	def __eq__(self, rhs: Any) -> bool:	1✔
139	return isinstance(rhs, SourceNode) and self.pos == rhs.pos	1✔
140
141	@property	1✔
142	def children(self) -> Iterable["SourceNode"]:	1✔
143	"""
144	:type: Iterable[SourceNode]
145
146	Yield all child nodes
147	"""
148	return []	1✔
149
150
151	class ValidationError(Exception):	1✔
152	"""
153	Base class for a WDL validation error (when the document loads and parses, but fails typechecking or other static
154	validity tests)
155	"""
156
157	pos: SourcePosition	1✔
UNCOV 158	""":type: SourcePosition"""	×
159
160	node: Optional[SourceNode] = None	1✔
UNCOV 161	""":type: Optional[SourceNode]"""	×
162
163	source_text: Optional[str] = None	1✔
UNCOV 164	""":type: Optional[str]	×
165
166	The complete source text of the WDL document (if available)"""
167
168	declared_wdl_version: Optional[str] = None	1✔
169
170	def __init__(self, node: Union[SourceNode, SourcePosition], message: str) -> None:	1✔
171	if isinstance(node, SourceNode):	1✔
172	self.node = node	1✔
173	self.pos = node.pos	1✔
174	else:
175	self.pos = node	1✔
176	super().__init__(message)	1✔
177
178
179	class InvalidType(ValidationError):	1✔
180	pass	1✔
181
182
183	class IndeterminateType(ValidationError):	1✔
184	pass	1✔
185
186
187	class NoSuchTask(ValidationError):	1✔
188	def __init__(self, node: Union[SourceNode, SourcePosition], name: str) -> None:	1✔
189	super().__init__(node, "No such task/workflow: " + name)	1✔
190
191
192	class NoSuchCall(ValidationError):	1✔
193	def __init__(self, node: Union[SourceNode, SourcePosition], name: str) -> None:	1✔
194	super().__init__(node, "No such call in this workflow: " + name)	1✔
195
196
197	class NoSuchFunction(ValidationError):	1✔
198	def __init__(self, node: Union[SourceNode, SourcePosition], name: str) -> None:	1✔
199	super().__init__(node, "No such function: " + name)	1✔
200
201
202	class WrongArity(ValidationError):	1✔
203	def __init__(self, node: Union[SourceNode, SourcePosition], expected: int) -> None:	1✔
204	# avoiding circular dep:
205	# assert isinstance(node, WDL.Expr.Apply)
206	msg = "{} expects {} argument(s)".format(getattr(node, "function_name"), expected)	1✔
207	super().__init__(node, msg)	1✔
208
209
210	class NotAnArray(ValidationError):	1✔
211	def __init__(self, node: Union[SourceNode, SourcePosition]) -> None:	1✔
212	super().__init__(node, "Not an array")	1✔
213
214
215	class NoSuchMember(ValidationError):	1✔
216	def __init__(self, node: Union[SourceNode, SourcePosition], member: str) -> None:	1✔
217	super().__init__(node, "No such member '{}'".format(member))	1✔
218
219
220	class StaticTypeMismatch(ValidationError):	1✔
221	message: str	1✔
222
223	def __init__(	1✔
224	self, node: SourceNode, expected: Type.Base, actual: Type.Base, message: str = ""
225	) -> None:
226	self.expected = expected	1✔
227	self.actual = actual	1✔
228	self.message = message	1✔
229	super().__init__(node, message)	1✔
230
231	def __str__(self) -> str:	1✔
232	msg = f"Expected {self.expected} instead of {self.actual}"	1✔
233	if self.message:	1✔
234	msg += "; " + self.message	1✔
UNCOV 235	elif isinstance(self.expected, Type.Int) and isinstance(self.actual, Type.Float):	×
UNCOV 236	msg += "; perhaps try floor() or round()"	×
UNCOV 237	elif str(self.actual).replace("?", "") == str(self.expected):	×
UNCOV 238	msg += (	×
239	" -- to coerce T? X into T, try select_first([X,defaultValue])"
240	" or select_first([X]) (which might fail at runtime);"
241	" to coerce Array[T?] X into Array[T], try select_all(X)"
242	)
243	return msg	1✔
244
245
246	class IncompatibleOperand(ValidationError):	1✔
247	def __init__(self, node: SourceNode, message: str) -> None:	1✔
248	super().__init__(node, message)	1✔
249
250
251	class UnknownIdentifier(ValidationError):	1✔
252	def __init__(self, node: SourceNode, message: Optional[str] = None) -> None:	1✔
253	# avoiding circular dep:
254	# assert isinstance(node, WDL.Expr.Ident)
255	if not message:	1✔
256	message = "Unknown identifier " + str(node)	1✔
257	super().__init__(node, message)	1✔
258
259
260	class NoSuchInput(ValidationError):	1✔
261	def __init__(self, node: SourceNode, name: str) -> None:	1✔
262	super().__init__(node, "No such input " + name)	1✔
263
264
265	class UncallableWorkflow(ValidationError):	1✔
266	def __init__(self, node: SourceNode, name: str) -> None:	1✔
267	super().__init__(	1✔
268	node,
269	(
270	"Cannot call subworkflow {} because its own calls have missing required inputs, "
271	"and/or it lacks an output section"
272	).format(name),
273	)
274
275
276	class MultipleDefinitions(ValidationError):	1✔
277	pass	1✔
278
279
280	class StrayInputDeclaration(ValidationError):	1✔
281	pass	1✔
282
283
284	class CircularDependencies(ValidationError):	1✔
285	def __init__(self, node: SourceNode) -> None:	1✔
286	msg = "circular dependencies"	1✔
287	nm = next(	1✔
288	(getattr(node, attr) for attr in ("name", "workflow_node_id") if hasattr(node, attr)),
289	None,
290	)
291	if nm:	1✔
292	nm += " involving " + nm	1✔
293	super().__init__(node, msg)	1✔
294
295
296	class MultipleValidationErrors(Exception):	1✔
297	"""Propagates several validation/typechecking errors"""
298
299	exceptions: List[ValidationError]	1✔
UNCOV 300	""":type: List[ValidationError]"""	×
301
302	source_text: Optional[str] = None	1✔
303
304	declared_wdl_version: Optional[str] = None	1✔
305
306	def __init__(	1✔
307	self, *exceptions: List[Union[ValidationError, "MultipleValidationErrors"]]
308	) -> None:
309	super().__init__()	1✔
310	self.exceptions = []	1✔
311	for exn in exceptions:	1✔
312	if isinstance(exn, ValidationError):	1✔
313	self.exceptions.append(exn)	1✔
314	elif isinstance(exn, MultipleValidationErrors):	1✔
315	self.exceptions.extend(exn.exceptions)	1✔
316	else:
UNCOV 317	assert False	×
318	assert self.exceptions	1✔
319	self.exceptions = sorted(self.exceptions, key=lambda exn: getattr(exn, "pos"))	1✔
320
321
322	class _MultiContext:	1✔
323	""""""
324
325	_exceptions: List[Union[ValidationError, MultipleValidationErrors]]	1✔
326
327	def __init__(self) -> None:	1✔
328	self._exceptions = []	1✔
329
330	def try1(self, fn: Callable[[], Any]) -> Optional[Any]:	1✔
331	try:	1✔
332	return fn()	1✔
333	except (ValidationError, MultipleValidationErrors) as exn:	1✔
334	self._exceptions.append(exn)	1✔
335	return None	1✔
336
337	def append(self, exn: Union[ValidationError, MultipleValidationErrors]) -> None:	1✔
338	self._exceptions.append(exn)	1✔
339
340	def maybe_raise(self) -> None:	1✔
341	if len(self._exceptions) == 1:	1✔
342	raise self._exceptions[0]	1✔
343	if self._exceptions:	1✔
344	raise MultipleValidationErrors(*self._exceptions) from self._exceptions[0] # type: ignore	1✔
345
346
347	@contextmanager	1✔
348	def multi_context() -> Generator[_MultiContext, None, None]:	1✔
349	""""""
350	# Context manager to assist with catching and propagating multiple
351	# validation/typechecking errors
352	#
353	# with WDL.Error.multi_context() as errors:
354	#
355	# result = errors.try1(lambda: perform_validation())
356	# # Returns the result of invoking the lambda. If the lambda invocation
357	# # raises WDL.Error.ValidationError or
358	# # WDL.Error.MultipleValidationErrors, records the error and returns
359	# # None. (Other exceptions would halt execution and propagate
360	# # normally.)
361	#
362	# errors.append(WDL.Error.NullValue())
363	# # errors.append() manually records one error.
364	#
365	# When the context closes, any exceptions recorded with errors.try1() or
366	# errors.append() are raised at that point. Note that any exception raised
367	# outside of errors.try1() will exit the context immediately and discard
368	# any previously-recorded errors.
369	#
370	# Lastly, you can call errors.maybe_raise() to immediately propagate any
371	# exceptions recorded so far, or if none, proceed with the remainder of
372	# the context body.
373	ctx = _MultiContext()	1✔
374	yield ctx	1✔
375	ctx.maybe_raise()	1✔
376
377
378	class RuntimeError(Exception):	1✔
379	more_info: Dict[str, Any]	1✔
UNCOV 380	"""	×
381	Backend-specific information about an error (for example, pointer to a centralized log system)
382	"""
383
384	def __init__(self, args, more_info: Optional[Dict[str, Any]] = None, *kwargs) -> None:	1✔
385	super().__init__(args, *kwargs)	1✔
386	self.more_info = more_info if more_info else {}	1✔
387
388
389	class EvalError(RuntimeError):	1✔
390	"""Error evaluating a WDL expression or declaration"""
391
392	pos: SourcePosition	1✔
UNCOV 393	""":type: SourcePosition"""	×
394
395	node: Optional[SourceNode] = None	1✔
UNCOV 396	""":type: Optional[SourceNode]"""	×
397
398	def __init__(self, node: Union[SourceNode, SourcePosition], message: str) -> None:	1✔
399	if isinstance(node, SourceNode):	1✔
400	self.node = node	1✔
401	self.pos = node.pos	1✔
402	else:
UNCOV 403	self.pos = node	×
404	super().__init__(message)	1✔
405
406
407	class OutOfBounds(EvalError):	1✔
408	pass	1✔
409
410
411	class EmptyArray(EvalError):	1✔
412	def __init__(self, node: SourceNode) -> None:	1✔
413	super().__init__(node, "Empty array for Array+ input/declaration")	1✔
414
415
416	class NullValue(EvalError):	1✔
417	def __init__(self, node: Union[SourceNode, SourcePosition]) -> None:	1✔
418	super().__init__(node, "Null value")	1✔
419
420
421	class InputError(RuntimeError):	1✔
422	"""Error reading an input value/file"""
423
424	pass	1✔

chanzuckerberg / miniwdl / 28270112145

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