diff --git a/dpnp/tests/conftest.py b/dpnp/tests/conftest.py index bd6c71f9a92..5d766566bca 100644 --- a/dpnp/tests/conftest.py +++ b/dpnp/tests/conftest.py @@ -44,6 +44,7 @@ import dpnp from .helper import get_dev_id +from .infra_warning_utils import register_infra_warnings_plugin_if_enabled skip_mark = pytest.mark.skip(reason="Skipping test.") @@ -114,6 +115,8 @@ def pytest_configure(config): "ignore:invalid value encountered in arccosh:RuntimeWarning", ) + register_infra_warnings_plugin_if_enabled(config) + def pytest_collection_modifyitems(config, items): test_path = os.path.split(__file__)[0] diff --git a/dpnp/tests/infra_warning_utils.py b/dpnp/tests/infra_warning_utils.py new file mode 100644 index 00000000000..5672cbce437 --- /dev/null +++ b/dpnp/tests/infra_warning_utils.py @@ -0,0 +1,239 @@ +import json +import os +import sys + +from collections import Counter + +import dpctl +import numpy + +import dpnp + + +def _env_check(var_name: str, *, default: bool = False) -> bool: + value = os.getenv(var_name) + if value is None: + return default + return value.strip().lower() in {"1", "true", "yes", "y", "on"} + + +def _origin_from_filename(filename: str) -> str: + file = (filename or "").replace("\\", "/") + if "/dpnp/" in file or file.startswith("dpnp/"): + return "dpnp" + if "/numpy/" in file or file.startswith("numpy/"): + return "numpy" + if "/dpctl/" in file or file.startswith("dpctl/"): + return "dpctl" + return "third_party" + + +def _json_dumps_one_line(obj) -> str: + return json.dumps(obj, separators=(",", ":")) + + +class DpnpInfraWarningsPlugin: + """Pytest custom plugin that records pytest-captured warnings. + + It only records what pytest already captures (via pytest_warning_recorded). + Does not change warnings filters. + + Env vars: + - DPNP_INFRA_WARNINGS_ENABLE=1 (enables the plugin) + - DPNP_INFRA_WARNINGS_DIRECTORY= (writes artifacts) + - DPNP_INFRA_WARNINGS_EVENTS_ARTIFACT (optional filename) + - DPNP_INFRA_WARNINGS_SUMMARY_ARTIFACT (optional filename) + """ + + SUMMARY_BEGIN = "DPNP_WARNINGS_SUMMARY_BEGIN" + SUMMARY_END = "DPNP_WARNINGS_SUMMARY_END" + EVENT_PREFIX = "DPNP_WARNING_EVENT " + + def __init__(self): + self.enabled = _env_check("DPNP_INFRA_WARNINGS_ENABLE", default=False) + self.directory = os.getenv("DPNP_INFRA_WARNINGS_DIRECTORY", None) + self.events_artifact = os.getenv( + "DPNP_INFRA_WARNINGS_EVENTS_ARTIFACT", "dpnp_infra_warnings_events.jsonl" + ) + self.summary_artifact = os.getenv( + "DPNP_INFRA_WARNINGS_SUMMARY_ARTIFACT", "dpnp_infra_warnings_summary.json" + ) + + self.print_events = self.enabled + + self._counts = Counter() + self._examples = {} + self._totals = Counter() + self._env = {} + + self._events_fp = None + self._events_file = None + + def pytest_configure(self, config): + if not self.enabled: + return + + try: + numpy_version = numpy.__version__ + numpy_path = getattr(numpy, "__file__", "unknown") + dpnp_version = dpnp.__version__ + dpnp_path = getattr(dpnp, "__file__", "unknown") + dpctl_version = dpctl.__version__ + dpctl_path = getattr(dpctl, "__file__", "unknown") + except Exception: + numpy_version = "unknown" + numpy_path = "unknown" + dpnp_version = "unknown" + dpnp_path = "unknown" + dpctl_version = "unknown" + dpctl_path = "unknown" + + if self.directory: + os.makedirs(self.directory, exist_ok=True) + self._events_file = os.path.join(self.directory, self.events_artifact) + self._events_fp = open( + self._events_file, + "w", + encoding="utf-8", + buffering=1, + newline="\n", + ) + + self._env.update( + { + "numpy_version": numpy_version, + "numpy_path": numpy_path, + "dpnp_version": dpnp_version, + "dpnp_path": dpnp_path, + "dpctl_version": dpctl_version, + "dpctl_path": dpctl_path, + "job": os.getenv("JOB_NAME", "unknown"), + "build_number": os.getenv("BUILD_NUMBER", "unknown"), + "git_sha": os.getenv("GIT_COMMIT", "unknown"), + "events_file": self._events_file, + } + ) + + def pytest_warning_recorded(self, warning_message, when, nodeid, location): + if not self.enabled: + return + + category = getattr( + getattr(warning_message, "category", None), + "__name__", + str(getattr(warning_message, "category", "Warning")), + ) + message = str(getattr(warning_message, "message", warning_message)) + + filename = getattr(warning_message, "filename", None) or ( + location[0] if location and len(location) > 0 else None + ) + lineno = getattr(warning_message, "lineno", None) or ( + location[1] if location and len(location) > 1 else None + ) + func = location[2] if location and len(location) > 2 else None + + origin = _origin_from_filename(filename or "") + key = f"{category}||{origin}||{message}" + self._counts[key] += 1 + self._totals[f"category::{category}"] += 1 + self._totals[f"origin::{origin}"] += 1 + self._totals[f"phase::{when}"] += 1 + + if key not in self._examples: + self._examples[key] = { + "category": category, + "origin": origin, + "when": when, + "nodeid": nodeid, + "filename": filename, + "lineno": lineno, + "function": func, + "message": message, + } + + event = { + "when": when, + "nodeid": nodeid, + "category": category, + "origin": origin, + "message": message, + "filename": filename, + "lineno": lineno, + "function": func, + } + + if self._events_fp is not None: + try: + self._events_fp.write(_json_dumps_one_line(event) + "\n") + except Exception: + pass + + if self.print_events: + try: + sys.stderr.write(self.EVENT_PREFIX + _json_dumps_one_line(event) + "\n") + sys.stderr.flush() + except Exception: + pass + + def pytest_terminal_summary(self, terminalreporter, exitstatus, config): + if not self.enabled: + return + + summary = { + "schema_version": "1.0", + "exit_status": exitstatus, + "environment": dict(self._env), + "total_warning_events": int(sum(self._counts.values())), + "unique_warning_types": int(len(self._counts)), + "totals": dict(self._totals), + "top_unique_warnings": [ + dict(self._examples[k], count=c) + for k, c in self._counts.most_common(50) + if k in self._examples + ], + } + + if self.directory: + output_file = os.path.join(self.directory, self.summary_artifact) + try: + with open(output_file, "w", encoding="utf-8") as f: + json.dump(summary, f, indent=2, sort_keys=True) + terminalreporter.write_line( + f"DPNP infrastructure warnings summary written to: {output_file}" + ) + except Exception as exc: + terminalreporter.write_line( + f"Failed to write DPNP infrastructure warnings summary to: {output_file}. Error: {exc}" + ) + + self._close_events_fp() + + terminalreporter.write_line(self.SUMMARY_BEGIN) + terminalreporter.write_line(_json_dumps_one_line(summary)) + terminalreporter.write_line(self.SUMMARY_END) + + def pytest_unconfigure(self, config): + self._close_events_fp() + + def _close_events_fp(self): + if self._events_fp is None: + return + try: + self._events_fp.close() + except Exception: + pass + self._events_fp = None + + +def register_infra_warnings_plugin_if_enabled(config) -> None: + """Register infra warnings plugin if enabled via env var.""" + + if not _env_check("DPNP_INFRA_WARNINGS_ENABLE"): + return + + plugin_name = "dpnp-infra-warnings" + if config.pluginmanager.get_plugin(plugin_name) is not None: + return + + config.pluginmanager.register(DpnpInfraWarningsPlugin(), plugin_name)