Skip to content

Commit 218575d

Browse files
tobiasraabetobiasraabe
authored
Move and publish some functions related to warnings. (#298)
1 parent 8d91eab commit 218575d

File tree

4 files changed

+188
-162
lines changed

4 files changed

+188
-162
lines changed

‎docs/source/changes.md

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,12 @@ chronological order. Releases follow [semantic versioning](https://semver.org/)
55
releases are available on [PyPI](https://pypi.org/project/pytask) and
66
[Anaconda.org](https://anaconda.org/conda-forge/pytask).
77

8-
## 0.2.5 - 2022-xx-xx
8+
## 0.2.6 - 2022-xx-xx
9+
10+
- {pull}`297` moves non-hook functions from `warnings.py` to `warnings_utils.py` and
11+
publishes them so that pytask-parallel can import them.
12+
13+
## 0.2.5 - 2022-08-02
914

1015
- {pull}`288` fixes pinning pybaum to v0.1.1 or a version that supports `tree_yield()`.
1116
- {pull}`289` shortens the task ids when using `pytask collect`. Fixes {issue}`286`.

‎src/_pytask/warnings.py

Lines changed: 3 additions & 161 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,8 @@
11
"""This module contains code for capturing warnings."""
22
from __future__ import annotations
33

4-
import functools
5-
import re
6-
import textwrap
7-
import warnings
84
from collections import defaultdict
9-
from contextlib import contextmanager
105
from typing import Any
11-
from typing import cast
126
from typing import Dict
137
from typing import Generator
148
from typing import List
@@ -17,11 +11,10 @@
1711
import click
1812
from _pytask.config import hookimpl
1913
from _pytask.console import console
20-
from _pytask.mark_utils import get_marks
2114
from _pytask.nodes import Task
22-
from _pytask.outcomes import Exit
2315
from _pytask.session import Session
24-
from _pytask.warnings_utils import WarningReport
16+
from _pytask.warnings_utils import catch_warnings_for_item
17+
from _pytask.warnings_utils import parse_filterwarnings
2518
from rich.console import Console
2619
from rich.console import ConsoleOptions
2720
from rich.console import RenderResult
@@ -50,7 +43,7 @@ def pytask_parse_config(
5043
) -> None:
5144
"""Parse the configuration."""
5245
config["disable_warnings"] = config_from_cli.get("disable_warnings", False)
53-
config["filterwarnings"] = _parse_filterwarnings(
46+
config["filterwarnings"] = parse_filterwarnings(
5447
config_from_file.get("filterwarnings")
5548
)
5649
config["markers"]["filterwarnings"] = "Add a filter for a warning to a task."
@@ -63,157 +56,6 @@ def pytask_post_parse(config: dict[str, Any]) -> None:
6356
config["pm"].register(WarningsNameSpace)
6457

6558

66-
def _parse_filterwarnings(x: str | list[str] | None) -> list[str]:
67-
"""Parse filterwarnings."""
68-
if x is None:
69-
return []
70-
elif isinstance(x, str):
71-
return [i.strip() for i in x.split("\n")]
72-
elif isinstance(x, list):
73-
return [i.strip() for i in x]
74-
else:
75-
raise TypeError("'filterwarnings' must be a str, list[str] or None.")
76-
77-
78-
@contextmanager
79-
def catch_warnings_for_item(
80-
session: Session,
81-
task: Task | None = None,
82-
when: str | None = None,
83-
) -> Generator[None, None, None]:
84-
"""Context manager that catches warnings generated in the contained execution block.
85-
86-
``item`` can be None if we are not in the context of an item execution. Each warning
87-
captured triggers the ``pytest_warning_recorded`` hook.
88-
89-
"""
90-
with warnings.catch_warnings(record=True) as log:
91-
# mypy can't infer that record=True means log is not None; help it.
92-
assert log is not None
93-
94-
for arg in session.config["filterwarnings"]:
95-
warnings.filterwarnings(*parse_warning_filter(arg, escape=False))
96-
97-
# apply filters from "filterwarnings" marks
98-
if task is not None:
99-
for mark in get_marks(task, "filterwarnings"):
100-
for arg in mark.args:
101-
warnings.filterwarnings(*parse_warning_filter(arg, escape=False))
102-
103-
yield
104-
105-
if task is not None:
106-
id_ = task.short_name
107-
else:
108-
id_ = when
109-
110-
for warning_message in log:
111-
fs_location = warning_message.filename, warning_message.lineno
112-
session.warnings.append(
113-
WarningReport(
114-
message=warning_record_to_str(warning_message),
115-
fs_location=fs_location,
116-
id_=id_,
117-
)
118-
)
119-
120-
121-
@functools.lru_cache(maxsize=50)
122-
def parse_warning_filter(
123-
arg: str, *, escape: bool
124-
) -> tuple[warnings._ActionKind, str, type[Warning], str, int]:
125-
"""Parse a warnings filter string.
126-
127-
This is copied from warnings._setoption with the following changes:
128-
129-
- Does not apply the filter.
130-
- Escaping is optional.
131-
- Raises UsageError so we get nice error messages on failure.
132-
133-
"""
134-
__tracebackhide__ = True
135-
error_template = textwrap.dedent(
136-
f"""\
137-
while parsing the following warning configuration:
138-
{arg}
139-
This error occurred:
140-
{{error}}
141-
"""
142-
)
143-
144-
parts = arg.split(":")
145-
if len(parts) > 5:
146-
doc_url = (
147-
"https://docs.python.org/3/library/warnings.html#describing-warning-filters"
148-
)
149-
error = textwrap.dedent(
150-
f"""\
151-
Too many fields ({len(parts)}), expected at most 5 separated by colons:
152-
action:message:category:module:line
153-
For more information please consult: {doc_url}
154-
"""
155-
)
156-
raise Exit(error_template.format(error=error))
157-
158-
while len(parts) < 5:
159-
parts.append("")
160-
action_, message, category_, module, lineno_ = (s.strip() for s in parts)
161-
try:
162-
action: warnings._ActionKind = warnings._getaction(action_) # type: ignore
163-
except warnings._OptionError as e:
164-
raise Exit(error_template.format(error=str(e)))
165-
try:
166-
category: type[Warning] = _resolve_warning_category(category_)
167-
except Exit as e:
168-
raise Exit(str(e))
169-
if message and escape:
170-
message = re.escape(message)
171-
if module and escape:
172-
module = re.escape(module) + r"\Z"
173-
if lineno_:
174-
try:
175-
lineno = int(lineno_)
176-
if lineno < 0:
177-
raise ValueError("number is negative")
178-
except ValueError as e:
179-
raise Exit(error_template.format(error=f"invalid lineno {lineno_!r}: {e}"))
180-
else:
181-
lineno = 0
182-
return action, message, category, module, lineno
183-
184-
185-
def _resolve_warning_category(category: str) -> type[Warning]:
186-
"""Copied from warnings._getcategory, but changed so it lets exceptions (specially
187-
ImportErrors) propagate so we can get access to their tracebacks (#9218)."""
188-
__tracebackhide__ = True
189-
if not category:
190-
return Warning
191-
192-
if "." not in category:
193-
import builtins as m
194-
195-
klass = category
196-
else:
197-
module, _, klass = category.rpartition(".")
198-
m = __import__(module, None, None, [klass])
199-
cat = getattr(m, klass)
200-
if not issubclass(cat, Warning):
201-
raise Exception(f"{cat} is not a Warning subclass")
202-
return cast(type[Warning], cat)
203-
204-
205-
def warning_record_to_str(warning_message: warnings.WarningMessage) -> str:
206-
"""Convert a warnings.WarningMessage to a string."""
207-
msg = warnings.formatwarning(
208-
message=warning_message.message,
209-
category=warning_message.category,
210-
filename=warning_message.filename,
211-
lineno=warning_message.lineno,
212-
line=warning_message.line,
213-
)
214-
return msg
215-
216-
21759
class WarningsNameSpace:
21860
"""A namespace for the warnings plugin."""
21961

‎src/_pytask/warnings_utils.py

Lines changed: 173 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,185 @@
11
from __future__ import annotations
22

3+
import functools
4+
import re
5+
import textwrap
6+
import warnings
7+
from contextlib import contextmanager
8+
from typing import cast
9+
from typing import Generator
310
from typing import Tuple
11+
from typing import TYPE_CHECKING
412

513
import attr
14+
from _pytask.mark_utils import get_marks
15+
from _pytask.nodes import Task
16+
from _pytask.outcomes import Exit
17+
18+
if TYPE_CHECKING:
19+
from _pytask.session import Session
20+
21+
22+
__all__ = [
23+
"catch_warnings_for_item",
24+
"parse_filterwarnings",
25+
"parse_warning_filter",
26+
"WarningReport",
27+
]
628

729

830
@attr.s(kw_only=True)
931
class WarningReport:
1032
message = attr.ib(type=str)
1133
fs_location = attr.ib(type=Tuple[str, int])
1234
id_ = attr.ib(type=str)
35+
36+
37+
@functools.lru_cache(maxsize=50)
38+
def parse_warning_filter(
39+
arg: str, *, escape: bool
40+
) -> tuple[warnings._ActionKind, str, type[Warning], str, int]:
41+
"""Parse a warnings filter string.
42+
43+
This is copied from warnings._setoption with the following changes:
44+
45+
- Does not apply the filter.
46+
- Escaping is optional.
47+
- Raises UsageError so we get nice error messages on failure.
48+
49+
"""
50+
__tracebackhide__ = True
51+
error_template = textwrap.dedent(
52+
f"""\
53+
while parsing the following warning configuration:
54+
{arg}
55+
This error occurred:
56+
{{error}}
57+
"""
58+
)
59+
60+
parts = arg.split(":")
61+
if len(parts) > 5:
62+
doc_url = (
63+
"https://docs.python.org/3/library/warnings.html#describing-warning-filters"
64+
)
65+
error = textwrap.dedent(
66+
f"""\
67+
Too many fields ({len(parts)}), expected at most 5 separated by colons:
68+
action:message:category:module:line
69+
For more information please consult: {doc_url}
70+
"""
71+
)
72+
raise Exit(error_template.format(error=error))
73+
74+
while len(parts) < 5:
75+
parts.append("")
76+
action_, message, category_, module, lineno_ = (s.strip() for s in parts)
77+
try:
78+
action: warnings._ActionKind = warnings._getaction(action_) # type: ignore
79+
except warnings._OptionError as e:
80+
raise Exit(error_template.format(error=str(e)))
81+
try:
82+
category: type[Warning] = _resolve_warning_category(category_)
83+
except Exit as e:
84+
raise Exit(str(e))
85+
if message and escape:
86+
message = re.escape(message)
87+
if module and escape:
88+
module = re.escape(module) + r"\Z"
89+
if lineno_:
90+
try:
91+
lineno = int(lineno_)
92+
if lineno < 0:
93+
raise ValueError("number is negative")
94+
except ValueError as e:
95+
raise Exit(error_template.format(error=f"invalid lineno {lineno_!r}: {e}"))
96+
else:
97+
lineno = 0
98+
return action, message, category, module, lineno
99+
100+
101+
def _resolve_warning_category(category: str) -> type[Warning]:
102+
"""Copied from warnings._getcategory, but changed so it lets exceptions (specially
103+
ImportErrors) propagate so we can get access to their tracebacks (#9218)."""
104+
__tracebackhide__ = True
105+
if not category:
106+
return Warning
107+
108+
if "." not in category:
109+
import builtins as m
110+
111+
klass = category
112+
else:
113+
module, _, klass = category.rpartition(".")
114+
m = __import__(module, None, None, [klass])
115+
cat = getattr(m, klass)
116+
if not issubclass(cat, Warning):
117+
raise Exception(f"{cat} is not a Warning subclass")
118+
return cast(type[Warning], cat)
119+
120+
121+
def warning_record_to_str(warning_message: warnings.WarningMessage) -> str:
122+
"""Convert a warnings.WarningMessage to a string."""
123+
msg = warnings.formatwarning(
124+
message=warning_message.message,
125+
category=warning_message.category,
126+
filename=warning_message.filename,
127+
lineno=warning_message.lineno,
128+
line=warning_message.line,
129+
)
130+
return msg
131+
132+
133+
def parse_filterwarnings(x: str | list[str] | None) -> list[str]:
134+
"""Parse filterwarnings."""
135+
if x is None:
136+
return []
137+
elif isinstance(x, str):
138+
return [i.strip() for i in x.split("\n")]
139+
elif isinstance(x, list):
140+
return [i.strip() for i in x]
141+
else:
142+
raise TypeError("'filterwarnings' must be a str, list[str] or None.")
143+
144+
145+
@contextmanager
146+
def catch_warnings_for_item(
147+
session: Session,
148+
task: Task | None = None,
149+
when: str | None = None,
150+
) -> Generator[None, None, None]:
151+
"""Context manager that catches warnings generated in the contained execution block.
152+
153+
``item`` can be None if we are not in the context of an item execution. Each warning
154+
captured triggers the ``pytest_warning_recorded`` hook.
155+
156+
"""
157+
with warnings.catch_warnings(record=True) as log:
158+
# mypy can't infer that record=True means log is not None; help it.
159+
assert log is not None
160+
161+
for arg in session.config["filterwarnings"]:
162+
warnings.filterwarnings(*parse_warning_filter(arg, escape=False))
163+
164+
# apply filters from "filterwarnings" marks
165+
if task is not None:
166+
for mark in get_marks(task, "filterwarnings"):
167+
for arg in mark.args:
168+
warnings.filterwarnings(*parse_warning_filter(arg, escape=False))
169+
170+
yield
171+
172+
if task is not None:
173+
id_ = task.short_name
174+
else:
175+
id_ = when
176+
177+
for warning_message in log:
178+
fs_location = warning_message.filename, warning_message.lineno
179+
session.warnings.append(
180+
WarningReport(
181+
message=warning_record_to_str(warning_message),
182+
fs_location=fs_location,
183+
id_=id_,
184+
)
185+
)

0 commit comments

Comments
 (0)