wehub-resource-sync
263 lines
9.0 KiB
Python
263 lines
9.0 KiB
Python
"""Flag patches that globally clobber ``asyncio`` module attributes.
|
|
|
|
What goes wrong
|
|
---------------
|
|
|
|
When test code writes::
|
|
|
|
with patch("omnigent.tools.mcp.asyncio.sleep", new_callable=AsyncMock):
|
|
...
|
|
|
|
``unittest.mock``'s dotted-path resolver walks:
|
|
|
|
1. ``import omnigent.tools.mcp``
|
|
2. ``getattr(omnigent.tools.mcp, "asyncio")`` -> returns the
|
|
``asyncio`` module singleton (because production code did
|
|
``import asyncio`` at module top).
|
|
3. ``setattr(asyncio_module, "sleep", mock)`` -> globally clobbers
|
|
``asyncio.sleep`` for the entire process for the patch's lifetime.
|
|
|
|
Under ``pytest-xdist``, concurrent tests in other workers that touch
|
|
``asyncio.sleep`` (directly or via any of the many helpers built on
|
|
top of it) silently get the mock, causing hard-to-debug cross-test
|
|
flakiness.
|
|
|
|
The same bug shape applies to:
|
|
|
|
- ``monkeypatch.setattr("...asyncio.sleep", ...)`` (pytest)
|
|
- ``patch.object(mod.asyncio, "sleep", ...)``
|
|
|
|
What this catches
|
|
-----------------
|
|
|
|
1. ``patch("...asyncio.<name>", ...)`` / ``mock.patch(...)`` where the
|
|
first positional arg is a string literal containing ``.asyncio.``
|
|
or ending in ``.asyncio``.
|
|
2. ``monkeypatch.setattr("...asyncio.<name>", ...)`` with the same
|
|
string-literal shape.
|
|
3. ``patch.object(<expr>, "<name>", ...)`` where ``<expr>`` is an
|
|
attribute access whose leaf attribute is ``asyncio``
|
|
(e.g. ``patch.object(mcp_mod.asyncio, "sleep", ...)``).
|
|
|
|
Correct shapes
|
|
--------------
|
|
|
|
Option A (preferred, when production has few sleep sites): add a
|
|
thin ``_sleep`` indirection in the production module and patch
|
|
``module._sleep`` from tests. The real ``asyncio.sleep`` is never
|
|
touched.
|
|
|
|
Option B (when A is too invasive): replace the ``asyncio`` binding
|
|
in the target module's namespace with a ``SimpleNamespace`` that
|
|
mirrors the real module but with the desired attribute swapped.
|
|
The real ``asyncio`` module is left alone.
|
|
|
|
Exit code 1 if any hits are found.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
def _is_global_asyncio_literal(value: str) -> bool:
|
|
"""Return True if ``value`` is a one-segment getattr off the
|
|
stdlib ``asyncio`` module name (the shape that clobbers the
|
|
singleton).
|
|
|
|
The dangerous shape is::
|
|
|
|
patch("<pkg>.<mod>.asyncio.<leaf>", ...)
|
|
|
|
where ``<leaf>`` is exactly one segment. The dotted-path resolver
|
|
imports ``<pkg>.<mod>``, calls ``getattr(_, "asyncio")`` -- which
|
|
returns the stdlib ``asyncio`` module singleton because the
|
|
consumer did ``import asyncio`` -- then sets ``<leaf>`` on it.
|
|
|
|
NOT flagged:
|
|
|
|
* ``"<pkg>.asyncio.<sub>.<leaf>"`` -- here ``asyncio`` is a
|
|
subpackage name (e.g. ``websockets.asyncio.client``) and the
|
|
resolver descends *through* a package directory, not through
|
|
a getattr on the stdlib module.
|
|
* ``"<pkg>.asyncio"`` -- the bare module path is used to *replace*
|
|
the binding (the safe Option B pattern), not to clobber an
|
|
attribute inside it.
|
|
|
|
:param value: The dotted-path string passed to ``patch`` /
|
|
``monkeypatch.setattr``.
|
|
:returns: ``True`` if the path has exactly one segment after the
|
|
last ``.asyncio.``.
|
|
"""
|
|
marker = ".asyncio."
|
|
idx = value.rfind(marker)
|
|
if idx == -1:
|
|
return False
|
|
tail = value[idx + len(marker) :]
|
|
# Exactly one segment after `.asyncio.` -> getattr on the stdlib
|
|
# module singleton. Two or more segments -> the path descends
|
|
# through a subpackage named `asyncio`.
|
|
return bool(tail) and "." not in tail
|
|
|
|
|
|
def _is_patch_func(func: ast.expr) -> bool:
|
|
"""Return True if ``func`` spells ``patch`` or ``mock.patch``.
|
|
|
|
Plain ``patch.object`` is handled separately -- this only matches
|
|
the dotted-path form ``patch("...")`` / ``mock.patch("...")``.
|
|
"""
|
|
if isinstance(func, ast.Name) and func.id == "patch":
|
|
return True
|
|
if isinstance(func, ast.Attribute) and func.attr == "patch":
|
|
return True
|
|
return False
|
|
|
|
|
|
def _is_patch_object(func: ast.expr) -> bool:
|
|
"""Return True if ``func`` spells ``patch.object`` or ``mock.patch.object``."""
|
|
if not isinstance(func, ast.Attribute) or func.attr != "object":
|
|
return False
|
|
inner = func.value
|
|
if isinstance(inner, ast.Name) and inner.id == "patch":
|
|
return True
|
|
if isinstance(inner, ast.Attribute) and inner.attr == "patch":
|
|
return True
|
|
return False
|
|
|
|
|
|
def _is_monkeypatch_setattr(func: ast.expr) -> bool:
|
|
"""Return True if ``func`` spells ``<monkeypatch>.setattr``.
|
|
|
|
Conservative: matches any attribute access where the leaf is
|
|
``setattr`` and the receiver is a ``Name`` containing
|
|
``monkeypatch`` (case-insensitive). This covers the common
|
|
``monkeypatch.setattr(...)`` and ``m.setattr(...)`` (from
|
|
``with monkeypatch.context() as m:``).
|
|
|
|
:param func: The function-position expression of a ``Call``.
|
|
:returns: ``True`` if this looks like ``monkeypatch.setattr``.
|
|
"""
|
|
if not isinstance(func, ast.Attribute) or func.attr != "setattr":
|
|
return False
|
|
base = func.value
|
|
if isinstance(base, ast.Name):
|
|
return "monkeypatch" in base.id.lower() or base.id == "m"
|
|
return False
|
|
|
|
|
|
def _attribute_ends_in_asyncio(node: ast.expr) -> bool:
|
|
"""Return True if ``node`` is an attribute access ending in ``.asyncio``.
|
|
|
|
e.g. matches ``mcp_mod.asyncio`` and ``omnigent.tools.mcp.asyncio``.
|
|
"""
|
|
return isinstance(node, ast.Attribute) and node.attr == "asyncio"
|
|
|
|
|
|
def _check_string_literal_first_arg(node: ast.Call) -> bool:
|
|
"""Return True if ``node.args[0]`` is a string literal flagged by
|
|
:func:`_is_global_asyncio_literal`."""
|
|
if not node.args:
|
|
return False
|
|
first = node.args[0]
|
|
if not isinstance(first, ast.Constant) or not isinstance(first.value, str):
|
|
return False
|
|
return _is_global_asyncio_literal(first.value)
|
|
|
|
|
|
def scan(path: Path) -> list[tuple[int, str]]:
|
|
"""
|
|
Return ``[(lineno, snippet), ...]`` for each violation in ``path``.
|
|
|
|
:param path: File to scan.
|
|
:returns: One entry per flagged call site.
|
|
"""
|
|
try:
|
|
source = path.read_text()
|
|
tree = ast.parse(source)
|
|
except (SyntaxError, UnicodeDecodeError):
|
|
return []
|
|
source_lines = source.splitlines()
|
|
hits: list[tuple[int, str]] = []
|
|
for node in ast.walk(tree):
|
|
if not isinstance(node, ast.Call):
|
|
continue
|
|
if not _is_flagged_call(node):
|
|
continue
|
|
line = node.lineno
|
|
snippet = source_lines[line - 1].strip() if 0 < line <= len(source_lines) else ""
|
|
hits.append((line, snippet))
|
|
return hits
|
|
|
|
|
|
def _is_flagged_call(node: ast.Call) -> bool:
|
|
"""Return True if ``node`` is one of the banned patch shapes.
|
|
|
|
Covers three call shapes:
|
|
|
|
1. ``patch("...asyncio.<name>", ...)`` / ``mock.patch(...)``.
|
|
2. ``monkeypatch.setattr("...asyncio.<name>", ...)``.
|
|
3. ``patch.object(<expr ending in .asyncio>, "<name>", ...)``.
|
|
|
|
:param node: A ``Call`` node from the AST.
|
|
:returns: ``True`` if the call matches a banned shape.
|
|
"""
|
|
# patch("...asyncio.<name>", ...) / mock.patch("...asyncio.<name>", ...)
|
|
if _is_patch_func(node.func) and _check_string_literal_first_arg(node):
|
|
return True
|
|
# monkeypatch.setattr("...asyncio.<name>", ...)
|
|
if _is_monkeypatch_setattr(node.func) and _check_string_literal_first_arg(node):
|
|
return True
|
|
# patch.object(<...>.asyncio, "<name>", ...)
|
|
if (
|
|
_is_patch_object(node.func)
|
|
and len(node.args) >= 1
|
|
and _attribute_ends_in_asyncio(node.args[0])
|
|
):
|
|
return True
|
|
return False
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
"""
|
|
Scan every file in ``argv[1:]`` and report violations.
|
|
|
|
:param argv: Command-line args (``argv[0]`` is the script name).
|
|
:returns: Exit code -- 0 on clean scan, 1 if any hits.
|
|
"""
|
|
failed = False
|
|
for arg in argv[1:]:
|
|
path = Path(arg)
|
|
if not path.is_file():
|
|
continue
|
|
hits = scan(path)
|
|
if hits:
|
|
failed = True
|
|
for line, snippet in hits:
|
|
sys.stdout.write(
|
|
f"{path}:{line}: globally-clobbering asyncio patch detected: {snippet}\n"
|
|
)
|
|
sys.stdout.write(
|
|
" hint: see dev/lint/lint_no_global_asyncio_patch.py "
|
|
"docstring for correct shapes.\n"
|
|
)
|
|
if failed:
|
|
sys.stdout.write(
|
|
"\nPatching ``module.asyncio.sleep`` (or any other asyncio "
|
|
"attribute via a dotted path that walks through the asyncio "
|
|
"module) mutates the real asyncio module singleton, which "
|
|
"leaks into every other test running in the same process "
|
|
"(critical under pytest-xdist). Prefer adding a thin "
|
|
"``_sleep`` indirection in the production module and "
|
|
"patching that, or replace the module's ``asyncio`` binding "
|
|
"wholesale with a ``SimpleNamespace`` -- see the script "
|
|
"docstring for Option A / Option B examples.\n"
|
|
)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv))
|