Merge remote-tracking branch 'origin/main' into woa-nvidia-wsl-fallback
This commit is contained in:
Daniel Han
commit
05f909915c
383 changed files with 7003 additions and 12043 deletions
|
|
@ -52,9 +52,7 @@ EXPECTED_NOISE_FILES = {
|
|||
}
|
||||
|
||||
# Only quoted-string occurrences in these file types can be module specifiers.
|
||||
JS_LIKE_EXT = re.compile(
|
||||
r"\.(ts|tsx|js|jsx|mjs|cjs|html|htm|css|scss|sass|json|jsonc)$"
|
||||
)
|
||||
JS_LIKE_EXT = re.compile(r"\.(ts|tsx|js|jsx|mjs|cjs|html|htm|css|scss|sass|json|jsonc)$")
|
||||
# Files where JS-syntactic import patterns (static/dynamic/require/re-export)
|
||||
# could be a real module reference. Markdown gets a separate gate (.mdx is
|
||||
# real ESM; .md code fences are not).
|
||||
|
|
@ -273,9 +271,7 @@ def classify(pkg: str, file: str, content: str) -> str | None:
|
|||
if is_script and re.search(rf"\bimport\(\s*['\"]{esc}{sub}['\"]\s*\)", content):
|
||||
return "dynamic_import"
|
||||
# require / require.resolve
|
||||
if is_script and re.search(
|
||||
rf"\brequire(?:\.resolve)?\(\s*['\"]{esc}{sub}['\"]\s*\)", content
|
||||
):
|
||||
if is_script and re.search(rf"\brequire(?:\.resolve)?\(\s*['\"]{esc}{sub}['\"]\s*\)", content):
|
||||
return "require"
|
||||
# Re-exports: `export * from "pkg"`, `export { x } from "pkg"`,
|
||||
# `export type { Foo } from "pkg"`. Multi-line supported.
|
||||
|
|
@ -289,16 +285,12 @@ def classify(pkg: str, file: str, content: str) -> str | None:
|
|||
# segment bounded by a quote / `#` / `?` or a subpath `/`, so
|
||||
# `/node_modules/foo-extra/...` is NOT treated as usage of `foo`.
|
||||
html_pkg = rf"{esc}(?:/[^'\"#?]*)?(?=['\"#?])"
|
||||
if is_html and re.search(
|
||||
rf"<script[^>]*src\s*=\s*['\"][^'\"]*/{html_pkg}", content
|
||||
):
|
||||
if is_html and re.search(rf"<script[^>]*src\s*=\s*['\"][^'\"]*/{html_pkg}", content):
|
||||
return "html_script"
|
||||
if is_html and re.search(rf"<link[^>]*href\s*=\s*['\"][^'\"]*/{html_pkg}", content):
|
||||
return "html_link"
|
||||
# TypeScript triple-slash
|
||||
if is_ts and re.search(
|
||||
rf"///\s*<reference\s+types\s*=\s*['\"]{esc}{sub}['\"]", content
|
||||
):
|
||||
if is_ts and re.search(rf"///\s*<reference\s+types\s*=\s*['\"]{esc}{sub}['\"]", content):
|
||||
return "tsc_triple_slash"
|
||||
# new URL("pkg/...", import.meta.url)
|
||||
if is_script and re.search(rf"\bnew\s+URL\(\s*['\"]{esc}{sub}['\"]", content):
|
||||
|
|
@ -544,19 +536,13 @@ def _next_real_bin(words: list[str], idx: int) -> str | None:
|
|||
if first in {"npx", "pnpx", "bunx"} and idx + 1 < len(words):
|
||||
idx += 1
|
||||
continue
|
||||
if (
|
||||
first in {"pnpm", "yarn"}
|
||||
and idx + 2 < len(words)
|
||||
and words[idx + 1] in {"exec", "dlx"}
|
||||
):
|
||||
if first in {"pnpm", "yarn"} and idx + 2 < len(words) and words[idx + 1] in {"exec", "dlx"}:
|
||||
idx += 2
|
||||
continue
|
||||
|
||||
# 3. Wrapper bin (cross-env, dotenv, etc.). Skip the wrapper's
|
||||
# own flags and any subsequent env-prefix tokens, then re-loop.
|
||||
bin_token = first.removeprefix("./node_modules/.bin/").removeprefix(
|
||||
"node_modules/.bin/"
|
||||
)
|
||||
bin_token = first.removeprefix("./node_modules/.bin/").removeprefix("node_modules/.bin/")
|
||||
if bin_token in _SCRIPT_WRAPPERS and bin_token not in seen_wrappers:
|
||||
seen_wrappers.add(bin_token)
|
||||
idx += 1
|
||||
|
|
@ -585,9 +571,7 @@ def _next_real_bin(words: list[str], idx: int) -> str | None:
|
|||
return None
|
||||
|
||||
|
||||
def scripts_bin_refs(
|
||||
head_pkg: dict, bin_to_pkg: dict[str, str]
|
||||
) -> dict[str, list[str]]:
|
||||
def scripts_bin_refs(head_pkg: dict, bin_to_pkg: dict[str, str]) -> dict[str, list[str]]:
|
||||
"""Return `{package_name: ['scripts.X: cmd', ...]}` listing every
|
||||
package referenced via its bin name in package.json scripts.
|
||||
|
||||
|
|
@ -652,11 +636,7 @@ def tsconfig_compiler_types_refs() -> set[str]:
|
|||
if not isinstance(t, str):
|
||||
continue
|
||||
# `vite/client` resolves to `vite` package.
|
||||
pkg = (
|
||||
t.split("/", 1)[0]
|
||||
if not t.startswith("@")
|
||||
else "/".join(t.split("/", 2)[:2])
|
||||
)
|
||||
pkg = t.split("/", 1)[0] if not t.startswith("@") else "/".join(t.split("/", 2)[:2])
|
||||
out.add(pkg)
|
||||
return out
|
||||
|
||||
|
|
@ -820,9 +800,7 @@ _file_lines_cache: dict[str, list[str]] = {}
|
|||
def _read_file(path: str) -> list[str]:
|
||||
if path not in _file_lines_cache:
|
||||
try:
|
||||
_file_lines_cache[path] = (
|
||||
Path(path).read_text(errors = "replace").splitlines()
|
||||
)
|
||||
_file_lines_cache[path] = Path(path).read_text(errors = "replace").splitlines()
|
||||
except (OSError, UnicodeDecodeError):
|
||||
_file_lines_cache[path] = []
|
||||
return _file_lines_cache[path]
|
||||
|
|
@ -952,18 +930,14 @@ def find_types_runtime_usage(pkg: str, tsc_types: set[str]) -> list[Hit]:
|
|||
|
||||
|
||||
def main() -> int:
|
||||
p = argparse.ArgumentParser(
|
||||
description = __doc__, formatter_class = argparse.RawTextHelpFormatter
|
||||
)
|
||||
p = argparse.ArgumentParser(description = __doc__, formatter_class = argparse.RawTextHelpFormatter)
|
||||
p.add_argument(
|
||||
"--base",
|
||||
default = "origin/main",
|
||||
help = "git ref to diff against (default: origin/main). "
|
||||
"Examples: HEAD~1, main, a-tag, a-sha.",
|
||||
)
|
||||
p.add_argument(
|
||||
"--base-pkg", help = "optional override: read base package.json from this path"
|
||||
)
|
||||
p.add_argument("--base-pkg", help = "optional override: read base package.json from this path")
|
||||
p.add_argument(
|
||||
"--base-lock",
|
||||
help = "optional override: read base package-lock.json from this path. "
|
||||
|
|
@ -1057,9 +1031,7 @@ def main() -> int:
|
|||
print(f" - {w}")
|
||||
print()
|
||||
if missing_imports:
|
||||
print(
|
||||
f"Imports without a matching package.json dep ({len(missing_imports)}):"
|
||||
)
|
||||
print(f"Imports without a matching package.json dep ({len(missing_imports)}):")
|
||||
for file, ln, spec in missing_imports[:20]:
|
||||
print(f" - {file}:{ln} imports '{spec}'")
|
||||
print()
|
||||
|
|
@ -1097,9 +1069,7 @@ def main() -> int:
|
|||
return 1
|
||||
return 0
|
||||
|
||||
print(
|
||||
f"Checking {len(removed)} removed package(s) from studio/frontend/package.json"
|
||||
)
|
||||
print(f"Checking {len(removed)} removed package(s) from studio/frontend/package.json")
|
||||
print(f"Base: {args.base} Head: working tree")
|
||||
print()
|
||||
|
||||
|
|
@ -1127,9 +1097,7 @@ def main() -> int:
|
|||
top = f"node_modules/{name}"
|
||||
top_path = top if top in reachable_paths else None
|
||||
nested = sorted(
|
||||
p
|
||||
for p in reachable_paths
|
||||
if p != top and p.endswith(f"/node_modules/{name}")
|
||||
p for p in reachable_paths if p != top and p.endswith(f"/node_modules/{name}")
|
||||
)
|
||||
return top_path, nested
|
||||
|
||||
|
|
@ -1177,9 +1145,7 @@ def main() -> int:
|
|||
_print_hygiene()
|
||||
|
||||
if failures:
|
||||
print(
|
||||
f"FAIL: {len(failures)} removed package(s) still referenced and not resolvable"
|
||||
)
|
||||
print(f"FAIL: {len(failures)} removed package(s) still referenced and not resolvable")
|
||||
for name, _ in failures:
|
||||
print(f" - {name}")
|
||||
return 1
|
||||
|
|
|
|||
|
|
@ -53,9 +53,7 @@ HIGH = "HIGH"
|
|||
class Finding:
|
||||
__slots__ = ("severity", "name", "version", "kind", "detail")
|
||||
|
||||
def __init__(
|
||||
self, severity: str, name: str, version: str, kind: str, detail: str
|
||||
) -> None:
|
||||
def __init__(self, severity: str, name: str, version: str, kind: str, detail: str) -> None:
|
||||
self.severity = severity
|
||||
self.name = name
|
||||
self.version = version
|
||||
|
|
@ -206,9 +204,7 @@ def diff_new_install_scripts(base_lock: dict, head_lock: dict) -> list[Finding]:
|
|||
continue # pre-existing install-script dep; not in scope
|
||||
name = head[key]
|
||||
# key is "name@version"; rsplit("@", 1) handles scoped names.
|
||||
version = (
|
||||
key[len(name) + 1 :] if key.startswith(name + "@") else "<unversioned>"
|
||||
)
|
||||
version = key[len(name) + 1 :] if key.startswith(name + "@") else "<unversioned>"
|
||||
scripts = _fetch_registry_scripts(name, version)
|
||||
if scripts:
|
||||
detail = "; ".join(f"{h}={cmd!r}" for h, cmd in scripts.items())
|
||||
|
|
@ -238,8 +234,7 @@ def diff_new_install_scripts(base_lock: dict, head_lock: dict) -> list[Finding]:
|
|||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(
|
||||
description = (
|
||||
"Diff two package-lock.json files and refuse any newly-"
|
||||
"added install-script dep."
|
||||
"Diff two package-lock.json files and refuse any newly-added install-script dep."
|
||||
),
|
||||
)
|
||||
parser.add_argument(
|
||||
|
|
|
|||
|
|
@ -1,5 +1,10 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Ensure keyword arguments use spaces around '=', prune redundant pass statements."""
|
||||
"""Ensure keyword arguments use spaces around '=', prune redundant pass statements,
|
||||
drop the blank line after a short indented import block, merge adjacent same-line
|
||||
string literals, normalize def-signature magic commas (pre-ruff) so a def with
|
||||
>= 3 params and a default goes one-per-line while everything else stays
|
||||
collapsible, and collapse a short multi-line assert onto one line (pre-ruff) by
|
||||
stripping the magic trailing comma that holds it open."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
|
@ -123,9 +128,7 @@ def remove_redundant_passes(text: str) -> tuple[str, bool]:
|
|||
lines = text.splitlines(keepends=True)
|
||||
changed = False
|
||||
|
||||
for node in sorted(
|
||||
redundant, key=lambda item: (item.lineno, item.col_offset), reverse=True
|
||||
):
|
||||
for node in sorted(redundant, key=lambda item: (item.lineno, item.col_offset), reverse=True):
|
||||
start = node.lineno - 1
|
||||
end = (node.end_lineno or node.lineno) - 1
|
||||
if start >= len(lines):
|
||||
|
|
@ -160,7 +163,470 @@ def remove_redundant_passes(text: str) -> tuple[str, bool]:
|
|||
return "".join(result_lines), changed
|
||||
|
||||
|
||||
def process_file(path: Path) -> bool:
|
||||
def remove_blank_after_short_import(text: str) -> tuple[str, bool]:
|
||||
"""Drop blank line(s) after an import block in a *small* nested suite.
|
||||
|
||||
Inside an indented suite of <= 3 statements (function/try/if/with/etc., never
|
||||
module level), when a run of consecutive ``import`` / ``from ... import``
|
||||
statements is directly followed -- across one or more blank lines and nothing
|
||||
else -- by another statement in the same suite, remove those blank lines so
|
||||
the import sits next to the code that uses it. A comment in the gap blocks the
|
||||
rule. Removing blank lines never changes the AST, so this is always
|
||||
semantics-preserving.
|
||||
"""
|
||||
try:
|
||||
tree = ast.parse(text)
|
||||
except SyntaxError:
|
||||
return text, False
|
||||
|
||||
lines = text.splitlines(keepends=True)
|
||||
import_types = (ast.Import, ast.ImportFrom)
|
||||
drop: set[int] = set() # 1-based physical line numbers to delete
|
||||
|
||||
def suites_of(node: ast.AST) -> list[list[ast.stmt]]:
|
||||
if isinstance(node, ast.Module):
|
||||
return [] # module-level import spacing is left alone
|
||||
out: list[list[ast.stmt]] = []
|
||||
for attr in ("body", "orelse", "finalbody"):
|
||||
val = getattr(node, attr, None)
|
||||
if isinstance(val, list) and val and all(isinstance(s, ast.stmt) for s in val):
|
||||
out.append(val)
|
||||
return out
|
||||
|
||||
for node in ast.walk(tree):
|
||||
for suite in suites_of(node):
|
||||
if len(suite) > 3: # only small blocks
|
||||
continue
|
||||
i = 0
|
||||
while i < len(suite):
|
||||
if not isinstance(suite[i], import_types):
|
||||
i += 1
|
||||
continue
|
||||
j = i
|
||||
while j + 1 < len(suite) and isinstance(suite[j + 1], import_types):
|
||||
j += 1
|
||||
if j + 1 < len(suite): # an import block followed by another statement
|
||||
last_imp, nxt = suite[j], suite[j + 1]
|
||||
gap = range((last_imp.end_lineno or last_imp.lineno) + 1, nxt.lineno)
|
||||
nums = [n for n in gap if 1 <= n <= len(lines)]
|
||||
if nums and all(lines[n - 1].strip() == "" for n in nums):
|
||||
drop.update(nums)
|
||||
i = j + 1
|
||||
|
||||
if not drop:
|
||||
return text, False
|
||||
kept = [ln for idx, ln in enumerate(lines, start=1) if idx not in drop]
|
||||
return "".join(kept), True
|
||||
|
||||
|
||||
_STRING_TRIVIA = (tokenize.NL, tokenize.NEWLINE, tokenize.COMMENT, tokenize.INDENT, tokenize.DEDENT)
|
||||
|
||||
|
||||
_DEF_MIN_PARAMS_FOR_MULTILINE = 3 # signatures with < this many params stay one line
|
||||
|
||||
|
||||
def _def_specs_by_line(tree: ast.AST) -> dict[int, tuple[int, bool]]:
|
||||
"""Map the line of each def keyword to (param count, has-any-default).
|
||||
|
||||
One def per line, so the line is a stable key. ``*`` / ``/`` markers are not
|
||||
parameters and are not counted. A default exists if any positional default
|
||||
is present or any keyword-only default is not ``None`` (a ``None`` entry in
|
||||
``kw_defaults`` means a required keyword-only arg).
|
||||
"""
|
||||
out: dict[int, tuple[int, bool]] = {}
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
a = node.args
|
||||
count = (
|
||||
len(a.posonlyargs)
|
||||
+ len(a.args)
|
||||
+ len(a.kwonlyargs)
|
||||
+ (1 if a.vararg else 0)
|
||||
+ (1 if a.kwarg else 0)
|
||||
)
|
||||
has_default = bool(a.defaults) or any(d is not None for d in a.kw_defaults)
|
||||
out[node.lineno] = (count, has_default)
|
||||
return out
|
||||
|
||||
|
||||
def normalize_def_trailing_comma(text: str) -> tuple[str, bool]:
|
||||
"""Force a def / async-def signature one-per-line iff it has >= 3 parameters
|
||||
AND at least one default value; otherwise keep it collapsible.
|
||||
|
||||
Rationale: signatures with defaults read better one parameter per line, but
|
||||
only once they are non-trivial (< 3 params always stay on one line). A
|
||||
signature with >= 3 params and a default gets a magic trailing comma added
|
||||
(ruff then wraps it one-per-line regardless of length); every other
|
||||
signature has its trailing comma stripped so ruff collapses it onto one line
|
||||
when it fits (and wraps a genuinely long one by length alone).
|
||||
|
||||
Function-definition parameter lists only, never call sites or collection
|
||||
literals. Parameter counts and defaults come from the AST. Run BEFORE ruff
|
||||
format. Adding or removing a def trailing comma never changes the AST, which
|
||||
is re-checked before returning.
|
||||
"""
|
||||
try:
|
||||
tree = ast.parse(text)
|
||||
toks = list(tokenize.generate_tokens(io.StringIO(text).readline))
|
||||
except (tokenize.TokenError, IndentationError, SyntaxError):
|
||||
return text, False
|
||||
|
||||
specs = _def_specs_by_line(tree)
|
||||
n = len(toks)
|
||||
edits: list[tuple[int, int, str]] = [] # (row, col, "del" | "ins")
|
||||
i = 0
|
||||
while i < n:
|
||||
t = toks[i]
|
||||
if t.type == tokenize.NAME and t.string == "def" and t.start[0] in specs:
|
||||
cnt, has_default = specs[t.start[0]]
|
||||
force_multiline = cnt >= _DEF_MIN_PARAMS_FOR_MULTILINE and has_default
|
||||
j = i + 1
|
||||
while j < n and not (toks[j].type == tokenize.OP and toks[j].string == "("):
|
||||
if toks[j].type == tokenize.NEWLINE:
|
||||
break
|
||||
j += 1
|
||||
if j < n and toks[j].type == tokenize.OP and toks[j].string == "(":
|
||||
depth = 0
|
||||
k = j
|
||||
while k < n:
|
||||
tk = toks[k]
|
||||
if tk.type == tokenize.OP and tk.string == "(":
|
||||
depth += 1
|
||||
elif tk.type == tokenize.OP and tk.string == ")":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
m = k - 1
|
||||
while m > j and toks[m].type in _STRING_TRIVIA:
|
||||
m -= 1
|
||||
last = toks[m]
|
||||
has_comma = last.type == tokenize.OP and last.string == ","
|
||||
empty = m == j # nothing between ( and )
|
||||
if force_multiline and not has_comma and not empty:
|
||||
edits.append((last.end[0], last.end[1], "ins"))
|
||||
elif not force_multiline and has_comma:
|
||||
edits.append((last.start[0], last.start[1], "del"))
|
||||
break
|
||||
k += 1
|
||||
i = k + 1
|
||||
continue
|
||||
i += 1
|
||||
|
||||
if not edits:
|
||||
return text, False
|
||||
|
||||
lines = text.splitlines(keepends=True)
|
||||
for row, col, kind in sorted(edits, reverse=True):
|
||||
ln = lines[row - 1]
|
||||
if kind == "del":
|
||||
if col < len(ln) and ln[col] == ",":
|
||||
lines[row - 1] = ln[:col] + ln[col + 1 :]
|
||||
else: # ins
|
||||
lines[row - 1] = ln[:col] + "," + ln[col:]
|
||||
out = "".join(lines)
|
||||
try:
|
||||
if ast.dump(ast.parse(out)) != ast.dump(ast.parse(text)):
|
||||
return text, False
|
||||
except SyntaxError:
|
||||
return text, False
|
||||
return out, True
|
||||
|
||||
|
||||
def _split_string_token(s: str) -> tuple[str, str, str] | None:
|
||||
"""Split a string literal's source into (prefix, quote, body).
|
||||
|
||||
``prefix`` is the letters before the opening quote (``r``/``f``/``b``/``u``
|
||||
in any case/order), ``quote`` is the opening delimiter (``'``, ``"``,
|
||||
``'''`` or ``\"\"\"``) and ``body`` is everything between the delimiters.
|
||||
Returns ``None`` if ``s`` is not a recognizable string literal.
|
||||
"""
|
||||
i = 0
|
||||
while i < len(s) and s[i] not in ("'", '"'):
|
||||
i += 1
|
||||
if i >= len(s):
|
||||
return None
|
||||
prefix, rest = s[:i], s[i:]
|
||||
for q in ('"""', "'''", '"', "'"):
|
||||
if rest.startswith(q) and rest.endswith(q) and len(rest) >= 2 * len(q):
|
||||
return prefix, q, rest[len(q) : len(rest) - len(q)]
|
||||
return None
|
||||
|
||||
|
||||
# A "piece" is one string literal in source: a plain STRING token, or a whole
|
||||
# f-string spanning FSTRING_START..FSTRING_END. (kind, (row, col0), (row, col1), raw)
|
||||
def _string_pieces(
|
||||
toks: list[tokenize.TokenInfo], lines: list[str]
|
||||
) -> list[tuple[str, tuple[int, int], tuple[int, int], str | None]]:
|
||||
pieces: list[tuple[str, tuple[int, int], tuple[int, int], str | None]] = []
|
||||
n = len(toks)
|
||||
|
||||
def raw_of(start: tuple[int, int], end: tuple[int, int]) -> str | None:
|
||||
if start[0] != end[0]: # only single-physical-line pieces are mergeable
|
||||
return None
|
||||
return lines[start[0] - 1][start[1] : end[1]]
|
||||
|
||||
i = 0
|
||||
while i < n:
|
||||
t = toks[i]
|
||||
if t.type == tokenize.STRING:
|
||||
pieces.append(("str", t.start, t.end, raw_of(t.start, t.end)))
|
||||
i += 1
|
||||
elif t.type == tokenize.FSTRING_START:
|
||||
depth = 0
|
||||
j = i
|
||||
while j < n: # walk to the matching FSTRING_END (f-strings can nest)
|
||||
if toks[j].type == tokenize.FSTRING_START:
|
||||
depth += 1
|
||||
elif toks[j].type == tokenize.FSTRING_END:
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
break
|
||||
j += 1
|
||||
end = toks[j].end
|
||||
pieces.append(("f", t.start, end, raw_of(t.start, end)))
|
||||
i = j + 1
|
||||
else:
|
||||
pieces.append(("other", t.start, t.end, None))
|
||||
i += 1
|
||||
return pieces
|
||||
|
||||
|
||||
def _merge_string_run(pieces: list[tuple[str, str]]) -> str | None:
|
||||
"""Merge a run of adjacent string pieces into one literal's source text.
|
||||
|
||||
``pieces`` is a list of ``(kind, raw_source)`` where kind is ``"str"`` or
|
||||
``"f"``. Rules: bytes are left side-by-side (return ``None``); a run with no
|
||||
f-string merges plain/raw/unicode pieces sharing one prefix+quote by simple
|
||||
body concatenation; a run mixing an f-string with at least one plain string
|
||||
(and no bytes, no raw) folds into a single f-string -- f pieces keep their
|
||||
bodies verbatim and plain pieces have their braces escaped (``{`` -> ``{{``).
|
||||
Runs of only f-strings are left side-by-side. The caller re-checks the file
|
||||
AST and drops the change if it differs, so any subtle case (e.g. ``\\N{...}``)
|
||||
that this would mis-handle is caught and skipped.
|
||||
"""
|
||||
parsed = []
|
||||
for kind, raw in pieces:
|
||||
pqb = _split_string_token(raw)
|
||||
if pqb is None:
|
||||
return None
|
||||
prefix, quote, body = pqb
|
||||
if "b" in prefix.lower():
|
||||
return None # bytes: leave side-by-side
|
||||
parsed.append((kind, prefix, quote, body))
|
||||
if len({p[2] for p in parsed}) != 1:
|
||||
return None # mixed quote style: not a safe textual merge
|
||||
quote = parsed[0][2]
|
||||
if not any(p[0] == "f" for p in parsed):
|
||||
# No f-string: merge plain/raw/unicode sharing one prefix by concatenation.
|
||||
if len({p[1].lower() for p in parsed}) != 1:
|
||||
return None
|
||||
return f"{parsed[0][1]}{quote}{''.join(p[3] for p in parsed)}{quote}"
|
||||
# f-string fold only when a plain string is glued onto an f-string; a run of
|
||||
# only f-strings is left side-by-side (folding long ones would force ruff to
|
||||
# re-wrap the surrounding statement).
|
||||
if all(p[0] == "f" for p in parsed):
|
||||
return None
|
||||
# raw mixed with f is too subtle (backslash + brace escaping) -> skip.
|
||||
if any("r" in p[1].lower() for p in parsed):
|
||||
return None
|
||||
body = "".join(
|
||||
b if kind == "f" else b.replace("{", "{{").replace("}", "}}")
|
||||
for kind, _pfx, _q, b in parsed
|
||||
)
|
||||
return f"f{quote}{body}{quote}"
|
||||
|
||||
|
||||
_LINE_LENGTH = 100 # ruff line-length; an f-fold must not push a statement past it
|
||||
|
||||
|
||||
def _enclosing_stmt(tree: ast.AST, row: int) -> ast.stmt | None:
|
||||
"""The innermost statement whose physical-line span contains ``row``."""
|
||||
best: tuple[ast.stmt, int] | None = None
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.stmt):
|
||||
lo = node.lineno
|
||||
hi = node.end_lineno or lo
|
||||
if lo <= row <= hi and (best is None or hi - lo < best[1]):
|
||||
best = (node, hi - lo)
|
||||
return best[0] if best else None
|
||||
|
||||
|
||||
def _fold_collapses(
|
||||
tree: ast.AST, lines: list[str], row: int, c0: int, c1: int, merged: str
|
||||
) -> bool:
|
||||
"""Whether an f-string fold at ``row[c0:c1]`` -> ``merged`` is safe to apply.
|
||||
|
||||
Only ``assert`` statements wrap awkwardly when a message is folded: ruff
|
||||
parenthesizes the *condition* once ``assert cond, msg`` no longer fits on one
|
||||
line. For every other construct (call argument, ``raise``, assignment, ...) a
|
||||
folded long message wraps acceptably, so the fold is always allowed. For an
|
||||
``assert`` the fold is allowed only when the statement is already one physical
|
||||
line, or its estimated one-line length after folding fits the line length;
|
||||
otherwise the message is left side-by-side.
|
||||
"""
|
||||
stmt = _enclosing_stmt(tree, row)
|
||||
if not isinstance(stmt, ast.Assert):
|
||||
return True
|
||||
lo, hi = stmt.lineno, stmt.end_lineno or stmt.lineno
|
||||
if lo == hi:
|
||||
return True
|
||||
seg = []
|
||||
for k in range(lo, hi + 1):
|
||||
ln = lines[k - 1].rstrip("\n")
|
||||
if k == row:
|
||||
ln = ln[:c0] + merged + ln[c1:]
|
||||
seg.append(ln)
|
||||
indent = len(seg[0]) - len(seg[0].lstrip())
|
||||
# Conservative over-estimate: join continuation lines with a single space
|
||||
# (ruff joins bracketed wraps with none), so borderline cases skip the fold.
|
||||
joined = " ".join(s.strip() for s in seg)
|
||||
return indent + len(joined) <= _LINE_LENGTH
|
||||
|
||||
|
||||
def merge_adjacent_string_literals(text: str) -> tuple[str, bool]:
|
||||
"""Merge a run of adjacent string literals on ONE physical line into a single
|
||||
literal (the ``"a" "b"`` form ruff emits when it collapses an implicit
|
||||
concatenation). Plain/raw/unicode runs merge by concatenation; a run mixing
|
||||
an f-string with a plain string folds into one f-string (plain parts' braces
|
||||
escaped) -- but only when the statement still fits on one line, so a long
|
||||
message is left side-by-side rather than forcing the statement to re-wrap.
|
||||
Runs of only f-strings, and bytes, are left side-by-side. The whole file's AST
|
||||
is re-checked and the change is dropped if it would differ, so the transform
|
||||
can never change meaning.
|
||||
"""
|
||||
try:
|
||||
toks = list(tokenize.generate_tokens(io.StringIO(text).readline))
|
||||
tree = ast.parse(text)
|
||||
except (tokenize.TokenError, IndentationError, SyntaxError):
|
||||
return text, False
|
||||
|
||||
lines = text.splitlines(keepends=True)
|
||||
pieces = _string_pieces(toks, lines)
|
||||
|
||||
# Group consecutive mergeable pieces (str/f, single line, same physical line).
|
||||
runs: list[list[tuple[str, tuple[int, int], tuple[int, int], str]]] = []
|
||||
cur: list[tuple[str, tuple[int, int], tuple[int, int], str]] = []
|
||||
for kind, start, end, raw in pieces:
|
||||
if kind in ("str", "f") and raw is not None:
|
||||
if cur and cur[-1][2][0] != start[0]:
|
||||
if len(cur) >= 2:
|
||||
runs.append(cur)
|
||||
cur = []
|
||||
cur.append((kind, start, end, raw))
|
||||
else:
|
||||
if len(cur) >= 2:
|
||||
runs.append(cur)
|
||||
cur = []
|
||||
if len(cur) >= 2:
|
||||
runs.append(cur)
|
||||
if not runs:
|
||||
return text, False
|
||||
|
||||
edits = []
|
||||
for run in runs:
|
||||
merged = _merge_string_run([(kind, raw) for kind, _s, _e, raw in run])
|
||||
if merged is None:
|
||||
continue
|
||||
row, c0, c1 = run[0][1][0], run[0][1][1], run[-1][2][1]
|
||||
# An f-string fold must not push its statement onto extra lines; a plain
|
||||
# concatenation always collapses cleanly so it skips this check.
|
||||
if any(kind == "f" for kind, _s, _e, _r in run) and not _fold_collapses(
|
||||
tree, lines, row, c0, c1, merged
|
||||
):
|
||||
continue
|
||||
edits.append((row, c0, c1, merged))
|
||||
if not edits:
|
||||
return text, False
|
||||
|
||||
for row, c0, c1, repl in sorted(edits, key=lambda e: (e[0], e[1]), reverse=True):
|
||||
ln = lines[row - 1]
|
||||
lines[row - 1] = ln[:c0] + repl + ln[c1:]
|
||||
out = "".join(lines)
|
||||
try:
|
||||
if ast.dump(ast.parse(text)) != ast.dump(ast.parse(out)):
|
||||
return text, False
|
||||
except SyntaxError:
|
||||
return text, False
|
||||
return out, True
|
||||
|
||||
|
||||
def collapse_short_asserts(text: str) -> tuple[str, bool]:
|
||||
"""Collapse a multi-line ``assert`` onto one line when it would fit.
|
||||
|
||||
An ``assert`` is often kept multi-line only by a magic trailing comma inside
|
||||
a collection / call / tuple-message (``assert x == {\"a\": 1,}`` written
|
||||
across lines). When the whole statement's estimated one-line length fits the
|
||||
line length, strip those trailing commas (the comma before a ``)`` / ``]`` /
|
||||
``}``) so ruff joins it back onto one line on the following format pass.
|
||||
|
||||
Run BEFORE ruff format. Skips any assert that contains a comment (a comment
|
||||
forces ruff to keep it multi-line, which would oscillate). Stripping a
|
||||
trailing comma is non-semantic except for a one-element tuple ``(x,)``; the
|
||||
file AST is re-checked and any assert whose strip would change it is left
|
||||
alone.
|
||||
"""
|
||||
try:
|
||||
tree = ast.parse(text)
|
||||
toks = list(tokenize.generate_tokens(io.StringIO(text).readline))
|
||||
except (tokenize.TokenError, IndentationError, SyntaxError):
|
||||
return text, False
|
||||
|
||||
lines = text.splitlines(keepends=True)
|
||||
multiline = [
|
||||
(n.lineno, n.end_lineno)
|
||||
for n in ast.walk(tree)
|
||||
if isinstance(n, ast.Assert) and (n.end_lineno or n.lineno) > n.lineno
|
||||
]
|
||||
if not multiline:
|
||||
return text, False
|
||||
|
||||
comment_rows = {t.start[0] for t in toks if t.type == tokenize.COMMENT}
|
||||
|
||||
targets = [] # (lo, hi) spans whose one-line form fits and have no comment
|
||||
for lo, hi in multiline:
|
||||
if any(lo <= r <= hi for r in comment_rows):
|
||||
continue # a comment would keep ruff multi-line -> never collapses
|
||||
seg = [lines[k].rstrip("\n") for k in range(lo - 1, hi)]
|
||||
indent = len(seg[0]) - len(seg[0].lstrip())
|
||||
# Over-estimate (join with a space; keep the comma) so a "fits" verdict
|
||||
# is always at least as long as ruff's real one-line output -> no fight.
|
||||
if indent + len(" ".join(s.strip() for s in seg)) <= _LINE_LENGTH:
|
||||
targets.append((lo, hi))
|
||||
if not targets:
|
||||
return text, False
|
||||
|
||||
# Trailing commas (a ',' whose next significant token is a closer), grouped
|
||||
# by the target assert they belong to.
|
||||
sig = [t for t in toks if t.type not in _STRING_TRIVIA]
|
||||
by_target: dict[tuple[int, int], list[tuple[int, int]]] = defaultdict(list)
|
||||
for i, t in enumerate(sig):
|
||||
if t.type == tokenize.OP and t.string == ",":
|
||||
nxt = sig[i + 1] if i + 1 < len(sig) else None
|
||||
if nxt and nxt.type == tokenize.OP and nxt.string in (")", "]", "}"):
|
||||
for lo, hi in targets:
|
||||
if lo <= t.start[0] <= hi:
|
||||
by_target[(lo, hi)].append(t.start)
|
||||
break
|
||||
if not by_target:
|
||||
return text, False
|
||||
|
||||
base_dump = ast.dump(tree)
|
||||
working = lines[:]
|
||||
changed = False
|
||||
for positions in by_target.values(): # apply per assert; skip any that break AST
|
||||
trial = working[:]
|
||||
for row, col in sorted(positions, reverse=True):
|
||||
ln = trial[row - 1]
|
||||
if col < len(ln) and ln[col] == ",":
|
||||
trial[row - 1] = ln[:col] + ln[col + 1 :]
|
||||
try:
|
||||
if ast.dump(ast.parse("".join(trial))) == base_dump:
|
||||
working, changed = trial, True
|
||||
except SyntaxError:
|
||||
pass
|
||||
return ("".join(working), True) if changed else (text, False)
|
||||
|
||||
|
||||
def process_file(path: Path, pre: bool = False) -> bool:
|
||||
try:
|
||||
with tokenize.open(path) as handle:
|
||||
original = handle.read()
|
||||
|
|
@ -169,9 +635,23 @@ def process_file(path: Path) -> bool:
|
|||
print(f"Failed to read {path}: {exc}", file=sys.stderr)
|
||||
return False
|
||||
|
||||
if pre:
|
||||
# Pre-ruff: normalize def-signature magic commas (>=3 params + a default
|
||||
# add so ruff forces one-per-line; everything else strips so ruff
|
||||
# collapses), and strip the magic trailing comma from a short multi-line
|
||||
# assert so ruff joins it onto one line. Everything else runs post-ruff.
|
||||
updated, normalized = normalize_def_trailing_comma(original)
|
||||
updated, collapsed = collapse_short_asserts(updated)
|
||||
if normalized or collapsed:
|
||||
_atomic_write_text(path, updated, encoding)
|
||||
return True
|
||||
return False
|
||||
|
||||
updated, changed = enforce_spacing(original)
|
||||
updated, blanked = remove_blank_after_short_import(updated)
|
||||
updated, merged = merge_adjacent_string_literals(updated)
|
||||
updated, removed = remove_redundant_passes(updated)
|
||||
if changed or removed:
|
||||
if changed or blanked or merged or removed:
|
||||
_atomic_write_text(path, updated, encoding)
|
||||
return True
|
||||
return False
|
||||
|
|
@ -180,6 +660,11 @@ def process_file(path: Path) -> bool:
|
|||
def main(argv: list[str]) -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("files", nargs="+", help="Python files to fix")
|
||||
parser.add_argument(
|
||||
"--pre",
|
||||
action="store_true",
|
||||
help="pre-ruff pass: normalize def-signature commas + collapse short multi-line asserts",
|
||||
)
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
touched: list[Path] = []
|
||||
|
|
@ -192,7 +677,7 @@ def main(argv: list[str]) -> int:
|
|||
continue
|
||||
if not path.exists() or path.is_dir():
|
||||
continue
|
||||
if process_file(path):
|
||||
if process_file(path, pre=args.pre):
|
||||
touched.append(path)
|
||||
|
||||
if touched:
|
||||
|
|
|
|||
|
|
@ -47,9 +47,7 @@ from pathlib import Path
|
|||
try:
|
||||
import yaml
|
||||
except ImportError:
|
||||
print(
|
||||
"ERROR: PyYAML is required. Install with 'pip install pyyaml'", file = sys.stderr
|
||||
)
|
||||
print("ERROR: PyYAML is required. Install with 'pip install pyyaml'", file = sys.stderr)
|
||||
sys.exit(2)
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
|
|
@ -153,9 +151,7 @@ def main() -> int:
|
|||
)
|
||||
|
||||
if findings:
|
||||
print(
|
||||
"Workflow trigger lint failed with the following issues:", file = sys.stderr
|
||||
)
|
||||
print("Workflow trigger lint failed with the following issues:", file = sys.stderr)
|
||||
for f in findings:
|
||||
print(f" - {f}", file = sys.stderr)
|
||||
return 1
|
||||
|
|
|
|||
|
|
@ -541,9 +541,7 @@ def audit_npm_lockfile(path: Path) -> list[Finding]:
|
|||
path = str(path),
|
||||
package = key,
|
||||
kind = "blocked-known-malicious",
|
||||
detail = (
|
||||
f"{pkg_name}@{version} is on the " "BLOCKED_NPM_VERSIONS list"
|
||||
),
|
||||
detail = (f"{pkg_name}@{version} is on the BLOCKED_NPM_VERSIONS list"),
|
||||
)
|
||||
)
|
||||
|
||||
|
|
@ -765,10 +763,7 @@ def main(argv: list[str] | None = None) -> int:
|
|||
"--cargo-lockfile",
|
||||
action = "append",
|
||||
default = None,
|
||||
help = (
|
||||
"Path to a Cargo.lock (repeatable). "
|
||||
"Default: studio/src-tauri/Cargo.lock."
|
||||
),
|
||||
help = ("Path to a Cargo.lock (repeatable). Default: studio/src-tauri/Cargo.lock."),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--strict",
|
||||
|
|
|
|||
|
|
@ -186,9 +186,7 @@ def convert_cell_to_python(source: str, *, allow_shell: bool = True) -> str:
|
|||
cmd_lines.append(lines[i].strip())
|
||||
full_cmd = "\n".join(cmd_lines)
|
||||
|
||||
result.extend(
|
||||
_emit_shell_command(indent, full_cmd, allow_shell = allow_shell)
|
||||
)
|
||||
result.extend(_emit_shell_command(indent, full_cmd, allow_shell = allow_shell))
|
||||
|
||||
# %cd path -> os.chdir(path)
|
||||
elif stripped.startswith("%cd "):
|
||||
|
|
@ -313,9 +311,7 @@ def convert_notebook_to_script(
|
|||
# Generate output filename
|
||||
output_filename = filename.replace(".ipynb", ".py")
|
||||
# Clean up filename
|
||||
output_filename = (
|
||||
output_filename.replace("(", "").replace(")", "").replace("-", "_")
|
||||
)
|
||||
output_filename = output_filename.replace("(", "").replace(")", "").replace("-", "_")
|
||||
|
||||
# Add output directory if specified
|
||||
if output_dir:
|
||||
|
|
@ -337,9 +333,7 @@ def convert_notebook_to_script(
|
|||
def main():
|
||||
import argparse
|
||||
|
||||
class Formatter(
|
||||
argparse.ArgumentDefaultsHelpFormatter, argparse.RawDescriptionHelpFormatter
|
||||
):
|
||||
class Formatter(argparse.ArgumentDefaultsHelpFormatter, argparse.RawDescriptionHelpFormatter):
|
||||
pass
|
||||
|
||||
parser = argparse.ArgumentParser(
|
||||
|
|
@ -353,12 +347,8 @@ Examples:
|
|||
python notebook_to_python.py https://github.com/unslothai/notebooks/blob/main/nb/Oute_TTS_(1B).ipynb
|
||||
""",
|
||||
)
|
||||
parser.add_argument(
|
||||
"notebooks", nargs = "+", help = "Notebook files or URLs to convert."
|
||||
)
|
||||
parser.add_argument(
|
||||
"-o", "--output", dest = "output_dir", default = ".", help = "Output directory."
|
||||
)
|
||||
parser.add_argument("notebooks", nargs = "+", help = "Notebook files or URLs to convert.")
|
||||
parser.add_argument("-o", "--output", dest = "output_dir", default = ".", help = "Output directory.")
|
||||
# Default True for backwards compatibility: existing Colab notebooks
|
||||
# routinely use pipes / redirection / interpolation in `!cmd` lines
|
||||
# and the converted script needs to keep working. Operators who
|
||||
|
|
|
|||
|
|
@ -92,9 +92,7 @@ COLAB_ORACLE_FILES: dict[str, str] = {
|
|||
"apt-list-gpu.txt": "colab_apt_list.gpu.txt",
|
||||
"os-info-gpu.txt": "colab_os_info.gpu.txt",
|
||||
}
|
||||
COLAB_ORACLE_BASE_URL = (
|
||||
"https://raw.githubusercontent.com/googlecolab/backend-info/main/"
|
||||
)
|
||||
COLAB_ORACLE_BASE_URL = "https://raw.githubusercontent.com/googlecolab/backend-info/main/"
|
||||
|
||||
# ----- Compat tables. PRs add rows as new releases land. ----- #
|
||||
|
||||
|
|
@ -195,9 +193,7 @@ def install_cells(nb: dict[str, Any]) -> list[tuple[int, str]]:
|
|||
if first and first[0].strip().startswith("%%capture"):
|
||||
out.append((i, src))
|
||||
continue
|
||||
if re.search(
|
||||
r"^[ \t]*!\s*(uv\s+)?pip\s+(install|uninstall)\b", src, re.MULTILINE
|
||||
):
|
||||
if re.search(r"^[ \t]*!\s*(uv\s+)?pip\s+(install|uninstall)\b", src, re.MULTILINE):
|
||||
out.append((i, src))
|
||||
return out
|
||||
|
||||
|
|
@ -331,9 +327,7 @@ def parse_pip_line(line: str, line_no: int = 0) -> PipInvocation | None:
|
|||
if t in ("install", "uninstall"):
|
||||
continue
|
||||
packages.append(t)
|
||||
return PipInvocation(
|
||||
tool = tool, flags = flags, packages = packages, raw = line, line_no = line_no
|
||||
)
|
||||
return PipInvocation(tool = tool, flags = flags, packages = packages, raw = line, line_no = line_no)
|
||||
|
||||
|
||||
def _glue_line_continuations(text: str) -> list[tuple[int, str]]:
|
||||
|
|
@ -418,9 +412,7 @@ def pypi_metadata(name: str, version: str) -> dict[str, Any] | None:
|
|||
return data
|
||||
|
||||
|
||||
def transitive_constraint(
|
||||
name: str, version: str, target: str
|
||||
) -> tuple[str | None, list[str]]:
|
||||
def transitive_constraint(name: str, version: str, target: str) -> tuple[str | None, list[str]]:
|
||||
"""Return (raw_specifier_string_or_None, list_of_(op,version) tuples)
|
||||
for the constraint that `name==version` places on `target`.
|
||||
"""
|
||||
|
|
@ -501,10 +493,7 @@ def resolved_set(install_cell: str, colab: dict[str, str]) -> dict[str, str]:
|
|||
out[sp.name] = ver
|
||||
pinned.add(sp.name)
|
||||
elif op == "<=" and sp.name not in pinned:
|
||||
if (
|
||||
sp.name not in upper_bounds
|
||||
or cmp_versions(ver, upper_bounds[sp.name]) < 0
|
||||
):
|
||||
if sp.name not in upper_bounds or cmp_versions(ver, upper_bounds[sp.name]) < 0:
|
||||
upper_bounds[sp.name] = ver
|
||||
# Apply upper bounds where Colab's preinstall violates them.
|
||||
for name, ub in upper_bounds.items():
|
||||
|
|
@ -519,9 +508,7 @@ def resolved_set(install_cell: str, colab: dict[str, str]) -> dict[str, str]:
|
|||
# ----- Rules ----- #
|
||||
|
||||
|
||||
def rule_inst_001_git_plus(
|
||||
install_cell: str, file: str, cell_idx: int
|
||||
) -> list[Finding]:
|
||||
def rule_inst_001_git_plus(install_cell: str, file: str, cell_idx: int) -> list[Finding]:
|
||||
findings: list[Finding] = []
|
||||
for inv in iter_pip_invocations(install_cell):
|
||||
if any("git+" in p for p in inv.packages) or "git+" in inv.raw:
|
||||
|
|
@ -714,9 +701,7 @@ def rule_inst_005_transformers_tokenizers(
|
|||
_RE_DOUBLE_BANG = re.compile(r"^[ \t]*!{2,}\s*pip\b", re.MULTILINE)
|
||||
|
||||
|
||||
def rule_inst_006_double_bang(
|
||||
install_cell: str, file: str, cell_idx: int
|
||||
) -> list[Finding]:
|
||||
def rule_inst_006_double_bang(install_cell: str, file: str, cell_idx: int) -> list[Finding]:
|
||||
findings: list[Finding] = []
|
||||
for m in _RE_DOUBLE_BANG.finditer(install_cell):
|
||||
line_no = install_cell.count("\n", 0, m.start()) + 1
|
||||
|
|
@ -813,9 +798,7 @@ POLICY_CLAUSES_DEFAULT = [
|
|||
]
|
||||
|
||||
|
||||
def extract_policy_clauses(
|
||||
update_script: pathlib.Path,
|
||||
) -> list[tuple[str, re.Pattern[str], Any]]:
|
||||
def extract_policy_clauses(update_script: pathlib.Path) -> list[tuple[str, re.Pattern[str], Any]]:
|
||||
"""Best-effort: scan update_all_notebooks.py for canonical phrases used by
|
||||
multiple templates. Falls back to POLICY_CLAUSES_DEFAULT.
|
||||
|
||||
|
|
@ -879,11 +862,7 @@ def cmd_drift(args: argparse.Namespace) -> int:
|
|||
print(f"FAIL: {update_script} not found", file = sys.stderr)
|
||||
return 2
|
||||
# Stash any pre-existing dirty state, run the updater, diff, restore.
|
||||
head = (
|
||||
subprocess.check_output(["git", "rev-parse", "HEAD"], cwd = nbdir)
|
||||
.decode()
|
||||
.strip()
|
||||
)
|
||||
head = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd = nbdir).decode().strip()
|
||||
subprocess.run(
|
||||
["git", "-C", str(nbdir), "stash", "--include-untracked"],
|
||||
check = False,
|
||||
|
|
@ -990,9 +969,7 @@ def cmd_convert(args: argparse.Namespace) -> int:
|
|||
hint = proc.stderr[-200:].strip(),
|
||||
)
|
||||
)
|
||||
print(
|
||||
f"converted {len(notebooks) - len(failed)}/{len(notebooks)} notebooks to {out}"
|
||||
)
|
||||
print(f"converted {len(notebooks) - len(failed)}/{len(notebooks)} notebooks to {out}")
|
||||
_emit(failed)
|
||||
return 0 if not failed else 1
|
||||
|
||||
|
|
@ -1002,11 +979,7 @@ def cmd_convert(args: argparse.Namespace) -> int:
|
|||
|
||||
def cmd_lint(args: argparse.Namespace) -> int:
|
||||
nbdir = pathlib.Path(args.notebooks_dir).resolve()
|
||||
colab_path = (
|
||||
pathlib.Path(args.colab_pin).resolve()
|
||||
if args.colab_pin
|
||||
else COLAB_FALLBACK_FILE
|
||||
)
|
||||
colab_path = pathlib.Path(args.colab_pin).resolve() if args.colab_pin else COLAB_FALLBACK_FILE
|
||||
colab = parse_pip_freeze(colab_path)
|
||||
if not colab:
|
||||
print(
|
||||
|
|
@ -1049,13 +1022,9 @@ def cmd_lint(args: argparse.Namespace) -> int:
|
|||
first_cell = cells[0][0] if cells else None
|
||||
findings += rule_inst_003_peft_torchao(merged, oracle, rel, first_cell)
|
||||
findings += rule_inst_004_torchcodec_torch(merged, oracle, rel, first_cell)
|
||||
findings += rule_inst_005_transformers_tokenizers(
|
||||
merged, oracle, rel, first_cell
|
||||
)
|
||||
findings += rule_inst_005_transformers_tokenizers(merged, oracle, rel, first_cell)
|
||||
if not args.no_pypi:
|
||||
findings += rule_inst_002_no_deps_transitive(
|
||||
merged, oracle, rel, first_cell
|
||||
)
|
||||
findings += rule_inst_002_no_deps_transitive(merged, oracle, rel, first_cell)
|
||||
findings += scan_user_cells(nb, rel)
|
||||
_emit(findings)
|
||||
return 0 if not any(f.severity == "error" for f in findings) else 1
|
||||
|
|
@ -1232,9 +1201,7 @@ def cmd_colab_diff(args: argparse.Namespace) -> int:
|
|||
print(f"::warning::colab-diff: could not fetch {url}: {e}")
|
||||
continue
|
||||
if not snap_path.exists():
|
||||
print(
|
||||
f"::warning::colab-diff: no committed snapshot at {snap_path}; skipping"
|
||||
)
|
||||
print(f"::warning::colab-diff: no committed snapshot at {snap_path}; skipping")
|
||||
continue
|
||||
snapshot_text = snap_path.read_text(encoding = "utf-8", errors = "replace")
|
||||
parser = _COLAB_ORACLE_PARSERS[upstream_name]
|
||||
|
|
|
|||
|
|
@ -1,5 +1,7 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Run `ruff format` followed by kwarg spacing enforcement."""
|
||||
"""Run a pre-pass (normalize def-signature magic commas + collapse short
|
||||
multi-line asserts), then `ruff format`, then the kwarg-spacing / import /
|
||||
string-merge post-pass."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
|
@ -15,12 +17,22 @@ def main(argv: list[str]) -> int:
|
|||
if not files:
|
||||
return 0
|
||||
|
||||
spacing_script = HERE / "enforce_kwargs_spacing.py"
|
||||
|
||||
# Pre-ruff: normalize def-signature trailing commas (>=3 params with a
|
||||
# default -> one-per-line; everything else collapsible) and strip the magic
|
||||
# comma from a short multi-line assert, so ruff wraps signatures accordingly
|
||||
# and joins the assert back onto one line.
|
||||
pre_cmd = [sys.executable, str(spacing_script), "--pre", *files]
|
||||
pre_proc = subprocess.run(pre_cmd)
|
||||
if pre_proc.returncode != 0:
|
||||
return pre_proc.returncode
|
||||
|
||||
ruff_cmd = [sys.executable, "-m", "ruff", "format", *files]
|
||||
ruff_proc = subprocess.run(ruff_cmd)
|
||||
if ruff_proc.returncode != 0:
|
||||
return ruff_proc.returncode
|
||||
|
||||
spacing_script = HERE / "enforce_kwargs_spacing.py"
|
||||
spacing_cmd = [sys.executable, str(spacing_script), *files]
|
||||
spacing_proc = subprocess.run(spacing_cmd)
|
||||
return spacing_proc.returncode
|
||||
|
|
|
|||
|
|
@ -823,8 +823,7 @@ def download_tarball(
|
|||
written += len(chunk)
|
||||
if written > max_bytes:
|
||||
return dest, (
|
||||
f"download exceeded cap {max_bytes} bytes "
|
||||
f"after {written} bytes"
|
||||
f"download exceeded cap {max_bytes} bytes " f"after {written} bytes"
|
||||
)
|
||||
h.update(chunk)
|
||||
out.write(chunk)
|
||||
|
|
@ -926,11 +925,7 @@ def safe_extract(
|
|||
# get the generous binary cap. We bound BOTH cases.
|
||||
header = src.read(16)
|
||||
is_binary = _looks_binary(name, header)
|
||||
file_cap = (
|
||||
HARD_MAX_BINARY_FILE_BYTES
|
||||
if is_binary
|
||||
else HARD_MAX_TEXT_FILE_BYTES
|
||||
)
|
||||
file_cap = HARD_MAX_BINARY_FILE_BYTES if is_binary else HARD_MAX_TEXT_FILE_BYTES
|
||||
if declared > file_cap:
|
||||
return (
|
||||
f"member {name!r} declared size {declared} > "
|
||||
|
|
@ -963,7 +958,11 @@ def safe_extract(
|
|||
# ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _evidence(text: str, pat: re.Pattern, max_chars: int = 200) -> str:
|
||||
def _evidence(
|
||||
text: str,
|
||||
pat: re.Pattern,
|
||||
max_chars: int = 200,
|
||||
) -> str:
|
||||
m = pat.search(text)
|
||||
if not m:
|
||||
return ""
|
||||
|
|
@ -978,11 +977,7 @@ def _evidence(text: str, pat: re.Pattern, max_chars: int = 200) -> str:
|
|||
LIFECYCLE_HOOKS = ("preinstall", "install", "postinstall", "prepare")
|
||||
|
||||
|
||||
def scan_package_json(
|
||||
pkg: PackageEntry,
|
||||
rel: str,
|
||||
text: str,
|
||||
) -> list[Finding]:
|
||||
def scan_package_json(pkg: PackageEntry, rel: str, text: str) -> list[Finding]:
|
||||
findings: list[Finding] = []
|
||||
try:
|
||||
meta = json.loads(text)
|
||||
|
|
@ -1056,9 +1051,7 @@ def scan_package_json(
|
|||
if isinstance(opt, dict):
|
||||
for k, v in opt.items():
|
||||
if isinstance(v, str) and (
|
||||
v.startswith("github:")
|
||||
or v.startswith("git+")
|
||||
or v.startswith("git://")
|
||||
v.startswith("github:") or v.startswith("git+") or v.startswith("git://")
|
||||
):
|
||||
findings.append(
|
||||
Finding(
|
||||
|
|
@ -1117,11 +1110,7 @@ def _host_in_outbound_context(text: str, host: str) -> bool:
|
|||
return False
|
||||
|
||||
|
||||
def scan_text_blob(
|
||||
pkg: PackageEntry,
|
||||
rel: str,
|
||||
text: str,
|
||||
) -> list[Finding]:
|
||||
def scan_text_blob(pkg: PackageEntry, rel: str, text: str) -> list[Finding]:
|
||||
findings: list[Finding] = []
|
||||
|
||||
# IOC substrings (literal, case-sensitive).
|
||||
|
|
@ -1190,10 +1179,7 @@ def scan_text_blob(
|
|||
filename = rel,
|
||||
pattern = "js-fetch-eval",
|
||||
evidence = _evidence(text, _JS_FETCH_EVAL),
|
||||
detail = (
|
||||
"Function/eval against base64-decoded payload "
|
||||
"(obfuscated dropper shape)"
|
||||
),
|
||||
detail = ("Function/eval against base64-decoded payload (obfuscated dropper shape)"),
|
||||
)
|
||||
)
|
||||
if _JS_ENV_TOKEN.search(text):
|
||||
|
|
@ -1247,10 +1233,7 @@ _TEXT_SUFFIXES = (
|
|||
)
|
||||
|
||||
|
||||
def scan_extracted_tree(
|
||||
pkg: PackageEntry,
|
||||
root: Path,
|
||||
) -> list[Finding]:
|
||||
def scan_extracted_tree(pkg: PackageEntry, root: Path) -> list[Finding]:
|
||||
findings: list[Finding] = []
|
||||
for path in sorted(root.rglob("*")):
|
||||
if not path.is_file():
|
||||
|
|
@ -1304,10 +1287,7 @@ def scan_extracted_tree(
|
|||
# ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def scan_one(
|
||||
pkg: PackageEntry,
|
||||
workspace: Path,
|
||||
) -> tuple[list[Finding], str | None]:
|
||||
def scan_one(pkg: PackageEntry, workspace: Path) -> tuple[list[Finding], str | None]:
|
||||
"""Download + extract + scan a single package. Cleans up its dir.
|
||||
|
||||
Returns (findings, error). `error` is non-None only on hard
|
||||
|
|
@ -1444,8 +1424,7 @@ def main(argv: list[str] | None = None) -> int:
|
|||
if hard_errors or blocking:
|
||||
if blocking:
|
||||
print(
|
||||
f"\n[scan-npm] FAIL: {len(blocking)} finding(s) "
|
||||
f"at or above {threshold}",
|
||||
f"\n[scan-npm] FAIL: {len(blocking)} finding(s) " f"at or above {threshold}",
|
||||
file = sys.stderr,
|
||||
)
|
||||
return 1
|
||||
|
|
|
|||
|
|
@ -86,8 +86,7 @@ RE_SUBPROCESS = re.compile(
|
|||
|
||||
# Encoding / obfuscation
|
||||
RE_BASE64 = re.compile(
|
||||
r"\bbase64\s*\.\s*(b64decode|decodebytes|b32decode|b16decode)\b"
|
||||
r"|\bcodecs\s*\.\s*decode\b",
|
||||
r"\bbase64\s*\.\s*(b64decode|decodebytes|b32decode|b16decode)\b|\bcodecs\s*\.\s*decode\b",
|
||||
)
|
||||
|
||||
# exec / eval
|
||||
|
|
@ -299,9 +298,7 @@ RE_CRYPTO_THEFT = re.compile(
|
|||
RE_PTH_IMPORT = re.compile(r"^\s*import\s+", re.MULTILINE)
|
||||
|
||||
# openssl CLI invocations via subprocess (encrypted exfiltration)
|
||||
RE_OPENSSL_CLI = re.compile(
|
||||
r"\bopenssl\s+(enc|rand|rsautl|pkeyutl|genrsa|dgst|s_client)\b"
|
||||
)
|
||||
RE_OPENSSL_CLI = re.compile(r"\bopenssl\s+(enc|rand|rsautl|pkeyutl|genrsa|dgst|s_client)\b")
|
||||
|
||||
# Write to /tmp then execute (staged dropper)
|
||||
RE_TEMP_EXEC = re.compile(
|
||||
|
|
@ -962,7 +959,11 @@ def check_py_file(content: str, filename: str, package: str) -> list[Finding]:
|
|||
return findings
|
||||
|
||||
|
||||
def _extract_evidence(content: str, pattern: re.Pattern, max_matches: int = 3) -> str:
|
||||
def _extract_evidence(
|
||||
content: str,
|
||||
pattern: re.Pattern,
|
||||
max_matches: int = 3,
|
||||
) -> str:
|
||||
"""Pull matching lines as evidence snippets."""
|
||||
lines = content.splitlines()
|
||||
matches = []
|
||||
|
|
@ -1266,15 +1267,13 @@ def iter_archive_files(archive_path: str):
|
|||
# have historically dereferenced them on extract.
|
||||
if member.issym() or member.islnk():
|
||||
print(
|
||||
f" [WARN] {path.name}: refused link member "
|
||||
f"{member.name!r}",
|
||||
f" [WARN] {path.name}: refused link member " f"{member.name!r}",
|
||||
file = sys.stderr,
|
||||
)
|
||||
continue
|
||||
if member.isdev() or member.isfifo():
|
||||
print(
|
||||
f" [WARN] {path.name}: refused special member "
|
||||
f"{member.name!r}",
|
||||
f" [WARN] {path.name}: refused special member " f"{member.name!r}",
|
||||
file = sys.stderr,
|
||||
)
|
||||
continue
|
||||
|
|
@ -1379,9 +1378,7 @@ def scan_archive(archive_path: str, package: str) -> list[Finding]:
|
|||
_RE_PYPI_SPEC_VERSION = re.compile(r"==\s*([A-Za-z0-9_.\-+!]+)")
|
||||
|
||||
|
||||
def _check_blocked_pypi_versions(
|
||||
specs: list[str],
|
||||
) -> tuple[list[str], list[Finding]]:
|
||||
def _check_blocked_pypi_versions(specs: list[str]) -> tuple[list[str], list[Finding]]:
|
||||
"""Filter ``specs`` against ``BLOCKED_PYPI_VERSIONS``.
|
||||
|
||||
Returns ``(safe_specs, findings)``. Each blocked spec emits a CRITICAL
|
||||
|
|
@ -1502,9 +1499,7 @@ def download_packages(
|
|||
env = env,
|
||||
)
|
||||
if proc.returncode != 0:
|
||||
msg = (
|
||||
f"pip download (with deps) failed: " f"{proc.stderr.strip()[:500]}"
|
||||
)
|
||||
msg = f"pip download (with deps) failed: " f"{proc.stderr.strip()[:500]}"
|
||||
print(f" [ERROR] {msg}", file = sys.stderr)
|
||||
download_errors.append(msg)
|
||||
except subprocess.TimeoutExpired:
|
||||
|
|
@ -1547,10 +1542,7 @@ def download_packages(
|
|||
env = env,
|
||||
)
|
||||
if proc.returncode != 0:
|
||||
msg = (
|
||||
f"pip download failed for {spec}: "
|
||||
f"{proc.stderr.strip()[:500]}"
|
||||
)
|
||||
msg = f"pip download failed for {spec}: " f"{proc.stderr.strip()[:500]}"
|
||||
print(f" [ERROR] {msg}", file = sys.stderr)
|
||||
download_errors.append(msg)
|
||||
continue
|
||||
|
|
@ -1579,9 +1571,7 @@ def _extract_pkg_name(spec: str) -> str:
|
|||
"""Extract the package name from a pip spec string."""
|
||||
m = _RE_NAME.match(spec)
|
||||
return (
|
||||
m.group(1)
|
||||
if m
|
||||
else spec.split("==")[0].split(">=")[0].split("<=")[0].split("[")[0].strip()
|
||||
m.group(1) if m else spec.split("==")[0].split(">=")[0].split("<=")[0].split("[")[0].strip()
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -1917,11 +1907,7 @@ def update_req_file(filepath: str, updates: dict[int, str]) -> None:
|
|||
raise
|
||||
|
||||
|
||||
def _run_fix(
|
||||
critical_pkgs: set[str],
|
||||
entries: list[dict],
|
||||
max_search: int,
|
||||
) -> None:
|
||||
def _run_fix(critical_pkgs: set[str], entries: list[dict], max_search: int) -> None:
|
||||
"""Run the --fix flow: find safe versions, update requirements files."""
|
||||
# Map package names to their entries for source tracking
|
||||
pkg_entries: dict[str, list[dict]] = {}
|
||||
|
|
@ -1941,9 +1927,7 @@ def _run_fix(
|
|||
if git_entries:
|
||||
for e in git_entries:
|
||||
src = e["source_file"] or "CLI"
|
||||
print(
|
||||
f" [SKIP] {pkg_name} is a git URL dep in {src}, cannot auto-update"
|
||||
)
|
||||
print(f" [SKIP] {pkg_name} is a git URL dep in {src}, cannot auto-update")
|
||||
changes_summary.append(f" SKIP {pkg_name} (git URL)")
|
||||
continue
|
||||
|
||||
|
|
@ -1967,9 +1951,7 @@ def _run_fix(
|
|||
shutil.rmtree(dl_dir, ignore_errors = True)
|
||||
|
||||
if not current_ver:
|
||||
print(
|
||||
f" [WARN] Cannot determine current version of {pkg_name}, skipping fix"
|
||||
)
|
||||
print(f" [WARN] Cannot determine current version of {pkg_name}, skipping fix")
|
||||
changes_summary.append(f" SKIP {pkg_name} (version unknown)")
|
||||
continue
|
||||
|
||||
|
|
@ -1986,9 +1968,7 @@ def _run_fix(
|
|||
continue
|
||||
|
||||
print(f" [OK] {pkg_name}: {current_ver} -> {safe_ver}")
|
||||
changes_summary.append(
|
||||
f" FIX {pkg_name}=={current_ver} -> {pkg_name}=={safe_ver}"
|
||||
)
|
||||
changes_summary.append(f" FIX {pkg_name}=={current_ver} -> {pkg_name}=={safe_ver}")
|
||||
|
||||
# Update all occurrences in requirements files
|
||||
file_updates: dict[str, dict[int, str]] = {}
|
||||
|
|
@ -2038,9 +2018,7 @@ def _find_requirements_files(root: str) -> list[str]:
|
|||
dirnames[:] = [
|
||||
d
|
||||
for d in dirnames
|
||||
if not d.startswith(".")
|
||||
and d not in skip_dirs
|
||||
and not d.endswith(".egg-info")
|
||||
if not d.startswith(".") and d not in skip_dirs and not d.endswith(".egg-info")
|
||||
]
|
||||
dirname = os.path.basename(dirpath)
|
||||
for fname in sorted(filenames):
|
||||
|
|
@ -2115,9 +2093,7 @@ def main() -> int:
|
|||
print(f" {f}")
|
||||
req_files.extend(found)
|
||||
else:
|
||||
print(
|
||||
f" [WARN] No requirements files found in {scan_dir}/", file = sys.stderr
|
||||
)
|
||||
print(f" [WARN] No requirements files found in {scan_dir}/", file = sys.stderr)
|
||||
|
||||
# Build unified entry list: list of dicts with source tracking
|
||||
entries: list[dict] = []
|
||||
|
|
@ -2211,7 +2187,7 @@ def main() -> int:
|
|||
for err in download_errors:
|
||||
print(f" [ERROR] {err}", file = sys.stderr)
|
||||
print(
|
||||
" Refusing to report 'all clean' on a partial scan; " "exiting 2.",
|
||||
" Refusing to report 'all clean' on a partial scan; exiting 2.",
|
||||
file = sys.stderr,
|
||||
)
|
||||
return 2
|
||||
|
|
|
|||
|
|
@ -17,7 +17,11 @@ import zipfile
|
|||
from pathlib import Path
|
||||
|
||||
|
||||
def _atomic_write_text(path: Path, data: str, encoding: str = "utf-8") -> None:
|
||||
def _atomic_write_text(
|
||||
path: Path,
|
||||
data: str,
|
||||
encoding: str = "utf-8",
|
||||
) -> None:
|
||||
"""Atomic version of ``Path.write_text``.
|
||||
|
||||
A crash or signal mid-write leaves the prior file intact; the
|
||||
|
|
@ -41,9 +45,7 @@ def _atomic_write_text(path: Path, data: str, encoding: str = "utf-8") -> None:
|
|||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
BUILD_INFO_PATH = (
|
||||
REPO_ROOT / "studio" / "backend" / "utils" / "_studio_release_build.py"
|
||||
)
|
||||
BUILD_INFO_PATH = REPO_ROOT / "studio" / "backend" / "utils" / "_studio_release_build.py"
|
||||
BUILD_INFO_SUFFIX = "studio/backend/utils/_studio_release_build.py"
|
||||
VERSION_RE = re.compile(r"^v\d+\.\d+\.\d+(?:-[0-9A-Za-z.][0-9A-Za-z.-]*)?$")
|
||||
GIT_DESCRIBE_SUFFIX_RE = re.compile(r"-\d+-g[0-9A-Fa-f]+(?:-dirty)?$")
|
||||
|
|
|
|||
|
|
@ -128,12 +128,15 @@ def _normalize_yaml_run_strings(obj: Any) -> Any:
|
|||
return obj
|
||||
|
||||
|
||||
def _walk_yaml_diff(b: Any, a: Any, prefix: str = "") -> None:
|
||||
def _walk_yaml_diff(
|
||||
b: Any,
|
||||
a: Any,
|
||||
prefix: str = "",
|
||||
) -> None:
|
||||
"""Print a path-keyed summary of the first structural / scalar diff."""
|
||||
if type(b) is not type(a):
|
||||
print(
|
||||
f" type-diff at {prefix or '/'}: "
|
||||
f"{type(b).__name__} -> {type(a).__name__}",
|
||||
f" type-diff at {prefix or '/'}: " f"{type(b).__name__} -> {type(a).__name__}",
|
||||
)
|
||||
return
|
||||
if isinstance(b, dict):
|
||||
|
|
|
|||
|
|
@ -123,9 +123,7 @@ class _Builder(ast.NodeVisitor):
|
|||
def __init__(self):
|
||||
self.module = Scope("module", "<module>", None)
|
||||
self.uses: list[tuple[Scope, str, int]] = [] # (scope, name, lineno) hard loads
|
||||
self.soft_uses: list[
|
||||
tuple[Scope, str, int]
|
||||
] = [] # annotations: count as "used"
|
||||
self.soft_uses: list[tuple[Scope, str, int]] = [] # annotations: count as "used"
|
||||
# but never as "unresolved"
|
||||
# (forward refs / string annos)
|
||||
|
||||
|
|
@ -166,9 +164,7 @@ class _Builder(ast.NodeVisitor):
|
|||
|
||||
def _visit_stmt(self, node: ast.AST, scope: Scope) -> None:
|
||||
if isinstance(node, (ast.Import, ast.ImportFrom)):
|
||||
star = isinstance(node, ast.ImportFrom) and any(
|
||||
a.name == "*" for a in node.names
|
||||
)
|
||||
star = isinstance(node, ast.ImportFrom) and any(a.name == "*" for a in node.names)
|
||||
if star:
|
||||
scope.star_import = True
|
||||
for alias in node.names:
|
||||
|
|
@ -356,9 +352,7 @@ class _Builder(ast.NodeVisitor):
|
|||
self._bind_args(node.args, child)
|
||||
self._visit_expr(node.body, child)
|
||||
return
|
||||
if isinstance(
|
||||
node, (ast.ListComp, ast.SetComp, ast.GeneratorExp, ast.DictComp)
|
||||
):
|
||||
if isinstance(node, (ast.ListComp, ast.SetComp, ast.GeneratorExp, ast.DictComp)):
|
||||
child = Scope("comp", f"{scope.qualname}.<comp>", scope)
|
||||
for i, gen in enumerate(node.generators):
|
||||
# first iterable is evaluated in the enclosing scope
|
||||
|
|
@ -447,9 +441,7 @@ def _legb_chain(scope: Scope) -> list[Scope]:
|
|||
chain = [scope]
|
||||
p = scope.parent
|
||||
while p is not None:
|
||||
if (
|
||||
p.kind != "class" or p.parent is None
|
||||
): # module-level class never happens; keep module
|
||||
if p.kind != "class" or p.parent is None: # module-level class never happens; keep module
|
||||
if p.kind != "class":
|
||||
chain.append(p)
|
||||
p = p.parent
|
||||
|
|
@ -624,9 +616,7 @@ def compare(before_src: str, after_src: str, path: str) -> list[tuple[str, str]]
|
|||
for scope, names in b["ambiguous"].items():
|
||||
new = names - a["ambiguous"].get(scope, set())
|
||||
for n in sorted(new):
|
||||
findings.append(
|
||||
("WARN", f"{path}: AMBIGUOUS-BIND '{n}' import+non-import in {scope}")
|
||||
)
|
||||
findings.append(("WARN", f"{path}: AMBIGUOUS-BIND '{n}' import+non-import in {scope}"))
|
||||
|
||||
# 6. TARGET-MISSING (informational): a scope stopped resolving to an import
|
||||
# target. Real bugs are already covered above; remaining cases are code
|
||||
|
|
@ -639,9 +629,7 @@ def compare(before_src: str, after_src: str, path: str) -> list[tuple[str, str]]
|
|||
if t in added_module_targets
|
||||
else " [target not re-added here -> likely relocated/deleted]"
|
||||
)
|
||||
findings.append(
|
||||
("INFO", f"{path}: TARGET-MISSING {t} in scope {scope}{relocated}")
|
||||
)
|
||||
findings.append(("INFO", f"{path}: TARGET-MISSING {t} in scope {scope}{relocated}"))
|
||||
return findings
|
||||
|
||||
|
||||
|
|
@ -650,45 +638,38 @@ def compare(before_src: str, after_src: str, path: str) -> list[tuple[str, str]]
|
|||
_SELF_TESTS = {
|
||||
"dangling_alias": (
|
||||
# before: inline aliased import, used as _b
|
||||
"import os\n"
|
||||
"def f():\n"
|
||||
" import glob as _b\n"
|
||||
" return _b.glob('*')\n",
|
||||
"import os\ndef f():\n import glob as _b\n return _b.glob('*')\n",
|
||||
# after: hoisted to canonical, but reference NOT normalized -> _b dangles
|
||||
"import os\n" "import glob\n" "def f():\n" " return _b.glob('*')\n",
|
||||
"import os\nimport glob\ndef f():\n return _b.glob('*')\n",
|
||||
"BLOCKER",
|
||||
),
|
||||
"rename_clash": (
|
||||
# before: _b is a deliberate alias; `b` already means something else
|
||||
"import re as _b\n" "b = 123\n" "def f():\n" " return _b.compile('x'), b\n",
|
||||
"import re as _b\nb = 123\ndef f():\n return _b.compile('x'), b\n",
|
||||
# after: someone normalized _b -> b ; now f().b is the int, re is lost
|
||||
"import re\n" "b = 123\n" "def f():\n" " return b.compile('x'), b\n",
|
||||
"import re\nb = 123\ndef f():\n return b.compile('x'), b\n",
|
||||
"BLOCKER", # TARGET-MISSING from:.. or import:re in f
|
||||
),
|
||||
"clean_rename": (
|
||||
"def f():\n" " import glob as _g\n" " return _g.glob('*')\n",
|
||||
"import glob\n" "def f():\n" " return glob.glob('*')\n",
|
||||
"def f():\n import glob as _g\n return _g.glob('*')\n",
|
||||
"import glob\ndef f():\n return glob.glob('*')\n",
|
||||
None, # expect NO blocker
|
||||
),
|
||||
"clean_dedup_redundant": (
|
||||
"import sys\n" "def f():\n" " import sys\n" " return sys.argv\n",
|
||||
"import sys\n" "def f():\n" " return sys.argv\n",
|
||||
"import sys\ndef f():\n import sys\n return sys.argv\n",
|
||||
"import sys\ndef f():\n return sys.argv\n",
|
||||
None,
|
||||
),
|
||||
"from_import_dangling": (
|
||||
# from-import alias left un-normalized
|
||||
"def f():\n"
|
||||
" from importlib.metadata import version as _v\n"
|
||||
" return _v('x')\n",
|
||||
"from importlib.metadata import version\n" "def f():\n" " return _v('x')\n",
|
||||
"def f():\n from importlib.metadata import version as _v\n return _v('x')\n",
|
||||
"from importlib.metadata import version\ndef f():\n return _v('x')\n",
|
||||
"BLOCKER",
|
||||
),
|
||||
"local_var_clash": (
|
||||
# _b renamed to b, but b is a LOCAL variable in f -> import silently unused
|
||||
"def f(b):\n" " import re as _b\n" " return _b.compile(b)\n",
|
||||
"import re\n"
|
||||
"def f(b):\n"
|
||||
" return b.compile(b)\n", # 'b' is the param, not the module
|
||||
"def f(b):\n import re as _b\n return _b.compile(b)\n",
|
||||
"import re\ndef f(b):\n return b.compile(b)\n", # 'b' is the param, not the module
|
||||
"BLOCKER",
|
||||
),
|
||||
"substring_safe": (
|
||||
|
|
@ -705,11 +686,8 @@ _SELF_TESTS = {
|
|||
),
|
||||
"attr_access_not_a_use": (
|
||||
# x._b is attribute access, not a use of name _b; removing import _b is fine
|
||||
"import os\n"
|
||||
"def f(x):\n"
|
||||
" import sys as _b\n"
|
||||
" return x._b + _b.argv[0]\n",
|
||||
"import os\n" "import sys\n" "def f(x):\n" " return x._b + sys.argv[0]\n",
|
||||
"import os\ndef f(x):\n import sys as _b\n return x._b + _b.argv[0]\n",
|
||||
"import os\nimport sys\ndef f(x):\n return x._b + sys.argv[0]\n",
|
||||
None,
|
||||
),
|
||||
}
|
||||
|
|
@ -797,9 +775,7 @@ def audit_files(paths: list[str]) -> int:
|
|||
ok = n_err == 0 and n_fp == 0
|
||||
print(
|
||||
"\nAUDIT:",
|
||||
"ROBUST (no crashes, no false positives vs pyflakes)"
|
||||
if ok
|
||||
else "NEEDS WORK (see above)",
|
||||
"ROBUST (no crashes, no false positives vs pyflakes)" if ok else "NEEDS WORK (see above)",
|
||||
)
|
||||
return 0 if ok else 1
|
||||
|
||||
|
|
@ -835,18 +811,12 @@ def main() -> int:
|
|||
blockers = [f for f in findings if f[0] == "BLOCKER"]
|
||||
warns = [f for f in findings if f[0] == "WARN"]
|
||||
infos = [f for f in findings if f[0] == "INFO"]
|
||||
status = (
|
||||
"CLEAN"
|
||||
if not blockers and not warns
|
||||
else ("BLOCKERS" if blockers else "WARNINGS")
|
||||
)
|
||||
status = "CLEAN" if not blockers and not warns else ("BLOCKERS" if blockers else "WARNINGS")
|
||||
print(f"\n=== {path}: {status} ===")
|
||||
for sev, m in blockers + warns + infos:
|
||||
print(f" [{sev}] {m}")
|
||||
any_blocker = any_blocker or bool(blockers)
|
||||
print(
|
||||
"\nOVERALL:", "FAIL (blockers found)" if any_blocker else "PASS (no blockers)"
|
||||
)
|
||||
print("\nOVERALL:", "FAIL (blockers found)" if any_blocker else "PASS (no blockers)")
|
||||
return 1 if any_blocker else 0
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue