lastdragon/linux-kernel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Resguardo inicial

Browse Source
This commit is contained in:
lastdragon
2026-09-06 12:53:21 -06:00
parent 662efe3c2a
commit ad2088d178
94853 changed files with 43200404 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
linux-7.2.3/tools/lib/python/__init__.py
View File
linux-7.2.3/tools/lib/python/abi/__init__.py
View File
linux-7.2.3/tools/lib/python/abi/abi_parser.py
+631
View File
@@ -0,0 +1,631 @@
#!/usr/bin/env python3
# pylint: disable=R0902,R0903,R0911,R0912,R0913,R0914,R0915,R0917,C0302
# Copyright(c) 2025: Mauro Carvalho Chehab <mchehab@kernel.org>.
# SPDX-License-Identifier: GPL-2.0
"""
Parse ABI documentation and produce results from it.
"""
from argparse import Namespace
import logging
import os
import re
from pprint import pformat
from random import randrange, seed
# Import Python modules
from abi.helpers import AbiDebug, ABI_DIR
class AbiParser:
"""Main class to parse ABI files."""
#: Valid tags at Documentation/ABI.
TAGS = r"(what|where|date|kernelversion|contact|description|users)"
#: ABI elements that will auto-generate cross-references.
XREF = r"(?:^|\s|\()(\/(?:sys|config|proc|dev|kvd)\/[^,.:;\)\s]+)(?:[,.:;\)\s]|\Z)"
def __init__(self, directory, logger=None,
enable_lineno=False, show_warnings=True, debug=0):
"""Stores arguments for the class and initialize class vars."""
self.directory = directory
self.enable_lineno = enable_lineno
self.show_warnings = show_warnings
self.debug = debug
if not logger:
self.log = logging.getLogger("get_abi")
else:
self.log = logger
self.data = {}
self.what_symbols = {}
self.file_refs = {}
self.what_refs = {}
# Ignore files that contain such suffixes
self.ignore_suffixes = (".rej", ".org", ".orig", ".bak", "~")
# Regular expressions used on parser
self.re_abi_dir = re.compile(r"(.*)" + ABI_DIR)
self.re_tag = re.compile(r"(\S+)(:\s*)(.*)", re.I)
self.re_valid = re.compile(self.TAGS)
self.re_start_spc = re.compile(r"(\s*)(\S.*)")
self.re_whitespace = re.compile(r"^\s+")
# Regular used on print
self.re_what = re.compile(r"(\/?(?:[\w\-]+\/?){1,2})")
self.re_escape = re.compile(r"([\.\x01-\x08\x0e-\x1f\x21-\x2f\x3a-\x40\x7b-\xff])")
self.re_unprintable = re.compile(r"([\x00-\x2f\x3a-\x40\x5b-\x60\x7b-\xff]+)")
self.re_title_mark = re.compile(r"\n[\-\*\=\^\~]+\n")
self.re_doc = re.compile(r"Documentation/(?!devicetree)(\S+)\.rst")
self.re_abi = re.compile(r"(Documentation/ABI/)([\w\/\-]+)")
self.re_xref_node = re.compile(self.XREF)
def warn(self, fdata, msg, extra=None):
"""Displays a parse error if warning is enabled."""
if not self.show_warnings:
return
msg = f"{fdata.fname}:{fdata.ln}: {msg}"
if extra:
msg += "\n\t\t" + extra
self.log.warning(msg)
def add_symbol(self, what, fname, ln=None, xref=None):
"""Create a reference table describing where each 'what' is located."""
if what not in self.what_symbols:
self.what_symbols[what] = {"file": {}}
if fname not in self.what_symbols[what]["file"]:
self.what_symbols[what]["file"][fname] = []
if ln and ln not in self.what_symbols[what]["file"][fname]:
self.what_symbols[what]["file"][fname].append(ln)
if xref:
self.what_symbols[what]["xref"] = xref
def _parse_line(self, fdata, line):
"""Parse a single line of an ABI file."""
new_what = False
new_tag = False
content = None
match = self.re_tag.match(line)
if match:
new = match.group(1).lower()
sep = match.group(2)
content = match.group(3)
match = self.re_valid.search(new)
if match:
new_tag = match.group(1)
else:
if fdata.tag == "description":
# New "tag" is actually part of description.
# Don't consider it a tag
new_tag = False
elif fdata.tag != "":
self.warn(fdata, f"tag '{fdata.tag}' is invalid", line)
if new_tag:
# "where" is Invalid, but was a common mistake. Warn if found
if new_tag == "where":
self.warn(fdata, "tag 'Where' is invalid. Should be 'What:' instead")
new_tag = "what"
if new_tag == "what":
fdata.space = None
if content not in self.what_symbols:
self.add_symbol(what=content, fname=fdata.fname, ln=fdata.ln)
if fdata.tag == "what":
fdata.what.append(content.strip("\n"))
else:
if fdata.key:
if "description" not in self.data.get(fdata.key, {}):
self.warn(fdata, f"{fdata.key} doesn't have a description")
for w in fdata.what:
self.add_symbol(what=w, fname=fdata.fname,
ln=fdata.what_ln, xref=fdata.key)
fdata.label = content
new_what = True
key = "abi_" + content.lower()
fdata.key = self.re_unprintable.sub("_", key).strip("_")
# Avoid duplicated keys but using a defined seed, to make
# the namespace identical if there aren't changes at the
# ABI symbols
seed(42)
while fdata.key in self.data:
char = randrange(0, 51) + ord("A")
if char > ord("Z"):
char += ord("a") - ord("Z") - 1
fdata.key += chr(char)
if fdata.key and fdata.key not in self.data:
self.data[fdata.key] = {
"what": [content],
"file": [fdata.file_ref],
"path": fdata.ftype,
"line_no": fdata.ln,
}
fdata.what = self.data[fdata.key]["what"]
self.what_refs[content] = fdata.key
fdata.tag = new_tag
fdata.what_ln = fdata.ln
if fdata.nametag["what"]:
t = (content, fdata.key)
if t not in fdata.nametag["symbols"]:
fdata.nametag["symbols"].append(t)
return
if fdata.tag and new_tag:
fdata.tag = new_tag
if new_what:
fdata.label = ""
if "description" in self.data[fdata.key]:
self.data[fdata.key]["description"] += "\n\n"
if fdata.file_ref not in self.data[fdata.key]["file"]:
self.data[fdata.key]["file"].append(fdata.file_ref)
if self.debug == AbiDebug.WHAT_PARSING:
self.log.debug("what: %s", fdata.what)
if not fdata.what:
self.warn(fdata, "'What:' should come first:", line)
return
if new_tag == "description":
fdata.space = None
if content:
sep = sep.replace(":", " ")
c = " " * len(new_tag) + sep + content
c = c.expandtabs()
match = self.re_start_spc.match(c)
if match:
# Preserve initial spaces for the first line
fdata.space = match.group(1)
content = match.group(2) + "\n"
self.data[fdata.key][fdata.tag] = content
return
# Store any contents before tags at the database
if not fdata.tag and "what" in fdata.nametag:
fdata.nametag["description"] += line
return
if fdata.tag == "description":
content = line.expandtabs()
if self.re_whitespace.sub("", content) == "":
self.data[fdata.key][fdata.tag] += "\n"
return
if fdata.space is None:
match = self.re_start_spc.match(content)
if match:
# Preserve initial spaces for the first line
fdata.space = match.group(1)
content = match.group(2) + "\n"
else:
if content.startswith(fdata.space):
content = content[len(fdata.space):]
else:
fdata.space = ""
if fdata.tag == "what":
w = content.strip("\n")
if w:
self.data[fdata.key][fdata.tag].append(w)
else:
self.data[fdata.key][fdata.tag] += content
return
content = line.strip()
if fdata.tag:
if fdata.tag == "what":
w = content.strip("\n")
if w:
self.data[fdata.key][fdata.tag].append(w)
else:
self.data[fdata.key][fdata.tag] += "\n" + content.rstrip("\n")
return
# Everything else is error
if content:
self.warn(fdata, "Unexpected content", line)
def parse_readme(self, nametag, fname):
"""Parse ABI README file."""
nametag["what"] = ["Introduction"]
nametag["path"] = "README"
with open(fname, "r", encoding="utf8", errors="backslashreplace") as fp:
for line in fp:
match = self.re_tag.match(line)
if match:
new = match.group(1).lower()
match = self.re_valid.search(new)
if match:
nametag["description"] += "\n:" + line
continue
nametag["description"] += line
def parse_file(self, fname, path, basename):
"""Parse a single file."""
ref = f"abi_file_{path}_{basename}"
ref = self.re_unprintable.sub("_", ref).strip("_")
# Store per-file state into a namespace variable. This will be used
# by the per-line parser state machine and by the warning function.
fdata = Namespace
fdata.fname = fname
fdata.name = basename
pos = fname.find(ABI_DIR)
if pos > 0:
f = fname[pos:]
else:
f = fname
fdata.file_ref = (f, ref)
self.file_refs[f] = ref
fdata.ln = 0
fdata.what_ln = 0
fdata.tag = ""
fdata.label = ""
fdata.what = []
fdata.key = None
fdata.xrefs = None
fdata.space = None
fdata.ftype = path.split("/")[0]
fdata.nametag = {}
fdata.nametag["what"] = [f"ABI file {path}/{basename}"]
fdata.nametag["type"] = "File"
fdata.nametag["path"] = fdata.ftype
fdata.nametag["file"] = [fdata.file_ref]
fdata.nametag["line_no"] = 1
fdata.nametag["description"] = ""
fdata.nametag["symbols"] = []
self.data[ref] = fdata.nametag
if self.debug & AbiDebug.WHAT_OPEN:
self.log.debug("Opening file %s", fname)
if basename == "README":
self.parse_readme(fdata.nametag, fname)
return
with open(fname, "r", encoding="utf8", errors="backslashreplace") as fp:
for line in fp:
fdata.ln += 1
self._parse_line(fdata, line)
if "description" in fdata.nametag:
fdata.nametag["description"] = fdata.nametag["description"].lstrip("\n")
if fdata.key:
if "description" not in self.data.get(fdata.key, {}):
self.warn(fdata, f"{fdata.key} doesn't have a description")
for w in fdata.what:
self.add_symbol(what=w, fname=fname, xref=fdata.key)
def _parse_abi(self, root=None):
"""Internal function to parse documentation ABI recursively."""
if not root:
root = self.directory
with os.scandir(root) as obj:
for entry in obj:
name = os.path.join(root, entry.name)
if entry.is_dir():
self._parse_abi(name)
continue
if not entry.is_file():
continue
basename = os.path.basename(name)
if basename.startswith("."):
continue
if basename.endswith(self.ignore_suffixes):
continue
path = self.re_abi_dir.sub("", os.path.dirname(name))
self.parse_file(name, path, basename)
def parse_abi(self, root=None):
"""Parse documentation ABI."""
self._parse_abi(root)
if self.debug & AbiDebug.DUMP_ABI_STRUCTS:
self.log.debug(pformat(self.data))
def desc_txt(self, desc):
"""Print description as found inside ABI files."""
desc = desc.strip(" \t\n")
return desc + "\n\n"
def xref(self, fname):
"""
Converts a Documentation/ABI + basename into a ReST cross-reference.
"""
xref = self.file_refs.get(fname)
if not xref:
return None
else:
return xref
def desc_rst(self, desc):
"""Enrich ReST output by creating cross-references."""
# Remove title markups from the description
# Having titles inside ABI files will only work if extra
# care would be taken in order to strictly follow the same
# level order for each markup.
desc = self.re_title_mark.sub("\n\n", "\n" + desc)
desc = desc.rstrip(" \t\n").lstrip("\n")
# Python's regex performance for non-compiled expressions is a lot
# than Perl, as Perl automatically caches them at their
# first usage. Here, we'll need to do the same, as otherwise the
# performance penalty is be high
new_desc = ""
for d in desc.split("\n"):
if d == "":
new_desc += "\n"
continue
# Use cross-references for doc files where needed
d = self.re_doc.sub(r":doc:`/\1`", d)
# Use cross-references for ABI generated docs where needed
matches = self.re_abi.findall(d)
for m in matches:
abi = m[0] + m[1]
xref = self.file_refs.get(abi)
if not xref:
# This may happen if ABI is on a separate directory,
# like parsing ABI testing and symbol is at stable.
# The proper solution is to move this part of the code
# for it to be inside sphinx/kernel_abi.py
self.log.info("Didn't find ABI reference for '%s'", abi)
else:
new = self.re_escape.sub(r"\\\1", m[1])
d = re.sub(fr"\b{abi}\b", f":ref:`{new} <{xref}>`", d)
# Seek for cross reference symbols like /sys/...
# Need to be careful to avoid doing it on a code block
if d[0] not in [" ", "\t"]:
matches = self.re_xref_node.findall(d)
for m in matches:
# Finding ABI here is more complex due to wildcards
xref = self.what_refs.get(m)
if xref:
new = self.re_escape.sub(r"\\\1", m)
d = re.sub(fr"\b{m}\b", f":ref:`{new} <{xref}>`", d)
new_desc += d + "\n"
return new_desc + "\n\n"
def doc(self, output_in_txt=False, show_symbols=True, show_file=True,
filter_path=None):
"""Print ABI at stdout."""
part = None
for key, v in sorted(self.data.items(),
key=lambda x: (x[1].get("type", ""),
x[1].get("what"))):
wtype = v.get("type", "Symbol")
file_ref = v.get("file")
names = v.get("what", [""])
if wtype == "File":
if not show_file:
continue
else:
if not show_symbols:
continue
if filter_path:
if v.get("path") != filter_path:
continue
msg = ""
if wtype != "File":
cur_part = names[0]
if cur_part.find("/") >= 0:
match = self.re_what.match(cur_part)
if match:
symbol = match.group(1).rstrip("/")
cur_part = "Symbols under " + symbol
if cur_part and cur_part != part:
part = cur_part
msg += part + "\n"+ "-" * len(part) +"\n\n"
msg += f".. _{key}:\n\n"
max_len = 0
for i in range(0, len(names)): # pylint: disable=C0200
names[i] = "**" + self.re_escape.sub(r"\\\1", names[i]) + "**"
max_len = max(max_len, len(names[i]))
msg += "+-" + "-" * max_len + "-+\n"
for name in names:
msg += f"| {name}" + " " * (max_len - len(name)) + " |\n"
msg += "+-" + "-" * max_len + "-+\n"
msg += "\n"
for ref in file_ref:
if wtype == "File":
msg += f".. _{ref[1]}:\n\n"
else:
base = os.path.basename(ref[0])
msg += f"Defined on file :ref:`{base} <{ref[1]}>`\n\n"
if wtype == "File":
msg += names[0] +"\n" + "-" * len(names[0]) +"\n\n"
desc = v.get("description")
if not desc and wtype != "File":
msg += f"DESCRIPTION MISSING for {names[0]}\n\n"
if desc:
if output_in_txt:
msg += self.desc_txt(desc)
else:
msg += self.desc_rst(desc)
symbols = v.get("symbols")
if symbols:
msg += "Has the following ABI:\n\n"
for w, label in symbols:
# Escape special chars from content
content = self.re_escape.sub(r"\\\1", w)
msg += f"- :ref:`{content} <{label}>`\n\n"
users = v.get("users")
if users and users.strip(" \t\n"):
users = users.strip("\n").replace('\n', '\n\t')
msg += f"Users:\n\t{users}\n\n"
ln = v.get("line_no", 1)
yield (msg, file_ref[0][0], ln)
def check_issues(self):
"""Warn about duplicated ABI entries."""
for what, v in self.what_symbols.items():
files = v.get("file")
if not files:
# Should never happen if the parser works properly
self.log.warning("%s doesn't have a file associated", what)
continue
if len(files) == 1:
continue
f = []
for fname, lines in sorted(files.items()):
if not lines:
f.append(f"{fname}")
elif len(lines) == 1:
f.append(f"{fname}:{lines[0]}")
else:
m = fname + "lines "
m += ", ".join(str(x) for x in lines)
f.append(m)
self.log.warning("%s is defined %d times: %s", what, len(f), "; ".join(f))
def search_symbols(self, expr):
""" Searches for ABI symbols."""
regex = re.compile(expr, re.I)
found_keys = 0
for t in sorted(self.data.items(), key=lambda x: [0]):
v = t[1]
wtype = v.get("type", "")
if wtype == "File":
continue
for what in v.get("what", [""]):
if regex.search(what):
found_keys += 1
kernelversion = v.get("kernelversion", "").strip(" \t\n")
date = v.get("date", "").strip(" \t\n")
contact = v.get("contact", "").strip(" \t\n")
users = v.get("users", "").strip(" \t\n")
desc = v.get("description", "").strip(" \t\n")
files = []
for f in v.get("file", ()):
files.append(f[0])
what = str(found_keys) + ". " + what
title_tag = "-" * len(what)
print(f"\n{what}\n{title_tag}\n")
if kernelversion:
print(f"Kernel version:\t\t{kernelversion}")
if date:
print(f"Date:\t\t\t{date}")
if contact:
print(f"Contact:\t\t{contact}")
if users:
print(f"Users:\t\t\t{users}")
print("Defined on file(s):\t" + ", ".join(files))
if desc:
desc = desc.strip("\n")
print(f"\n{desc}\n")
if not found_keys:
print(f"Regular expression /{expr}/ not found.")
linux-7.2.3/tools/lib/python/abi/abi_regex.py
+248
View File
@@ -0,0 +1,248 @@
#!/usr/bin/env python3
# xxpylint: disable=R0903
# Copyright(c) 2025: Mauro Carvalho Chehab <mchehab@kernel.org>.
# SPDX-License-Identifier: GPL-2.0
"""
Convert ABI what into regular expressions
"""
import re
import sys
from pprint import pformat
from abi.abi_parser import AbiParser
from abi.helpers import AbiDebug
class AbiRegex(AbiParser):
"""
Extends AbiParser to search ABI nodes with regular expressions.
There some optimizations here to allow a quick symbol search:
instead of trying to place all symbols altogether an doing linear
search which is very time consuming, create a tree with one depth,
grouping similar symbols altogether.
Yet, sometimes a full search will be needed, so we have a special branch
on such group tree where other symbols are placed.
"""
#: Escape only ASCII visible characters.
escape_symbols = r"([\x21-\x29\x2b-\x2d\x3a-\x40\x5c\x60\x7b-\x7e])"
#: Special group for other nodes.
leave_others = "others"
# Tuples with regular expressions to be compiled and replacement data
re_whats = [
# Drop escape characters that might exist
(re.compile("\\\\"), ""),
# Temporarily escape dot characters
(re.compile(r"\."), "\xf6"),
# Temporarily change [0-9]+ type of patterns
(re.compile(r"\[0\-9\]\+"), "\xff"),
# Temporarily change [\d+-\d+] type of patterns
(re.compile(r"\[0\-\d+\]"), "\xff"),
(re.compile(r"\[0:\d+\]"), "\xff"),
(re.compile(r"\[(\d+)\]"), "\xf4\\\\d+\xf5"),
# Temporarily change [0-9] type of patterns
(re.compile(r"\[(\d)\-(\d)\]"), "\xf4\1-\2\xf5"),
# Handle multiple option patterns
(re.compile(r"[\{\<\[]([\w_]+)(?:[,|]+([\w_]+)){1,}[\}\>\]]"), r"(\1|\2)"),
# Handle wildcards
(re.compile(r"([^\/])\*"), "\\1\\\\w\xf7"),
(re.compile(r"/\*/"), "/.*/"),
(re.compile(r"/\xf6\xf6\xf6"), "/.*"),
(re.compile(r"\<[^\>]+\>"), "\\\\w\xf7"),
(re.compile(r"\{[^\}]+\}"), "\\\\w\xf7"),
(re.compile(r"\[[^\]]+\]"), "\\\\w\xf7"),
(re.compile(r"XX+"), "\\\\w\xf7"),
(re.compile(r"([^A-Z])[XYZ]([^A-Z])"), "\\1\\\\w\xf7\\2"),
(re.compile(r"([^A-Z])[XYZ]$"), "\\1\\\\w\xf7"),
(re.compile(r"_[AB]_"), "_\\\\w\xf7_"),
# Recover [0-9] type of patterns
(re.compile(r"\xf4"), "["),
(re.compile(r"\xf5"), "]"),
# Remove duplicated spaces
(re.compile(r"\s+"), r" "),
# Special case: drop comparison as in:
# What: foo = <something>
# (this happens on a few IIO definitions)
(re.compile(r"\s*\=.*$"), ""),
# Escape all other symbols
(re.compile(escape_symbols), r"\\\1"),
(re.compile(r"\\\\"), r"\\"),
(re.compile(r"\\([\[\]\(\)\|])"), r"\1"),
(re.compile(r"(\d+)\\(-\d+)"), r"\1\2"),
(re.compile(r"\xff"), r"\\d+"),
# Special case: IIO ABI which a parenthesis.
(re.compile(r"sqrt(.*)"), r"sqrt(.*)"),
# Simplify regexes with multiple .*
(re.compile(r"(?:\.\*){2,}"), ""),
# Recover dot characters
(re.compile(r"\xf6"), "\\."),
# Recover plus characters
(re.compile(r"\xf7"), "+"),
]
#: Regex to check if the symbol name has a number on it.
re_has_num = re.compile(r"\\d")
#: Symbol name after escape_chars that are considered a devnode basename.
re_symbol_name = re.compile(r"(\w|\\[\.\-\:])+$")
#: List of popular group names to be skipped to minimize regex group size
#: Use AbiDebug.SUBGROUP_SIZE to detect those.
skip_names = set(["devices", "hwmon"])
def regex_append(self, what, new):
"""
Get a search group for a subset of regular expressions.
As ABI may have thousands of symbols, using a for to search all
regular expressions is at least O(n^2). When there are wildcards,
the complexity increases substantially, eventually becoming exponential.
To avoid spending too much time on them, use a logic to split
them into groups. The smaller the group, the better, as it would
mean that searches will be confined to a small number of regular
expressions.
The conversion to a regex subset is tricky, as we need something
that can be easily obtained from the sysfs symbol and from the
regular expression. So, we need to discard nodes that have
wildcards.
If it can't obtain a subgroup, place the regular expression inside
a special group (self.leave_others).
"""
search_group = None
for search_group in reversed(new.split("/")):
if not search_group or search_group in self.skip_names:
continue
if self.re_symbol_name.match(search_group):
break
if not search_group:
search_group = self.leave_others
if self.debug & AbiDebug.SUBGROUP_MAP:
self.log.debug("%s: mapped as %s", what, search_group)
try:
if search_group not in self.regex_group:
self.regex_group[search_group] = []
self.regex_group[search_group].append(re.compile(new))
if self.search_string:
if what.find(self.search_string) >= 0:
print(f"What: {what}")
except re.PatternError:
self.log.warning("Ignoring '%s' as it produced an invalid regex:\n"
" '%s'", what, new)
def get_regexes(self, what):
"""
Given an ABI devnode, return a list of all regular expressions that
may match it, based on the sub-groups created by regex_append().
"""
re_list = []
patches = what.split("/")
patches.reverse()
patches.append(self.leave_others)
for search_group in patches:
if search_group in self.regex_group:
re_list += self.regex_group[search_group]
return re_list
def __init__(self, *args, **kwargs):
"""
Override init method to get verbose argument
"""
self.regex_group = None
self.search_string = None
self.re_string = None
if "search_string" in kwargs:
self.search_string = kwargs.get("search_string")
del kwargs["search_string"]
if self.search_string:
try:
self.re_string = re.compile(self.search_string)
except re.PatternError as e:
msg = f"{self.search_string} is not a valid regular expression"
raise ValueError(msg) from e
super().__init__(*args, **kwargs)
def parse_abi(self, *args, **kwargs):
super().parse_abi(*args, **kwargs)
self.regex_group = {}
print("Converting ABI What fields into regexes...", file=sys.stderr)
for t in sorted(self.data.items(), key=lambda x: x[0]):
v = t[1]
if v.get("type") == "File":
continue
v["regex"] = []
for what in v.get("what", []):
if not what.startswith("/sys"):
continue
new = what
for r, s in self.re_whats:
try:
new = r.sub(s, new)
except re.PatternError as e:
# Help debugging troubles with new regexes
raise re.PatternError(f"{e}\nwhile re.sub('{r.pattern}', {s}, str)") from e
v["regex"].append(new)
if self.debug & AbiDebug.REGEX:
self.log.debug("%-90s <== %s", new, what)
# Store regex into a subgroup to speedup searches
self.regex_append(what, new)
if self.debug & AbiDebug.SUBGROUP_DICT:
self.log.debug("%s", pformat(self.regex_group))
if self.debug & AbiDebug.SUBGROUP_SIZE:
biggestd_keys = sorted(self.regex_group.keys(),
key= lambda k: len(self.regex_group[k]),
reverse=True)
print("Top regex subgroups:", file=sys.stderr)
for k in biggestd_keys[:10]:
print(f"{k} has {len(self.regex_group[k])} elements", file=sys.stderr)
linux-7.2.3/tools/lib/python/abi/helpers.py
+40
View File
@@ -0,0 +1,40 @@
#!/usr/bin/env python3
# Copyright(c) 2025: Mauro Carvalho Chehab <mchehab@kernel.org>.
# pylint: disable=R0903
# SPDX-License-Identifier: GPL-2.0
"""
Helper classes for ABI parser
"""
ABI_DIR = "Documentation/ABI/"
class AbiDebug:
"""Debug levels"""
WHAT_PARSING = 1 #: Enable debug parsing logic.
WHAT_OPEN = 2 #: Enable debug messages on file open.
DUMP_ABI_STRUCTS = 4 #: Enable debug for ABI parse data.
UNDEFINED = 8 #: Enable extra undefined symbol data.
REGEX = 16 #: Enable debug for what to regex conversion.
SUBGROUP_MAP = 32 #: Enable debug for symbol regex subgroups
SUBGROUP_DICT = 64 #: Enable debug for sysfs graph tree variable.
SUBGROUP_SIZE = 128 #: Enable debug of search groups.
GRAPH = 256 #: Display ref tree graph for undefined symbols.
#: Helper messages for each debug variable
DEBUG_HELP = """
1 - enable debug parsing logic
2 - enable debug messages on file open
4 - enable debug for ABI parse data
8 - enable extra debug information to identify troubles
with ABI symbols found at the local machine that
weren't found on ABI documentation (used only for
undefined subcommand)
16 - enable debug for what to regex conversion
32 - enable debug for symbol regex subgroups
64 - enable debug for sysfs graph tree variable
128 - enable debug of search groups
256 - enable displaying refrence tree graphs for undefined symbols.
"""
linux-7.2.3/tools/lib/python/abi/system_symbols.py
+378
View File
@@ -0,0 +1,378 @@
#!/usr/bin/env python3
# pylint: disable=R0902,R0912,R0914,R0915,R1702
# Copyright(c) 2025: Mauro Carvalho Chehab <mchehab@kernel.org>.
# SPDX-License-Identifier: GPL-2.0
"""
Parse ABI documentation and produce results from it.
"""
import os
import re
import sys
from concurrent import futures
from datetime import datetime
from random import shuffle
from abi.helpers import AbiDebug
class SystemSymbols:
"""Stores arguments for the class and initialize class vars."""
def graph_add_file(self, path, link=None):
"""
add a file path to the sysfs graph stored at self.root.
"""
if path in self.files:
return
name = ""
ref = self.root
for edge in path.split("/"):
name += edge + "/"
if edge not in ref:
ref[edge] = {"__name": [name.rstrip("/")]}
ref = ref[edge]
if link and link not in ref["__name"]:
ref["__name"].append(link.rstrip("/"))
self.files.add(path)
def print_graph(self, root_prefix="", root=None, level=0):
"""Prints a reference tree graph using UTF-8 characters."""
if not root:
root = self.root
level = 0
# Prevent endless traverse
if level > 5:
return
if level > 0:
prefix = "├──"
last_prefix = "└──"
else:
prefix = ""
last_prefix = ""
items = list(root.items())
names = root.get("__name", [])
for k, edge in items:
if k == "__name":
continue
if not k:
k = "/"
if len(names) > 1:
k += " links: " + ",".join(names[1:])
if edge == items[-1][1]:
print(root_prefix + last_prefix + k)
p = root_prefix
if level > 0:
p += " "
self.print_graph(p, edge, level + 1)
else:
print(root_prefix + prefix + k)
p = root_prefix + "│ "
self.print_graph(p, edge, level + 1)
def _walk(self, root):
"""
Walk through sysfs to get all devnodes that aren't ignored.
By default, uses /sys as sysfs mounting point. If another
directory is used, it replaces them to /sys at the patches.
"""
with os.scandir(root) as obj:
for entry in obj:
path = os.path.join(root, entry.name)
if self.sysfs:
p = path.replace(self.sysfs, "/sys", count=1)
else:
p = path
if self.re_ignore.search(p):
return
# Handle link first to avoid directory recursion
if entry.is_symlink():
real = os.path.realpath(path)
if not self.sysfs:
self.aliases[path] = real
else:
real = real.replace(self.sysfs, "/sys", count=1)
# Add absfile location to graph if it doesn't exist
if not self.re_ignore.search(real):
# Add link to the graph
self.graph_add_file(real, p)
elif entry.is_file():
self.graph_add_file(p)
elif entry.is_dir():
self._walk(path)
def __init__(self, abi, sysfs="/sys", hints=False):
"""
Initialize internal variables and get a list of all files inside
sysfs that can currently be parsed.
Please notice that there are several entries on sysfs that aren't
documented as ABI. Ignore those.
The real paths will be stored under self.files. Aliases will be
stored in separate, as self.aliases.
"""
self.abi = abi
self.log = abi.log
if sysfs != "/sys":
self.sysfs = sysfs.rstrip("/")
else:
self.sysfs = None
self.hints = hints
self.root = {}
self.aliases = {}
self.files = set()
dont_walk = [
# Those require root access and aren't documented at ABI
f"^{sysfs}/kernel/debug",
f"^{sysfs}/kernel/tracing",
f"^{sysfs}/fs/pstore",
f"^{sysfs}/fs/bpf",
f"^{sysfs}/fs/fuse",
# This is not documented at ABI
f"^{sysfs}/module",
f"^{sysfs}/fs/cgroup", # this is big and has zero docs under ABI
f"^{sysfs}/firmware", # documented elsewhere: ACPI, DT bindings
"sections|notes", # aren't actually part of ABI
# kernel-parameters.txt - not easy to parse
"parameters",
]
self.re_ignore = re.compile("|".join(dont_walk))
print(f"Reading {sysfs} directory contents...", file=sys.stderr)
self._walk(sysfs)
def check_file(self, refs, found):
"""Check missing ABI symbols for a given sysfs file."""
res_list = []
try:
for names in refs:
fname = names[0]
res = {
"found": False,
"fname": fname,
"msg": "",
}
res_list.append(res)
re_what = self.abi.get_regexes(fname)
if not re_what:
self.abi.log.warning(f"missing rules for {fname}")
continue
for name in names:
for r in re_what:
if self.abi.debug & AbiDebug.UNDEFINED:
self.log.debug("check if %s matches '%s'", name, r.pattern)
if r.match(name):
res["found"] = True
if found:
res["msg"] += f" {fname}: regex:\n\t"
continue
if self.hints and not res["found"]:
res["msg"] += f" {fname} not found. Tested regexes:\n"
for r in re_what:
res["msg"] += " " + r.pattern + "\n"
except KeyboardInterrupt:
pass
return res_list
def _ref_interactor(self, root):
"""Recursive function to interact over the sysfs tree."""
for k, v in root.items():
if isinstance(v, dict):
yield from self._ref_interactor(v)
if root == self.root or k == "__name":
continue
if self.abi.re_string:
fname = v["__name"][0]
if self.abi.re_string.search(fname):
yield v
else:
yield v
def get_fileref(self, all_refs, chunk_size):
"""Interactor to group refs into chunks."""
n = 0
refs = []
for ref in all_refs:
refs.append(ref)
n += 1
if n >= chunk_size:
yield refs
n = 0
refs = []
yield refs
def check_undefined_symbols(self, max_workers=None, chunk_size=50,
found=None, dry_run=None):
"""Seach ABI for sysfs symbols missing documentation."""
self.abi.parse_abi()
if self.abi.debug & AbiDebug.GRAPH:
self.print_graph()
all_refs = []
for ref in self._ref_interactor(self.root):
all_refs.append(ref["__name"])
if dry_run:
print("Would check", file=sys.stderr)
for ref in all_refs:
print(", ".join(ref))
return
print("Starting to search symbols (it may take several minutes):",
file=sys.stderr)
start = datetime.now()
old_elapsed = None
# Python doesn't support multithreading due to limitations on its
# global lock (GIL). While Python 3.13 finally made GIL optional,
# there are still issues related to it. Also, we want to have
# backward compatibility with older versions of Python.
#
# So, use instead multiprocess. However, Python is very slow passing
# data from/to multiple processes. Also, it may consume lots of memory
# if the data to be shared is not small. So, we need to group workload
# in chunks that are big enough to generate performance gains while
# not being so big that would cause out-of-memory.
num_refs = len(all_refs)
print(f"Number of references to parse: {num_refs}", file=sys.stderr)
if not max_workers:
max_workers = os.cpu_count()
elif max_workers > os.cpu_count():
max_workers = os.cpu_count()
max_workers = max(max_workers, 1)
max_chunk_size = int((num_refs + max_workers - 1) / max_workers)
chunk_size = min(chunk_size, max_chunk_size)
chunk_size = max(1, chunk_size)
if max_workers > 1:
executor = futures.ProcessPoolExecutor
# Place references in a random order. This may help improving
# performance, by mixing complex/simple expressions when creating
# chunks
shuffle(all_refs)
else:
# Python has a high overhead with processes. When there's just
# one worker, it is faster to not create a new process.
# Yet, User still deserves to have a progress print. So, use
# python's "thread", which is actually a single process, using
# an internal schedule to switch between tasks. No performance
# gains for non-IO tasks, but still it can be quickly interrupted
# from time to time to display progress.
executor = futures.ThreadPoolExecutor
not_found = []
f_list = []
with executor(max_workers=max_workers) as exe:
for refs in self.get_fileref(all_refs, chunk_size):
if refs:
try:
f_list.append(exe.submit(self.check_file, refs, found))
except KeyboardInterrupt:
return
total = len(f_list)
if not total:
if self.abi.re_string:
print(f"No ABI symbol matches {self.abi.search_string}")
else:
self.abi.log.warning("No ABI symbols found")
return
print(f"{len(f_list):6d} jobs queued on {max_workers} workers",
file=sys.stderr)
while f_list:
try:
t = futures.wait(f_list, timeout=1,
return_when=futures.FIRST_COMPLETED)
done = t[0]
for fut in done:
res_list = fut.result()
for res in res_list:
if not res["found"]:
not_found.append(res["fname"])
if res["msg"]:
print(res["msg"])
f_list.remove(fut)
except KeyboardInterrupt:
return
except RuntimeError as e:
self.abi.log.warning(f"Future: {e}")
break
if sys.stderr.isatty():
elapsed = str(datetime.now() - start).split(".", maxsplit=1)[0]
if len(f_list) < total:
elapsed += f" ({total - len(f_list)}/{total} jobs completed). "
if elapsed != old_elapsed:
print(elapsed + "\r", end="", flush=True,
file=sys.stderr)
old_elapsed = elapsed
elapsed = str(datetime.now() - start).split(".", maxsplit=1)[0]
print(elapsed, file=sys.stderr)
for f in sorted(not_found):
print(f"{f} not found.")
linux-7.2.3/tools/lib/python/feat/parse_features.py
Executable
+507
View File
@@ -0,0 +1,507 @@
#!/usr/bin/env python3
# pylint: disable=R0902,R0911,R0912,R0914,R0915
# Copyright(c) 2025: Mauro Carvalho Chehab <mchehab@kernel.org>.
# SPDX-License-Identifier: GPL-2.0
"""
Library to parse the Linux Feature files and produce a ReST book.
"""
import os
import re
import sys
from glob import iglob
class ParseFeature:
"""
Parses Documentation/features, allowing to generate ReST documentation
from it.
"""
#: feature header string.
h_name = "Feature"
#: Kernel config header string.
h_kconfig = "Kconfig"
#: description header string.
h_description = "Description"
#: subsystem header string.
h_subsys = "Subsystem"
#: status header string.
h_status = "Status"
#: architecture header string.
h_arch = "Architecture"
#: Sort order for status. Others will be mapped at the end.
status_map = {
"ok": 0,
"TODO": 1,
"N/A": 2,
# The only missing status is "..", which was mapped as "---",
# as this is an special ReST cell value. Let it get the
# default order (99).
}
def __init__(self, prefix, debug=0, enable_fname=False):
"""
Sets internal variables.
"""
self.prefix = prefix
self.debug = debug
self.enable_fname = enable_fname
self.data = {}
# Initial maximum values use just the headers
self.max_size_name = len(self.h_name)
self.max_size_kconfig = len(self.h_kconfig)
self.max_size_description = len(self.h_description)
self.max_size_desc_word = 0
self.max_size_subsys = len(self.h_subsys)
self.max_size_status = len(self.h_status)
self.max_size_arch = len(self.h_arch)
self.max_size_arch_with_header = self.max_size_arch + self.max_size_arch
self.description_size = 1
self.msg = ""
def emit(self, msg="", end="\n"):
"""Helper function to append a new message for feature output."""
self.msg += msg + end
def parse_error(self, fname, ln, msg, data=None):
"""
Displays an error message, printing file name and line.
"""
if ln:
fname += f"#{ln}"
print(f"Warning: file {fname}: {msg}", file=sys.stderr, end="")
if data:
data = data.rstrip()
print(f":\n\t{data}", file=sys.stderr)
else:
print("", file=sys.stderr)
def parse_feat_file(self, fname):
"""Parses a single arch-support.txt feature file."""
if os.path.isdir(fname):
return
base = os.path.basename(fname)
if base != "arch-support.txt":
if self.debug:
print(f"ignoring {fname}", file=sys.stderr)
return
subsys = os.path.dirname(fname).split("/")[-2]
self.max_size_subsys = max(self.max_size_subsys, len(subsys))
feature_name = ""
kconfig = ""
description = ""
comments = ""
arch_table = {}
if self.debug > 1:
print(f"Opening {fname}", file=sys.stderr)
if self.enable_fname:
full_fname = os.path.abspath(fname)
self.emit(f".. FILE {full_fname}")
with open(fname, encoding="utf-8") as f:
for ln, line in enumerate(f, start=1):
line = line.strip()
match = re.match(r"^\#\s+Feature\s+name:\s*(.*\S)", line)
if match:
feature_name = match.group(1)
self.max_size_name = max(self.max_size_name,
len(feature_name))
continue
match = re.match(r"^\#\s+Kconfig:\s*(.*\S)", line)
if match:
kconfig = match.group(1)
self.max_size_kconfig = max(self.max_size_kconfig,
len(kconfig))
continue
match = re.match(r"^\#\s+description:\s*(.*\S)", line)
if match:
description = match.group(1)
self.max_size_description = max(self.max_size_description,
len(description))
words = re.split(r"\s+", line)[1:]
for word in words:
self.max_size_desc_word = max(self.max_size_desc_word,
len(word))
continue
if re.search(r"^\\s*$", line):
continue
if re.match(r"^\s*\-+\s*$", line):
continue
if re.search(r"^\s*\|\s*arch\s*\|\s*status\s*\|\s*$", line):
continue
match = re.match(r"^\#\s*(.*)$", line)
if match:
comments += match.group(1)
continue
match = re.match(r"^\s*\|\s*(\S+):\s*\|\s*(\S+)\s*\|\s*$", line)
if match:
arch = match.group(1)
status = match.group(2)
self.max_size_status = max(self.max_size_status,
len(status))
self.max_size_arch = max(self.max_size_arch, len(arch))
if status == "..":
status = "---"
arch_table[arch] = status
continue
self.parse_error(fname, ln, "Line is invalid", line)
if not feature_name:
self.parse_error(fname, 0, "Feature name not found")
return
if not subsys:
self.parse_error(fname, 0, "Subsystem not found")
return
if not kconfig:
self.parse_error(fname, 0, "Kconfig not found")
return
if not description:
self.parse_error(fname, 0, "Description not found")
return
if not arch_table:
self.parse_error(fname, 0, "Architecture table not found")
return
self.data[feature_name] = {
"where": fname,
"subsys": subsys,
"kconfig": kconfig,
"description": description,
"comments": comments,
"table": arch_table,
}
self.max_size_arch_with_header = self.max_size_arch + len(self.h_arch)
def parse(self):
"""Parses all arch-support.txt feature files inside self.prefix."""
path = os.path.expanduser(self.prefix)
if self.debug > 2:
print(f"Running parser for {path}")
example_path = os.path.join(path, "arch-support.txt")
for fname in iglob(os.path.join(path, "**"), recursive=True):
if fname != example_path:
self.parse_feat_file(fname)
return self.data
def output_arch_table(self, arch, feat=None):
"""
Output feature(s) for a given architecture.
"""
title = f"Feature status on {arch} architecture"
self.emit("=" * len(title))
self.emit(title)
self.emit("=" * len(title))
self.emit()
self.emit("=" * self.max_size_subsys + " ", end="")
self.emit("=" * self.max_size_name + " ", end="")
self.emit("=" * self.max_size_kconfig + " ", end="")
self.emit("=" * self.max_size_status + " ", end="")
self.emit("=" * self.max_size_description)
self.emit(f"{self.h_subsys:<{self.max_size_subsys}} ", end="")
self.emit(f"{self.h_name:<{self.max_size_name}} ", end="")
self.emit(f"{self.h_kconfig:<{self.max_size_kconfig}} ", end="")
self.emit(f"{self.h_status:<{self.max_size_status}} ", end="")
self.emit(f"{self.h_description:<{self.max_size_description}}")
self.emit("=" * self.max_size_subsys + " ", end="")
self.emit("=" * self.max_size_name + " ", end="")
self.emit("=" * self.max_size_kconfig + " ", end="")
self.emit("=" * self.max_size_status + " ", end="")
self.emit("=" * self.max_size_description)
sorted_features = sorted(self.data.keys(),
key=lambda x: (self.data[x]["subsys"],
x.lower()))
for name in sorted_features:
if feat and name != feat:
continue
arch_table = self.data[name]["table"]
if not arch in arch_table:
continue
self.emit(f"{self.data[name]['subsys']:<{self.max_size_subsys}} ",
end="")
self.emit(f"{name:<{self.max_size_name}} ", end="")
self.emit(f"{self.data[name]['kconfig']:<{self.max_size_kconfig}} ",
end="")
self.emit(f"{arch_table[arch]:<{self.max_size_status}} ",
end="")
self.emit(f"{self.data[name]['description']}")
self.emit("=" * self.max_size_subsys + " ", end="")
self.emit("=" * self.max_size_name + " ", end="")
self.emit("=" * self.max_size_kconfig + " ", end="")
self.emit("=" * self.max_size_status + " ", end="")
self.emit("=" * self.max_size_description)
return self.msg
def output_feature(self, feat):
"""
Output a feature on all architectures.
"""
title = f"Feature {feat}"
self.emit("=" * len(title))
self.emit(title)
self.emit("=" * len(title))
self.emit()
if not feat in self.data:
return
if self.data[feat]["subsys"]:
self.emit(f":Subsystem: {self.data[feat]['subsys']}")
if self.data[feat]["kconfig"]:
self.emit(f":Kconfig: {self.data[feat]['kconfig']}")
desc = self.data[feat]["description"]
desc = desc[0].upper() + desc[1:]
desc = desc.rstrip(". \t")
self.emit(f"\n{desc}.\n")
com = self.data[feat]["comments"].strip()
if com:
self.emit("Comments")
self.emit("--------")
self.emit(f"\n{com}\n")
self.emit("=" * self.max_size_arch + " ", end="")
self.emit("=" * self.max_size_status)
self.emit(f"{self.h_arch:<{self.max_size_arch}} ", end="")
self.emit(f"{self.h_status:<{self.max_size_status}}")
self.emit("=" * self.max_size_arch + " ", end="")
self.emit("=" * self.max_size_status)
arch_table = self.data[feat]["table"]
for arch in sorted(arch_table.keys()):
self.emit(f"{arch:<{self.max_size_arch}} ", end="")
self.emit(f"{arch_table[arch]:<{self.max_size_status}}")
self.emit("=" * self.max_size_arch + " ", end="")
self.emit("=" * self.max_size_status)
return self.msg
def matrix_lines(self, desc_size, max_size_status, header):
"""
Helper function to split element tables at the output matrix.
"""
if header:
ln_marker = "="
else:
ln_marker = "-"
self.emit("+" + ln_marker * self.max_size_name + "+", end="")
self.emit(ln_marker * desc_size, end="")
self.emit("+" + ln_marker * max_size_status + "+")
def output_matrix(self):
"""
Generates a set of tables, groped by subsystem, containing
what's the feature state on each architecture.
"""
title = "Feature status on all architectures"
self.emit("=" * len(title))
self.emit(title)
self.emit("=" * len(title))
self.emit()
desc_title = f"{self.h_kconfig} / {self.h_description}"
desc_size = self.max_size_kconfig + 4
if not self.description_size:
desc_size = max(self.max_size_description, desc_size)
else:
desc_size = max(self.description_size, desc_size)
desc_size = max(self.max_size_desc_word, desc_size, len(desc_title))
notcompat = "Not compatible"
self.max_size_status = max(self.max_size_status, len(notcompat))
min_status_size = self.max_size_status + self.max_size_arch + 4
max_size_status = max(min_status_size, self.max_size_status)
h_status_per_arch = "Status per architecture"
max_size_status = max(max_size_status, len(h_status_per_arch))
cur_subsys = None
for name in sorted(self.data.keys(),
key=lambda x: (self.data[x]["subsys"], x.lower())):
if not cur_subsys or cur_subsys != self.data[name]["subsys"]:
if cur_subsys:
self.emit()
cur_subsys = self.data[name]["subsys"]
title = f"Subsystem: {cur_subsys}"
self.emit(title)
self.emit("=" * len(title))
self.emit()
self.matrix_lines(desc_size, max_size_status, 0)
self.emit(f"|{self.h_name:<{self.max_size_name}}", end="")
self.emit(f"|{desc_title:<{desc_size}}", end="")
self.emit(f"|{h_status_per_arch:<{max_size_status}}|")
self.matrix_lines(desc_size, max_size_status, 1)
lines = []
descs = []
cur_status = ""
line = ""
arch_table = sorted(self.data[name]["table"].items(),
key=lambda x: (self.status_map.get(x[1], 99),
x[0].lower()))
for arch, status in arch_table:
if status == "---":
status = notcompat
if status != cur_status:
if line != "":
lines.append(line)
line = ""
line = f"- **{status}**: {arch}"
elif len(line) + len(arch) + 2 < max_size_status:
line += f", {arch}"
else:
lines.append(line)
line = f" {arch}"
cur_status = status
if line != "":
lines.append(line)
description = self.data[name]["description"]
while len(description) > desc_size:
desc_line = description[:desc_size]
last_space = desc_line.rfind(" ")
if last_space != -1:
desc_line = desc_line[:last_space]
descs.append(desc_line)
description = description[last_space + 1:]
else:
desc_line = desc_line[:-1]
descs.append(desc_line + "\\")
description = description[len(desc_line):]
if description:
descs.append(description)
while len(lines) < 2 + len(descs):
lines.append("")
for ln, line in enumerate(lines):
col = ["", ""]
if not ln:
col[0] = name
col[1] = f"``{self.data[name]['kconfig']}``"
else:
if ln >= 2 and descs:
col[1] = descs.pop(0)
self.emit(f"|{col[0]:<{self.max_size_name}}", end="")
self.emit(f"|{col[1]:<{desc_size}}", end="")
self.emit(f"|{line:<{max_size_status}}|")
self.matrix_lines(desc_size, max_size_status, 0)
return self.msg
def list_arch_features(self, arch, feat):
"""
Print a matrix of kernel feature support for the chosen architecture.
"""
self.emit("#")
self.emit(f"# Kernel feature support matrix of the '{arch}' architecture:")
self.emit("#")
# Sort by subsystem, then by feature name (case‑insensitive)
for name in sorted(self.data.keys(),
key=lambda n: (self.data[n]["subsys"].lower(),
n.lower())):
if feat and name != feat:
continue
feature = self.data[name]
arch_table = feature["table"]
status = arch_table.get(arch, "")
status = " " * ((4 - len(status)) // 2) + status
self.emit(f"{feature['subsys']:>{self.max_size_subsys + 1}}/ ",
end="")
self.emit(f"{name:<{self.max_size_name}}: ", end="")
self.emit(f"{status:<5}| ", end="")
self.emit(f"{feature['kconfig']:>{self.max_size_kconfig}} ",
end="")
self.emit(f"# {feature['description']}")
return self.msg
linux-7.2.3/tools/lib/python/jobserver.py
Executable
+195
View File
@@ -0,0 +1,195 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0+
#
# pylint: disable=C0103,C0209
#
#
"""
Interacts with the POSIX jobserver during the Kernel build time.
A "normal" jobserver task, like the one initiated by a make subprocess would do:
- open read/write file descriptors to communicate with the job server;
- ask for one slot by calling::
claim = os.read(reader, 1)
- when the job finishes, call::
os.write(writer, b"+") # os.write(writer, claim)
Here, the goal is different: This script aims to get the remaining number
of slots available, using all of them to run a command which handle tasks in
parallel. To to that, it has a loop that ends only after there are no
slots left. It then increments the number by one, in order to allow a
call equivalent to ``make -j$((claim+1))``, e.g. having a parent make creating
$claim child to do the actual work.
The end goal here is to keep the total number of build tasks under the
limit established by the initial ``make -j$n_proc`` call.
See:
https://www.gnu.org/software/make/manual/html_node/POSIX-Jobserver.html#POSIX-Jobserver
"""
import errno
import os
import subprocess
import sys
def warn(text, *args):
print(f'WARNING: {text}', *args, file = sys.stderr)
class JobserverExec:
"""
Claim all slots from make using POSIX Jobserver.
The main methods here are:
- open(): reserves all slots;
- close(): method returns all used slots back to make;
- run(): executes a command setting PARALLELISM=<available slots jobs + 1>.
"""
def __init__(self):
"""Initialize internal vars."""
self.claim = 0
self.jobs = b""
self.reader = None
self.writer = None
self.is_open = False
def open(self):
"""Reserve all available slots to be claimed later on."""
if self.is_open:
return
self.is_open = True # We only try once
self.claim = None
#
# Check the make flags for "--jobserver=R,W"
# Note that GNU Make has used --jobserver-fds and --jobserver-auth
# so this handles all of them.
#
flags = os.environ.get('MAKEFLAGS', '')
opts = [x for x in flags.split(" ") if x.startswith("--jobserver")]
if not opts:
return
#
# Separate out the provided file descriptors
#
split_opt = opts[-1].split('=', 1)
if len(split_opt) != 2:
warn('unparseable option:', opts[-1])
return
fds = split_opt[1]
#
# As of GNU Make 4.4, we'll be looking for a named pipe
# identified as fifo:path
#
if fds.startswith('fifo:'):
path = fds[len('fifo:'):]
try:
self.reader = os.open(path, os.O_RDONLY | os.O_NONBLOCK)
self.writer = os.open(path, os.O_WRONLY)
except (OSError, IOError):
warn('unable to open jobserver pipe', path)
return
#
# Otherwise look for integer file-descriptor numbers.
#
else:
split_fds = fds.split(',')
if len(split_fds) != 2:
warn('malformed jobserver file descriptors:', fds)
return
try:
self.reader = int(split_fds[0])
self.writer = int(split_fds[1])
except ValueError:
warn('non-integer jobserver file-descriptors:', fds)
return
try:
#
# Open a private copy of reader to avoid setting nonblocking
# on an unexpecting process with the same reader fd.
#
self.reader = os.open(f"/proc/self/fd/{self.reader}",
os.O_RDONLY | os.O_NONBLOCK)
except (IOError, OSError) as e:
warn('Unable to reopen jobserver read-side pipe:', repr(e))
return
#
# OK, we have the channel to the job server; read out as many jobserver
# slots as possible.
#
while True:
try:
slot = os.read(self.reader, 8)
if not slot:
#
# Something went wrong. Clear self.jobs to avoid writing
# weirdness back to the jobserver and give up.
self.jobs = b""
warn("unexpected empty token from jobserver;"
" possible invalid '--jobserver-auth=' setting")
self.claim = None
return
except (OSError, IOError) as e:
#
# If there is nothing more to read then we are done.
#
if e.errno == errno.EWOULDBLOCK:
break
#
# Anything else says that something went weird; give back
# the jobs and give up.
#
if self.jobs:
os.write(self.writer, self.jobs)
self.claim = None
warn('error reading from jobserver pipe', repr(e))
return
self.jobs += slot
#
# Add a bump for our caller's reserveration, since we're just going
# to sit here blocked on our child.
#
self.claim = len(self.jobs) + 1
def close(self):
"""Return all reserved slots to Jobserver."""
if not self.is_open:
return
# Return all the reserved slots.
if len(self.jobs):
os.write(self.writer, self.jobs)
self.is_open = False
def __enter__(self):
self.open()
return self
def __exit__(self, exc_type, exc_value, exc_traceback):
self.close()
def run(self, cmd, *args, **pwargs):
"""
Run a command setting PARALLELISM env variable to the number of
available job slots (claim) + 1, e.g. it will reserve claim slots
to do the actual build work, plus one to monitor its children.
"""
self.open() # Ensure that self.claim is set
# We can only claim parallelism if there was a jobserver (i.e. a
# top-level "-jN" argument) and there were no other failures. Otherwise
# leave out the environment variable and let the child figure out what
# is best.
if self.claim:
os.environ["PARALLELISM"] = str(self.claim)
return subprocess.call(cmd, *args, **pwargs)
linux-7.2.3/tools/lib/python/kdoc/__init__.py
View File
linux-7.2.3/tools/lib/python/kdoc/c_lex.py
+662
View File
@@ -0,0 +1,662 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
# Copyright(c) 2025: Mauro Carvalho Chehab <mchehab@kernel.org>.
"""
Regular expression ancillary classes.
Those help caching regular expressions and do matching for kernel-doc.
Please notice that the code here may rise exceptions to indicate bad
usage inside kdoc to indicate problems at the replace pattern.
Other errors are logged via log instance.
"""
import logging
import re
from copy import copy
from .kdoc_re import KernRe
log = logging.getLogger(__name__)
def tokenizer_set_log(logger, prefix = ""):
"""
Replace the module‑level logger with a LoggerAdapter that
prepends *prefix* to every message.
"""
global log
class PrefixAdapter(logging.LoggerAdapter):
"""
Ancillary class to set prefix on all message logs.
"""
def process(self, msg, kwargs):
return f"{prefix}{msg}", kwargs
# Wrap the provided logger in our adapter
log = PrefixAdapter(logger, {"prefix": prefix})
class CToken():
"""
Data class to define a C token.
"""
# Tokens that can be used by the parser. Works like an C enum.
COMMENT = 0 #: A standard C or C99 comment, including delimiter.
STRING = 1 #: A string, including quotation marks.
CHAR = 2 #: A character, including apostophes.
NUMBER = 3 #: A number.
PUNC = 4 #: A puntuation mark: / ``,`` / ``.``.
BEGIN = 5 #: A begin character: ``{`` / ``[`` / ``(``.
END = 6 #: A end character: ``}`` / ``]`` / ``)``.
CPP = 7 #: A preprocessor macro.
HASH = 8 #: The hash character - useful to handle other macros.
OP = 9 #: A C operator (add, subtract, ...).
STRUCT = 10 #: A ``struct`` keyword.
UNION = 11 #: An ``union`` keyword.
ENUM = 12 #: A ``struct`` keyword.
TYPEDEF = 13 #: A ``typedef`` keyword.
NAME = 14 #: A name. Can be an ID or a type.
SPACE = 15 #: Any space characters, including new lines
ENDSTMT = 16 #: End of an statement (``;``).
BACKREF = 17 #: Not a valid C sequence, but used at sub regex patterns.
MISMATCH = 255 #: an error indicator: should never happen in practice.
# Dict to convert from an enum interger into a string.
_name_by_val = {v: k for k, v in dict(vars()).items() if isinstance(v, int)}
# Dict to convert from string to an enum-like integer value.
_name_to_val = {k: v for v, k in _name_by_val.items()}
@staticmethod
def to_name(val):
"""Convert from an integer value from CToken enum into a string"""
return CToken._name_by_val.get(val, f"UNKNOWN({val})")
@staticmethod
def from_name(name):
"""Convert a string into a CToken enum value"""
if name in CToken._name_to_val:
return CToken._name_to_val[name]
return CToken.MISMATCH
def __init__(self, kind, value=None, pos=0,
brace_level=0, paren_level=0, bracket_level=0):
self.kind = kind
self.value = value
self.pos = pos
self.level = (bracket_level, paren_level, brace_level)
def __repr__(self):
name = self.to_name(self.kind)
if isinstance(self.value, str):
value = '"' + self.value + '"'
else:
value = self.value
return f"CToken(CToken.{name}, {value}, {self.pos}, {self.level})"
#: Regexes to parse C code, transforming it into tokens.
RE_SCANNER_LIST = [
#
# Note that \s\S is different than .*, as it also catches \n
#
(CToken.COMMENT, r"//[^\n]*|/\*[\s\S]*?\*/"),
(CToken.STRING, r'"(?:\\.|[^"\\])*"'),
(CToken.CHAR, r"'(?:\\.|[^'\\])'"),
(CToken.NUMBER, r"0[xX][\da-fA-F]+[uUlL]*|0[0-7]+[uUlL]*|"
r"\d+(?:\.\d*)?(?:[eE][+-]?\d+)?[fFlL]*"),
(CToken.ENDSTMT, r"(?:\s+;|;)"),
(CToken.PUNC, r"[,\.]"),
(CToken.BEGIN, r"[\[\(\{]"),
(CToken.END, r"[\]\)\}]"),
(CToken.CPP, r"#\s*(?:define|include|ifdef|ifndef|if|else|elif|endif|undef|pragma)\b"),
(CToken.HASH, r"#"),
(CToken.OP, r"\+\+|\-\-|\->|==|\!=|<=|>=|&&|\|\||<<|>>|\+=|\-=|\*=|/=|%="
r"|&=|\|=|\^=|[=\+\-\*/%<>&\|\^~!\?\:]"),
(CToken.STRUCT, r"\bstruct\b"),
(CToken.UNION, r"\bunion\b"),
(CToken.ENUM, r"\benum\b"),
(CToken.TYPEDEF, r"\btypedef\b"),
(CToken.NAME, r"[A-Za-z_]\w*"),
(CToken.SPACE, r"\s+"),
(CToken.BACKREF, r"\\\d+"),
(CToken.MISMATCH,r"."),
]
def fill_re_scanner(token_list):
"""Ancillary routine to convert RE_SCANNER_LIST into a finditer regex"""
re_tokens = []
for kind, pattern in token_list:
name = CToken.to_name(kind)
re_tokens.append(f"(?P<{name}>{pattern})")
return KernRe("|".join(re_tokens), re.MULTILINE | re.DOTALL)
#: Handle C continuation lines.
RE_CONT = KernRe(r"\\\n")
RE_COMMENT_START = KernRe(r'/\*\s*')
#: tokenizer regex. Will be filled at the first CTokenizer usage.
RE_SCANNER = fill_re_scanner(RE_SCANNER_LIST)
class CTokenizer():
"""
Scan C statements and definitions and produce tokens.
When converted to string, it drops comments and handle public/private
values, respecting depth.
"""
# This class is inspired and follows the basic concepts of:
# https://docs.python.org/3/library/re.html#writing-a-tokenizer
def __init__(self, source=None):
"""
Create a regular expression to handle RE_SCANNER_LIST.
While I generally don't like using regex group naming via:
(?P<name>...)
in this particular case, it makes sense, as we can pick the name
when matching a code via RE_SCANNER.
"""
#
# Store logger to allow parser classes to re-use it
#
global log
self.log = log
self.tokens = []
if not source:
return
if isinstance(source, list):
self.tokens = source
return
#
# While we could just use _tokenize directly via interator,
# As we'll need to use the tokenizer several times inside kernel-doc
# to handle macro transforms, cache the results on a list, as
# re-using it is cheaper than having to parse everytime.
#
for tok in self._tokenize(source):
self.tokens.append(tok)
def _tokenize(self, source):
"""
Iterator that parses ``source``, splitting it into tokens, as defined
at ``self.RE_SCANNER_LIST``.
The interactor returns a CToken class object.
"""
# Handle continuation lines. Note that kdoc_parser already has a
# logic to do that. Still, let's keep it for completeness, as we might
# end re-using this tokenizer outsize kernel-doc some day - or we may
# eventually remove from there as a future cleanup.
source = RE_CONT.sub("", source)
brace_level = 0
paren_level = 0
bracket_level = 0
for match in RE_SCANNER.finditer(source):
kind = CToken.from_name(match.lastgroup)
pos = match.start()
value = match.group()
if kind == CToken.MISMATCH:
log.error(f"Unexpected token '{value}' on pos {pos}:\n\t'{source}'")
elif kind == CToken.BEGIN:
if value == '(':
paren_level += 1
elif value == '[':
bracket_level += 1
else: # value == '{'
brace_level += 1
elif kind == CToken.END:
if value == ')' and paren_level > 0:
paren_level -= 1
elif value == ']' and bracket_level > 0:
bracket_level -= 1
elif brace_level > 0: # value == '}'
brace_level -= 1
yield CToken(kind, value, pos,
brace_level, paren_level, bracket_level)
def __str__(self):
out=""
show_stack = [True]
for i, tok in enumerate(self.tokens):
if tok.kind == CToken.BEGIN:
show_stack.append(show_stack[-1])
elif tok.kind == CToken.END:
prev = show_stack[-1]
if len(show_stack) > 1:
show_stack.pop()
if not prev and show_stack[-1]:
#
# Try to preserve indent
#
out += "\t" * (len(show_stack) - 1)
out += str(tok.value)
continue
elif tok.kind == CToken.COMMENT:
comment = RE_COMMENT_START.sub("", tok.value)
if comment.startswith("private:"):
show_stack[-1] = False
show = False
elif comment.startswith("public:"):
show_stack[-1] = True
continue
if not show_stack[-1]:
continue
if i < len(self.tokens) - 1:
next_tok = self.tokens[i + 1]
# Do some cleanups before ";"
if tok.kind == CToken.SPACE and next_tok.kind == CToken.ENDSTMT:
continue
if tok.kind == CToken.ENDSTMT and next_tok.kind == tok.kind:
continue
out += str(tok.value)
return out
class CTokenArgs:
"""
Ancillary class to help using backrefs from sub matches.
If the highest backref contain a "+" at the last element,
the logic will be greedy, picking all other delims.
This is needed to parse struct_group macros with end with ``MEMBERS...``.
"""
def __init__(self, sub_str):
self.sub_groups = set()
self.max_group = -1
self.greedy = None
for m in KernRe(r'\\(\d+)([+]?)').finditer(sub_str):
group = int(m.group(1))
if m.group(2) == "+":
if self.greedy and self.greedy != group:
raise ValueError("There are multiple greedy patterns!")
self.greedy = group
self.sub_groups.add(group)
self.max_group = max(self.max_group, group)
if self.greedy:
if self.greedy != self.max_group:
raise ValueError("Greedy pattern is not the last one!")
sub_str = KernRe(r'(\\\d+)[+]').sub(r"\1", sub_str)
self.sub_str = sub_str
self.sub_tokeninzer = CTokenizer(sub_str)
def groups(self, new_tokenizer):
r"""
Create replacement arguments for backrefs like:
``\0``, ``\1``, ``\2``, ... ``\{number}``
It also accepts a ``+`` character to the highest backref, like
``\4+``. When used, the backref will be greedy, picking all other
arguments afterwards.
The logic is smart enough to only go up to the maximum required
argument, even if there are more.
If there is a backref for an argument above the limit, it will
raise an exception. Please notice that, on C, square brackets
don't have any separator on it. Trying to use ``\1``..``\n`` for
brackets also raise an exception.
"""
level = (0, 0, 0)
if self.max_group < 0:
return level, []
tokens = new_tokenizer.tokens
#
# Fill \0 with the full token contents
#
groups_list = [ [] ]
if 0 in self.sub_groups:
inner_level = 0
for i in range(0, len(tokens)):
tok = tokens[i]
if tok.kind == CToken.BEGIN:
inner_level += 1
#
# Discard first begin
#
if not groups_list[0]:
continue
elif tok.kind == CToken.END:
inner_level -= 1
if inner_level < 0:
break
if inner_level:
groups_list[0].append(tok)
if not self.max_group:
return level, groups_list
delim = None
#
# Ignore everything before BEGIN. The value of begin gives the
# delimiter to be used for the matches
#
for i in range(0, len(tokens)):
tok = tokens[i]
if tok.kind == CToken.BEGIN:
if tok.value == "{":
delim = ";"
elif tok.value == "(":
delim = ","
else:
self.log.error(fr"Can't handle \1..\n on {sub_str}")
level = tok.level
break
pos = 1
groups_list.append([])
inner_level = 0
for i in range(i + 1, len(tokens)):
tok = tokens[i]
if tok.kind == CToken.BEGIN:
inner_level += 1
if tok.kind == CToken.END:
inner_level -= 1
if inner_level < 0:
break
if tok.kind in [CToken.PUNC, CToken.ENDSTMT] and delim == tok.value:
pos += 1
if self.greedy and pos > self.max_group:
pos -= 1
else:
groups_list.append([])
if pos > self.max_group:
break
continue
groups_list[pos].append(tok)
if pos < self.max_group:
log.error(fr"{self.sub_str} groups are up to {pos} instead of {self.max_group}")
return level, groups_list
def tokens(self, new_tokenizer):
level, groups = self.groups(new_tokenizer)
new = CTokenizer()
for tok in self.sub_tokeninzer.tokens:
if tok.kind == CToken.BACKREF:
group = int(tok.value[1:])
for group_tok in groups[group]:
new_tok = copy(group_tok)
new_level = [0, 0, 0]
for i in range(0, len(level)):
new_level[i] = new_tok.level[i] + level[i]
new_tok.level = tuple(new_level)
new.tokens += [ new_tok ]
else:
new.tokens += [ tok ]
return new.tokens
class CMatch:
"""
Finding nested delimiters is hard with regular expressions. It is
even harder on Python with its normal re module, as there are several
advanced regular expressions that are missing.
This is the case of this pattern::
'\\bSTRUCT_GROUP(\\(((?:(?>[^)(]+)|(?1))*)\\))[^;]*;'
which is used to properly match open/close parentheses of the
string search STRUCT_GROUP(),
Add a class that counts pairs of delimiters, using it to match and
replace nested expressions.
The original approach was suggested by:
https://stackoverflow.com/questions/5454322/python-how-to-match-nested-parentheses-with-regex
Although I re-implemented it to make it more generic and match 3 types
of delimiters. The logic checks if delimiters are paired. If not, it
will ignore the search string.
"""
def __init__(self, regex, delim="("):
self.regex = KernRe("^" + regex + r"\b")
self.start_delim = delim
def _search(self, tokenizer):
"""
Finds paired blocks for a regex that ends with a delimiter.
The suggestion of using finditer to match pairs came from:
https://stackoverflow.com/questions/5454322/python-how-to-match-nested-parentheses-with-regex
but I ended using a different implementation to align all three types
of delimiters and seek for an initial regular expression.
The algorithm seeks for open/close paired delimiters and places them
into a stack, yielding a start/stop position of each match when the
stack is zeroed.
The algorithm should work fine for properly paired lines, but will
silently ignore end delimiters that precede a start delimiter.
This should be OK for kernel-doc parser, as unaligned delimiters
would cause compilation errors. So, we don't need to raise exceptions
to cover such issues.
"""
start = None
started = False
import sys
stack = []
for i, tok in enumerate(tokenizer.tokens):
if start is None:
if tok.kind == CToken.NAME and self.regex.match(tok.value):
start = i
stack.append((start, tok.level))
started = False
continue
if not started:
if tok.kind == CToken.SPACE:
continue
if tok.kind == CToken.BEGIN and tok.value == self.start_delim:
started = True
continue
# Name only token without BEGIN/END
if i > start:
i -= 1
yield start, i
start = None
if tok.kind == CToken.END and tok.level == stack[-1][1]:
start, level = stack.pop()
yield start, i
start = None
#
# If an END zeroing levels is not there, return remaining stuff
# This is meant to solve cases where the caller logic might be
# picking an incomplete block.
#
if start and stack:
if started:
s = str(tokenizer)
log.warning(f"can't find a final end at {s}")
yield start, len(tokenizer.tokens)
def search(self, source):
"""
This is similar to re.search:
It matches a regex that it is followed by a delimiter,
returning occurrences only if all delimiters are paired.
"""
if isinstance(source, CTokenizer):
tokenizer = source
is_token = True
else:
tokenizer = CTokenizer(source)
is_token = False
for start, end in self._search(tokenizer):
new_tokenizer = CTokenizer(tokenizer.tokens[start:end + 1])
if is_token:
yield new_tokenizer
else:
yield str(new_tokenizer)
def sub(self, sub_str, source, count=0):
"""
This is similar to re.sub:
It matches a regex that it is followed by a delimiter,
replacing occurrences only if all delimiters are paired.
if the sub argument contains::
r'\0'
it will work just like re: it places there the matched paired data
with the delimiter stripped.
If count is different than zero, it will replace at most count
items.
"""
if isinstance(source, CTokenizer):
is_token = True
tokenizer = source
else:
is_token = False
tokenizer = CTokenizer(source)
# Detect if sub_str contains sub arguments
args_match = CTokenArgs(sub_str)
new_tokenizer = CTokenizer()
pos = 0
n = 0
#
# NOTE: the code below doesn't consider overlays at sub.
# We may need to add some extra unit tests to check if those
# would cause problems. When replacing by "", this should not
# be a problem, but other transformations could be problematic
#
for start, end in self._search(tokenizer):
new_tokenizer.tokens += tokenizer.tokens[pos:start]
new = CTokenizer(tokenizer.tokens[start:end + 1])
new_tokenizer.tokens += args_match.tokens(new)
pos = end + 1
n += 1
if count and n >= count:
break
new_tokenizer.tokens += tokenizer.tokens[pos:]
if not is_token:
return str(new_tokenizer)
return new_tokenizer
def __repr__(self):
"""
Returns a displayable version of the class init.
"""
return f'CMatch("{self.regex.regex.pattern}")'
linux-7.2.3/tools/lib/python/kdoc/enrich_formatter.py
+80
View File
@@ -0,0 +1,80 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
# Copyright (c) 2025 by Mauro Carvalho Chehab <mchehab@kernel.org>.
"""
Ancillary argparse HelpFormatter class that works on a similar way as
argparse.RawDescriptionHelpFormatter, e.g. description maintains line
breaks, but it also implement transformations to the help text. The
actual transformations ar given by enrich_text(), if the output is tty.
Currently, the follow transformations are done:
- Positional arguments are shown in upper cases;
- if output is TTY, ``var`` and positional arguments are shown prepended
by an ANSI SGR code. This is usually translated to bold. On some
terminals, like, konsole, this is translated into a colored bold text.
"""
import argparse
import re
import sys
class EnrichFormatter(argparse.HelpFormatter):
"""
Better format the output, making easier to identify the positional args
and how they're used at the __doc__ description.
"""
def __init__(self, *args, **kwargs):
"""
Initialize class and check if is TTY.
"""
super().__init__(*args, **kwargs)
self._tty = sys.stdout.isatty()
def enrich_text(self, text):
r"""
Handle ReST markups (currently, only \`\`text\`\` markups).
"""
if self._tty and text:
# Replace ``text`` with ANSI SGR (bold)
return re.sub(r'\`\`(.+?)\`\`',
lambda m: f'\033[1m{m.group(1)}\033[0m', text)
return text
def _fill_text(self, text, width, indent):
"""
Enrich descriptions with markups on it.
"""
enriched = self.enrich_text(text)
return "\n".join(indent + line for line in enriched.splitlines())
def _format_usage(self, usage, actions, groups, prefix):
"""
Enrich positional arguments at usage: line.
"""
prog = self._prog
parts = []
for action in actions:
if action.option_strings:
opt = action.option_strings[0]
if action.nargs != 0:
opt += f" {action.dest.upper()}"
parts.append(f"[{opt}]")
else:
# Positional argument
parts.append(self.enrich_text(f"``{action.dest.upper()}``"))
usage_text = f"{prefix or 'usage: '} {prog} {' '.join(parts)}\n"
return usage_text
def _format_action_invocation(self, action):
"""
Enrich argument names.
"""
if not action.option_strings:
return self.enrich_text(f"``{action.dest.upper()}``")
return ", ".join(action.option_strings)
linux-7.2.3/tools/lib/python/kdoc/kdoc_files.py
+380
View File
@@ -0,0 +1,380 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
# Copyright(c) 2025: Mauro Carvalho Chehab <mchehab@kernel.org>.
#
# pylint: disable=R0903,R0913,R0914,R0917
"""
Classes for navigating through the files that kernel-doc needs to handle
to generate documentation.
"""
import logging
import os
import re
from kdoc.kdoc_parser import KernelDoc
from kdoc.xforms_lists import CTransforms
from kdoc.kdoc_output import OutputFormat
from kdoc.kdoc_yaml_file import KDocTestFile
class GlobSourceFiles:
"""
Parse C source code file names and directories via an Interactor.
"""
def __init__(self, srctree=None, valid_extensions=None):
"""
Initialize valid extensions with a tuple.
If not defined, assume default C extensions (.c and .h)
It would be possible to use python's glob function, but it is
very slow, and it is not interactive. So, it would wait to read all
directories before actually do something.
So, let's use our own implementation.
"""
if not valid_extensions:
self.extensions = (".c", ".h")
else:
self.extensions = valid_extensions
self.srctree = srctree
def _parse_dir(self, dirname):
"""Internal function to parse files recursively."""
with os.scandir(dirname) as obj:
for entry in obj:
name = os.path.join(dirname, entry.name)
if entry.is_dir(follow_symlinks=False):
yield from self._parse_dir(name)
if not entry.is_file():
continue
basename = os.path.basename(name)
if not basename.endswith(self.extensions):
continue
yield name
def parse_files(self, file_list, file_not_found_cb):
"""
Define an iterator to parse all source files from file_list,
handling directories if any.
"""
if not file_list:
return
for fname in file_list:
if self.srctree:
f = os.path.join(self.srctree, fname)
else:
f = fname
if os.path.isdir(f):
yield from self._parse_dir(f)
elif os.path.isfile(f):
yield f
elif file_not_found_cb:
file_not_found_cb(fname)
class KdocConfig():
"""
Stores all configuration attributes that kdoc_parser and kdoc_output
needs.
"""
def __init__(self, verbose=False, werror=False, wreturn=False,
wshort_desc=False, wcontents_before_sections=False,
logger=None):
self.verbose = verbose
self.werror = werror
self.wreturn = wreturn
self.wshort_desc = wshort_desc
self.wcontents_before_sections = wcontents_before_sections
if logger:
self.log = logger
else:
self.log = logging.getLogger(__file__)
self.warning = self.log.warning
class KernelFiles():
"""
Parse kernel-doc tags on multiple kernel source files.
This is the main entry point to run kernel-doc. This class is initialized
using a series of optional arguments:
``verbose``
If True, enables kernel-doc verbosity. Default: False.
``out_style``
Class to be used to format output. If None (default),
only report errors.
``xforms``
Transforms to be applied to C prototypes and data structs.
If not specified, defaults to xforms = CFunction()
``werror``
If True, treat warnings as errors, retuning an error code on warnings.
Default: False.
``wreturn``
If True, warns about the lack of a return markup on functions.
Default: False.
``wshort_desc``
If True, warns if initial short description is missing.
Default: False.
``wcontents_before_sections``
If True, warn if there are contents before sections (deprecated).
This option is kept just for backward-compatibility, but it does
nothing, neither here nor at the original Perl script.
Default: False.
``logger``
Optional logger class instance.
If not specified, defaults to use: ``logging.getLogger("kernel-doc")``
``yaml_file``
If defined, stores the output inside a YAML file.
``yaml_content``
Defines what will be inside the YAML file.
Note:
There are two type of parsers defined here:
- self.parse_file(): parses both kernel-doc markups and
``EXPORT_SYMBOL*`` macros;
- self.process_export_file(): parses only ``EXPORT_SYMBOL*`` macros.
"""
def warning(self, msg):
"""Ancillary routine to output a warning and increment error count."""
self.config.log.warning(msg)
self.errors += 1
def error(self, msg):
"""Ancillary routine to output an error and increment error count."""
self.config.log.error(msg)
self.errors += 1
def parse_file(self, fname):
"""
Parse a single Kernel source.
"""
# Prevent parsing the same file twice if results are cached
if fname in self.files:
return
if self.test_file:
store_src = True
else:
store_src = False
doc = KernelDoc(self.config, fname, self.xforms, store_src=store_src)
export_table, entries = doc.parse_kdoc()
self.export_table[fname] = export_table
self.files.add(fname)
self.export_files.add(fname) # parse_kdoc() already check exports
self.results[fname] = entries
def process_export_file(self, fname):
"""
Parses ``EXPORT_SYMBOL*`` macros from a single Kernel source file.
"""
# Prevent parsing the same file twice if results are cached
if fname in self.export_files:
return
doc = KernelDoc(self.config, fname)
export_table = doc.parse_export()
if not export_table:
self.error(f"Error: Cannot check EXPORT_SYMBOL* on {fname}")
export_table = set()
self.export_table[fname] = export_table
self.export_files.add(fname)
def file_not_found_cb(self, fname):
"""
Callback to warn if a file was not found.
"""
self.error(f"Cannot find file {fname}")
def __init__(self, verbose=False, out_style=None, xforms=None,
werror=False, wreturn=False, wshort_desc=False,
wcontents_before_sections=False,
yaml_file=None, yaml_content=None, logger=None):
"""
Initialize startup variables and parse all files.
"""
if not verbose:
try:
verbose = bool(int(os.environ.get("KBUILD_VERBOSE", 0)))
except ValueError:
# Handles an eventual case where verbosity is not a number
# like KBUILD_VERBOSE=""
verbose = False
if out_style is None:
out_style = OutputFormat()
if not werror:
kcflags = os.environ.get("KCFLAGS", None)
if kcflags:
match = re.search(r"(\s|^)-Werror(\s|$)/", kcflags)
if match:
werror = True
# reading this variable is for backwards compat just in case
# someone was calling it with the variable from outside the
# kernel's build system
kdoc_werror = os.environ.get("KDOC_WERROR", None)
if kdoc_werror:
werror = kdoc_werror
if not logger:
logger = logging.getLogger("kernel-doc")
else:
logger = logger
# Some variables are global to the parser logic as a whole as they are
# used to send control configuration to KernelDoc class. As such,
# those variables are read-only inside the KernelDoc.
self.config = KdocConfig(verbose, werror, wreturn, wshort_desc,
wcontents_before_sections, logger)
# Override log warning, as we want to count errors
self.config.warning = self.warning
if yaml_file:
self.test_file = KDocTestFile(self.config, yaml_file, yaml_content)
else:
self.test_file = None
if xforms:
self.xforms = xforms
else:
self.xforms = CTransforms()
self.config.src_tree = os.environ.get("SRCTREE", None)
# Initialize variables that are internal to KernelFiles
self.out_style = out_style
self.out_style.set_config(self.config)
self.errors = 0
self.results = {}
self.files = set()
self.export_files = set()
self.export_table = {}
def parse(self, file_list, export_file=None):
"""
Parse all files.
"""
glob = GlobSourceFiles(srctree=self.config.src_tree)
for fname in glob.parse_files(file_list, self.file_not_found_cb):
self.parse_file(fname)
for fname in glob.parse_files(export_file, self.file_not_found_cb):
self.process_export_file(fname)
def out_msg(self, fname, name, arg):
"""
Return output messages from a file name using the output style
filtering.
If output type was not handled by the styler, return None.
"""
# NOTE: we can add rules here to filter out unwanted parts,
# although OutputFormat.msg already does that.
return self.out_style.msg(fname, name, arg)
def msg(self, enable_lineno=False, export=False, internal=False,
symbol=None, nosymbol=None, no_doc_sections=False,
filenames=None, export_file=None):
"""
Interacts over the kernel-doc results and output messages,
returning kernel-doc markups on each interaction.
"""
if not filenames:
filenames = sorted(self.results.keys())
glob = GlobSourceFiles(srctree=self.config.src_tree)
for fname in filenames:
function_table = set()
if internal or export:
if not export_file:
export_file = [fname]
for f in glob.parse_files(export_file, self.file_not_found_cb):
function_table |= self.export_table[f]
if symbol:
for s in symbol:
function_table.add(s)
if fname not in self.results:
self.config.log.warning("No kernel-doc for file %s", fname)
continue
symbols = self.results[fname]
if self.test_file:
self.test_file.set_filter(export, internal, symbol, nosymbol,
function_table, enable_lineno,
no_doc_sections)
self.test_file.output_symbols(fname, symbols)
continue
self.out_style.set_filter(export, internal, symbol, nosymbol,
function_table, enable_lineno,
no_doc_sections)
msg = self.out_style.output_symbols(fname, symbols)
if msg:
yield fname, msg
if self.test_file:
self.test_file.write()
linux-7.2.3/tools/lib/python/kdoc/kdoc_item.py
+98
View File
@@ -0,0 +1,98 @@
# SPDX-License-Identifier: GPL-2.0
#
# A class that will, eventually, encapsulate all of the parsed data that we
# then pass into the output modules.
#
"""
Data class to store a kernel-doc Item.
"""
class KdocItem:
"""
A class that will, eventually, encapsulate all of the parsed data that we
then pass into the output modules.
"""
def __init__(self, name, fname, type, start_line,
**other_stuff):
self.name = name
self.fname = fname
self.type = type
self.declaration_start_line = start_line
self.sections = {}
self.sections_start_lines = {}
self.parameterlist = []
self.parameterdesc_start_lines = {}
self.parameterdescs = {}
self.parametertypes = {}
self.warnings = []
#
# Just save everything else into our own dict so that the output
# side can grab it directly as before. As we move things into more
# structured data, this will, hopefully, fade away.
#
known_keys = {
'declaration_start_line',
'sections',
'sections_start_lines',
'parameterlist',
'parameterdesc_start_lines',
'parameterdescs',
'parametertypes',
'warnings',
}
self.other_stuff = {}
for k, v in other_stuff.items():
if k in known_keys:
setattr(self, k, v) # real attribute
else:
self.other_stuff[k] = v
def get(self, key, default = None):
"""
Get a value from optional keys.
"""
return self.other_stuff.get(key, default)
def __getitem__(self, key):
return self.get(key)
def __repr__(self):
return f"KdocItem({self.name}, {self.fname}, {self.type}, {self.declaration_start_line})"
@classmethod
def from_dict(cls, d):
"""Create a KdocItem from a plain dict."""
cp = d.copy()
name = cp.pop('name', None)
fname = cp.pop('fname', None)
type = cp.pop('type', None)
start_line = cp.pop('start_line', 1)
other_stuff = cp.pop('other_stuff', {})
# Everything that’s left goes straight to __init__
return cls(name, fname, type, start_line, **cp, **other_stuff)
#
# Tracking of section and parameter information.
#
def set_sections(self, sections, start_lines):
"""
Set sections and start lines.
"""
self.sections = sections
self.sections_start_lines = start_lines
def set_params(self, names, descs, types, starts):
"""
Set parameter list: names, descriptions, types and start lines.
"""
self.parameterlist = names
self.parameterdescs = descs
self.parametertypes = types
self.parameterdesc_start_lines = starts
linux-7.2.3/tools/lib/python/kdoc/kdoc_output.py
+1160
View File
File diff suppressed because it is too large Load Diff
linux-7.2.3/tools/lib/python/kdoc/kdoc_parser.py
+1731
View File
File diff suppressed because it is too large Load Diff
linux-7.2.3/tools/lib/python/kdoc/kdoc_re.py
+147
View File
@@ -0,0 +1,147 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
# Copyright(c) 2025: Mauro Carvalho Chehab <mchehab@kernel.org>.
"""
Regular expression ancillary classes.
Those help caching regular expressions and do matching for kernel-doc.
"""
import re
# Local cache for regular expressions
re_cache = {}
class KernRe:
"""
Helper class to simplify regex declaration and usage.
It calls re.compile for a given pattern. It also allows adding
regular expressions and define sub at class init time.
Regular expressions can be cached via an argument, helping to speedup
searches.
"""
def _add_regex(self, string, flags):
"""
Adds a new regex or reuses it from the cache.
"""
self.regex = re_cache.get(string, None)
if not self.regex:
self.regex = re.compile(string, flags=flags)
if self.cache:
re_cache[string] = self.regex
def __init__(self, string, cache=True, flags=0):
"""
Compile a regular expression and initialize internal vars.
"""
self.cache = cache
self.last_match = None
self._add_regex(string, flags)
def __str__(self):
"""
Return the regular expression pattern.
"""
return self.regex.pattern
def __repr__(self):
"""
Returns a displayable version of the class init.
"""
flag_map = {
re.IGNORECASE: "re.I",
re.MULTILINE: "re.M",
re.DOTALL: "re.S",
re.VERBOSE: "re.X",
}
flags = []
for flag, name in flag_map.items():
if self.regex.flags & flag:
flags.append(name)
flags_name = " | ".join(flags)
max_len = 60
pattern = ""
for pos in range(0, len(self.regex.pattern), max_len):
pattern += '"' + self.regex.pattern[pos:max_len + pos] + '" '
if flags_name:
return f'KernRe({pattern}, {flags_name})'
else:
return f'KernRe({pattern})'
def __add__(self, other):
"""
Allows adding two regular expressions into one.
"""
return KernRe(str(self) + str(other), cache=self.cache or other.cache,
flags=self.regex.flags | other.regex.flags)
def match(self, string):
"""
Handles a re.match storing its results.
"""
self.last_match = self.regex.match(string)
return self.last_match
def search(self, string):
"""
Handles a re.search storing its results.
"""
self.last_match = self.regex.search(string)
return self.last_match
def finditer(self, string):
"""
Alias to re.finditer.
"""
return self.regex.finditer(string)
def findall(self, string):
"""
Alias to re.findall.
"""
return self.regex.findall(string)
def split(self, string):
"""
Alias to re.split.
"""
return self.regex.split(string)
def sub(self, sub, string, count=0):
"""
Alias to re.sub.
"""
return self.regex.sub(sub, string, count=count)
def group(self, num):
"""
Returns the group results of the last match.
"""
return self.last_match.group(num)
def groups(self):
"""
Returns the group results of the last match
"""
return self.last_match.groups()
linux-7.2.3/tools/lib/python/kdoc/kdoc_yaml_file.py
+178
View File
@@ -0,0 +1,178 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
# Copyright(c) 2026: Mauro Carvalho Chehab <mchehab@kernel.org>.
import os
from kdoc.kdoc_output import ManFormat, RestFormat
class KDocTestFile():
"""
Handles the logic needed to store kernel‑doc output inside a YAML file.
Useful for unit tests and regression tests.
"""
def __init__(self, config, yaml_file, yaml_content):
#
# Bail out early if yaml is not available
#
try:
import yaml
except ImportError:
sys.exit("Warning: yaml package not available. Aborting it.")
self.config = config
self.test_file = os.path.expanduser(yaml_file)
self.yaml_content = yaml_content
self.test_names = set()
self.tests = []
out_dir = os.path.dirname(self.test_file)
if out_dir and not os.path.isdir(out_dir):
sys.exit(f"Directory {out_dir} doesn't exist.")
self.out_style = []
if "man" in self.yaml_content:
out_style = ManFormat()
out_style.set_config(self.config)
self.out_style.append(out_style)
if "rst" in self.yaml_content:
out_style = RestFormat()
out_style.set_config(self.config)
self.out_style.append(out_style)
def set_filter(self, export, internal, symbol, nosymbol,
function_table, enable_lineno, no_doc_sections):
"""
Set filters at the output classes.
"""
for out_style in self.out_style:
out_style.set_filter(export, internal, symbol,
nosymbol, function_table,
enable_lineno, no_doc_sections)
@staticmethod
def get_kdoc_item(arg, start_line=1):
d = vars(arg)
declaration_start_line = d.get("declaration_start_line")
if not declaration_start_line:
return d
d["declaration_start_line"] = start_line
parameterdesc_start_lines = d.get("parameterdesc_start_lines")
if parameterdesc_start_lines:
for key in parameterdesc_start_lines:
ln = parameterdesc_start_lines[key]
ln += start_line - declaration_start_line
parameterdesc_start_lines[key] = ln
sections_start_lines = d.get("sections_start_lines")
if sections_start_lines:
for key in sections_start_lines:
ln = sections_start_lines[key]
ln += start_line - declaration_start_line
sections_start_lines[key] = ln
return d
def output_symbols(self, fname, symbols):
"""
Store source, symbols and output strings at self.tests.
"""
#
# KdocItem needs to be converted into dicts
#
kdoc_item = []
expected = []
#
# Source code didn't produce any symbol
#
if not symbols:
return
expected_dict = {}
start_line=1
for arg in symbols:
source = arg.get("source", "")
if arg and "KdocItem" in self.yaml_content:
msg = self.get_kdoc_item(arg)
other_stuff = msg.get("other_stuff", {})
if "source" in other_stuff:
del other_stuff["source"]
expected_dict["kdoc_item"] = msg
base_name = arg.name
if not base_name:
base_name = fname
base_name = base_name.lower().replace(".", "_").replace("/", "_")
# Don't add duplicated names
i = 0
name = base_name
while name in self.test_names:
i += 1
name = f"{base_name}_{i:03d}"
self.test_names.add(name)
for out_style in self.out_style:
if isinstance(out_style, ManFormat):
key = "man"
else:
key = "rst"
expected_dict[key]= out_style.output_symbols(fname, [arg]).strip()
test = {
"name": name,
"description": f"{fname} line {arg.declaration_start_line}",
"fname": fname,
"source": source,
"expected": [expected_dict]
}
self.tests.append(test)
expected_dict = {}
def write(self):
"""
Output the content of self.tests to self.test_file.
"""
import yaml
# Helper function to better handle multilines
def str_presenter(dumper, data):
if "\n" in data:
return dumper.represent_scalar("tag:yaml.org,2002:str", data, style="|")
return dumper.represent_scalar("tag:yaml.org,2002:str", data)
# Register the representer
yaml.add_representer(str, str_presenter)
data = {"tests": self.tests}
with open(self.test_file, "w", encoding="utf-8") as fp:
yaml.dump(data, fp,
sort_keys=False, width=120, indent=2,
default_flow_style=False, allow_unicode=True,
explicit_start=False, explicit_end=False)
linux-7.2.3/tools/lib/python/kdoc/latex_fonts.py
Executable
+184
View File
@@ -0,0 +1,184 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0-only
# Copyright (C) Akira Yokosawa, 2024
#
# Ported to Python by (c) Mauro Carvalho Chehab, 2025
"""
Detect problematic Noto CJK variable fonts
==========================================
For ``make pdfdocs``, reports of build errors of translations.pdf started
arriving early 2024 [1]_ [2]_. It turned out that Fedora and openSUSE
tumbleweed have started deploying variable-font [3]_ format of "Noto CJK"
fonts [4]_ [5]_. For PDF, a LaTeX package named xeCJK is used for CJK
(Chinese, Japanese, Korean) pages. xeCJK requires XeLaTeX/XeTeX, which
does not (and likely never will) understand variable fonts for historical
reasons.
The build error happens even when both of variable- and non-variable-format
fonts are found on the build system. To make matters worse, Fedora enlists
variable "Noto CJK" fonts in the requirements of langpacks-ja, -ko, -zh_CN,
-zh_TW, etc. Hence developers who have interest in CJK pages are more
likely to encounter the build errors.
This script is invoked from the error path of "make pdfdocs" and emits
suggestions if variable-font files of "Noto CJK" fonts are in the list of
fonts accessible from XeTeX.
.. [1] https://lore.kernel.org/r/8734tqsrt7.fsf@meer.lwn.net/
.. [2] https://lore.kernel.org/r/1708585803.600323099@f111.i.mail.ru/
.. [3] https://en.wikipedia.org/wiki/Variable_font
.. [4] https://fedoraproject.org/wiki/Changes/Noto_CJK_Variable_Fonts
.. [5] https://build.opensuse.org/request/show/1157217
Workarounds for building translations.pdf
-----------------------------------------
* Denylist "variable font" Noto CJK fonts.
- Create $HOME/deny-vf/fontconfig/fonts.conf from template below, with
tweaks if necessary. Remove leading "".
- Path of fontconfig/fonts.conf can be overridden by setting an env
variable FONTS_CONF_DENY_VF.
* Template::
<?xml version="1.0"?>
<!DOCTYPE fontconfig SYSTEM "urn:fontconfig:fonts.dtd">
<fontconfig>
<!--
Ignore variable-font glob (not to break xetex)
-->
<selectfont>
<rejectfont>
<!--
for Fedora
-->
<glob>/usr/share/fonts/google-noto-*-cjk-vf-fonts</glob>
<!--
for openSUSE tumbleweed
-->
<glob>/usr/share/fonts/truetype/Noto*CJK*-VF.otf</glob>
</rejectfont>
</selectfont>
</fontconfig>
The denylisting is activated for "make pdfdocs".
* For skipping CJK pages in PDF
- Uninstall texlive-xecjk.
Denylisting is not needed in this case.
* For printing CJK pages in PDF
- Need non-variable "Noto CJK" fonts.
* Fedora
- google-noto-sans-cjk-fonts
- google-noto-serif-cjk-fonts
* openSUSE tumbleweed
- Non-variable "Noto CJK" fonts are not available as distro packages
as of April, 2024. Fetch a set of font files from upstream Noto
CJK Font released at:
https://github.com/notofonts/noto-cjk/tree/main/Sans#super-otc
and at:
https://github.com/notofonts/noto-cjk/tree/main/Serif#super-otc
then uncompress and deploy them.
- Remember to update fontconfig cache by running fc-cache.
.. caution::
Uninstalling "variable font" packages can be dangerous.
They might be depended upon by other packages important for your work.
Denylisting should be less invasive, as it is effective only while
XeLaTeX runs in "make pdfdocs".
"""
import os
import re
import subprocess
import textwrap
import sys
class LatexFontChecker:
"""
Detect problems with CJK variable fonts that affect PDF builds for
translations.
"""
def __init__(self, deny_vf=None):
if not deny_vf:
deny_vf = os.environ.get('FONTS_CONF_DENY_VF', "~/deny-vf")
self.environ = os.environ.copy()
self.environ['XDG_CONFIG_HOME'] = os.path.expanduser(deny_vf)
self.re_cjk = re.compile(r"([^:]+):\s*Noto\s+(Sans|Sans Mono|Serif) CJK")
def description(self):
"""
Returns module description.
"""
return __doc__
def get_noto_cjk_vf_fonts(self):
"""
Get Noto CJK fonts.
"""
cjk_fonts = set()
cmd = ["fc-list", ":", "file", "family", "variable"]
try:
result = subprocess.run(cmd,stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True,
env=self.environ,
check=True)
except subprocess.CalledProcessError as exc:
sys.exit(f"Error running fc-list: {repr(exc)}")
for line in result.stdout.splitlines():
if 'variable=True' not in line:
continue
match = self.re_cjk.search(line)
if match:
cjk_fonts.add(match.group(1))
return sorted(cjk_fonts)
def check(self):
"""
Check for problems with CJK fonts.
"""
fonts = textwrap.indent("\n".join(self.get_noto_cjk_vf_fonts()), " ")
if not fonts:
return None
rel_file = os.path.relpath(__file__, os.getcwd())
msg = "=" * 77 + "\n"
msg += 'XeTeX is confused by "variable font" files listed below:\n'
msg += fonts + "\n"
msg += textwrap.dedent(f"""
For CJK pages in PDF, they need to be hidden from XeTeX by denylisting.
Or, CJK pages can be skipped by uninstalling texlive-xecjk.
For more info on denylisting, other options, and variable font, run:
tools/docs/check-variable-fonts.py -h
""")
msg += "=" * 77
return msg
linux-7.2.3/tools/lib/python/kdoc/parse_data_structs.py
Executable
+498
View File
@@ -0,0 +1,498 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
# Copyright (c) 2016-2025 by Mauro Carvalho Chehab <mchehab@kernel.org>.
# pylint: disable=R0912,R0915
"""
Parse a source file or header, creating ReStructured Text cross references.
It accepts an optional file to change the default symbol reference or to
suppress symbols from the output.
It is capable of identifying ``define``, function, ``struct``, ``typedef``,
``enum`` and ``enum`` symbols and create cross-references for all of them.
It is also capable of distinguish #define used for specifying a Linux
ioctl.
The optional rules file contains a set of rules like::
ignore ioctl VIDIOC_ENUM_FMT
replace ioctl VIDIOC_DQBUF vidioc_qbuf
replace define V4L2_EVENT_MD_FL_HAVE_FRAME_SEQ :c:type:`v4l2_event_motion_det`
"""
import os
import re
import sys
class ParseDataStructs:
"""
Creates an enriched version of a Kernel header file with cross-links
to each C data structure type.
It is meant to allow having a more comprehensive documentation, where
uAPI headers will create cross-reference links to the code.
It is capable of identifying ``define``, function, ``struct``, ``typedef``,
``enum`` and ``enum`` symbols and create cross-references for all of them.
It is also capable of distinguish #define used for specifying a Linux
ioctl.
By default, it create rules for all symbols and defines, but it also
allows parsing an exception file. Such file contains a set of rules
using the syntax below:
1. Ignore rules::
ignore <type> <symbol>`
Removes the symbol from reference generation.
2. Replace rules::
replace <type> <old_symbol> <new_reference>
Replaces how old_symbol with a new reference. The new_reference can be:
- A simple symbol name;
- A full Sphinx reference.
3. Namespace rules::
namespace <namespace>
Sets C namespace to be used during cross-reference generation. Can
be overridden by replace rules.
On ignore and replace rules, ``<type>`` can be:
- ``ioctl``: for defines that end with ``_IO*``, e.g. ioctl definitions
- ``define``: for other defines
- ``symbol``: for symbols defined within enums;
- ``typedef``: for typedefs;
- ``enum``: for the name of a non-anonymous enum;
- ``struct``: for structs.
Examples::
ignore define __LINUX_MEDIA_H
ignore ioctl VIDIOC_ENUM_FMT
replace ioctl VIDIOC_DQBUF vidioc_qbuf
replace define V4L2_EVENT_MD_FL_HAVE_FRAME_SEQ :c:type:`v4l2_event_motion_det`
namespace MC
"""
#: Parser regex with multiple ways to capture enums.
RE_ENUMS = [
re.compile(r"^\s*enum\s+([\w_]+)\s*\{"),
re.compile(r"^\s*enum\s+([\w_]+)\s*$"),
re.compile(r"^\s*typedef\s*enum\s+([\w_]+)\s*\{"),
re.compile(r"^\s*typedef\s*enum\s+([\w_]+)\s*$"),
]
#: Parser regex with multiple ways to capture structs.
RE_STRUCTS = [
re.compile(r"^\s*struct\s+([_\w][\w\d_]+)\s*\{"),
re.compile(r"^\s*struct\s+([_\w][\w\d_]+)$"),
re.compile(r"^\s*typedef\s*struct\s+([_\w][\w\d_]+)\s*\{"),
re.compile(r"^\s*typedef\s*struct\s+([_\w][\w\d_]+)$"),
]
# NOTE: the original code was written a long time before Sphinx C
# domain to have multiple namespaces. To avoid to much turn at the
# existing hyperlinks, the code kept using "c:type" instead of the
# right types. To change that, we need to change the types not only
# here, but also at the uAPI media documentation.
#: Dictionary containing C type identifiers to be transformed.
DEF_SYMBOL_TYPES = {
"ioctl": {
"prefix": "\\ ",
"suffix": "\\ ",
"ref_type": ":ref",
"description": "IOCTL Commands",
},
"define": {
"prefix": "\\ ",
"suffix": "\\ ",
"ref_type": ":ref",
"description": "Macros and Definitions",
},
# We're calling each definition inside an enum as "symbol"
"symbol": {
"prefix": "\\ ",
"suffix": "\\ ",
"ref_type": ":ref",
"description": "Enumeration values",
},
"typedef": {
"prefix": "\\ ",
"suffix": "\\ ",
"ref_type": ":c:type",
"description": "Type Definitions",
},
# This is the description of the enum itself
"enum": {
"prefix": "\\ ",
"suffix": "\\ ",
"ref_type": ":c:type",
"description": "Enumerations",
},
"struct": {
"prefix": "\\ ",
"suffix": "\\ ",
"ref_type": ":c:type",
"description": "Structures",
},
}
def __init__(self, debug: bool = False):
"""Initialize internal vars"""
self.debug = debug
self.data = ""
self.symbols = {}
self.namespace = None
self.ignore = []
self.replace = []
for symbol_type in self.DEF_SYMBOL_TYPES:
self.symbols[symbol_type] = {}
def read_exceptions(self, fname: str):
"""
Read an optional exceptions file, used to override defaults.
"""
if not fname:
return
name = os.path.basename(fname)
with open(fname, "r", encoding="utf-8", errors="backslashreplace") as f:
for ln, line in enumerate(f):
ln += 1
line = line.strip()
if not line or line.startswith("#"):
continue
# ignore rules
match = re.match(r"^ignore\s+(\w+)\s+(\S+)", line)
if match:
self.ignore.append((ln, match.group(1), match.group(2)))
continue
# replace rules
match = re.match(r"^replace\s+(\S+)\s+(\S+)\s+(\S+)", line)
if match:
self.replace.append((ln, match.group(1), match.group(2),
match.group(3)))
continue
match = re.match(r"^namespace\s+(\S+)", line)
if match:
self.namespace = match.group(1)
continue
sys.exit(f"{name}:{ln}: invalid line: {line}")
def apply_exceptions(self):
"""
Process exceptions file with rules to ignore or replace references.
"""
# Handle ignore rules
for ln, c_type, symbol in self.ignore:
if c_type not in self.DEF_SYMBOL_TYPES:
sys.exit(f"{name}:{ln}: {c_type} is invalid")
d = self.symbols[c_type]
if symbol in d:
del d[symbol]
# Handle replace rules
for ln, c_type, old, new in self.replace:
if c_type not in self.DEF_SYMBOL_TYPES:
sys.exit(f"{name}:{ln}: {c_type} is invalid")
reftype = None
# Parse reference type when the type is specified
match = re.match(r"^\:c\:(\w+)\:\`(.+)\`", new)
if match:
reftype = f":c:{match.group(1)}"
new = match.group(2)
else:
match = re.search(r"(\:ref)\:\`(.+)\`", new)
if match:
reftype = match.group(1)
new = match.group(2)
# If the replacement rule doesn't have a type, get default
if not reftype:
reftype = self.DEF_SYMBOL_TYPES[c_type].get("ref_type")
if not reftype:
reftype = self.DEF_SYMBOL_TYPES[c_type].get("real_type")
new_ref = f"{reftype}:`{old} <{new}>`"
# Change self.symbols to use the replacement rule
if old in self.symbols[c_type]:
(_, ln) = self.symbols[c_type][old]
self.symbols[c_type][old] = (new_ref, ln)
else:
print(f"{name}:{ln}: Warning: can't find {old} {c_type}")
def store_type(self, ln, symbol_type: str, symbol: str,
ref_name: str = None, replace_underscores: bool = True):
"""
Store a new symbol at self.symbols under symbol_type.
By default, underscores are replaced by ``-``.
"""
defs = self.DEF_SYMBOL_TYPES[symbol_type]
prefix = defs.get("prefix", "")
suffix = defs.get("suffix", "")
ref_type = defs.get("ref_type")
# Determine ref_link based on symbol type
if ref_type or self.namespace:
if not ref_name:
ref_name = symbol.lower()
# c-type references don't support hash
if ref_type == ":ref" and replace_underscores:
ref_name = ref_name.replace("_", "-")
# C domain references may have namespaces
if ref_type.startswith(":c:"):
if self.namespace:
ref_name = f"{self.namespace}.{ref_name}"
if ref_type:
ref_link = f"{ref_type}:`{symbol} <{ref_name}>`"
else:
ref_link = f"`{symbol} <{ref_name}>`"
else:
ref_link = symbol
self.symbols[symbol_type][symbol] = (f"{prefix}{ref_link}{suffix}", ln)
def store_line(self, line):
"""
Store a line at self.data, properly indented.
"""
line = " " + line.expandtabs()
self.data += line.rstrip(" ")
def parse_file(self, file_in: str, exceptions: str = None):
"""
Read a C source file and get identifiers.
"""
self.data = ""
is_enum = False
is_comment = False
multiline = ""
self.read_exceptions(exceptions)
with open(file_in, "r",
encoding="utf-8", errors="backslashreplace") as f:
for line_no, line in enumerate(f):
self.store_line(line)
line = line.strip("\n")
# Handle continuation lines
if line.endswith(r"\\"):
multiline += line[-1]
continue
if multiline:
line = multiline + line
multiline = ""
# Handle comments. They can be multilined
if not is_comment:
if re.search(r"/\*.*", line):
is_comment = True
else:
# Strip C99-style comments
line = re.sub(r"(//.*)", "", line)
if is_comment:
if re.search(r".*\*/", line):
is_comment = False
else:
multiline = line
continue
# At this point, line variable may be a multilined statement,
# if lines end with \ or if they have multi-line comments
# With that, it can safely remove the entire comments,
# and there's no need to use re.DOTALL for the logic below
line = re.sub(r"(/\*.*\*/)", "", line)
if not line.strip():
continue
# It can be useful for debug purposes to print the file after
# having comments stripped and multi-lines grouped.
if self.debug > 1:
print(f"line {line_no + 1}: {line}")
# Now the fun begins: parse each type and store it.
# We opted for a two parsing logic here due to:
# 1. it makes easier to debug issues not-parsed symbols;
# 2. we want symbol replacement at the entire content, not
# just when the symbol is detected.
if is_enum:
match = re.match(r"^\s*([_\w][\w\d_]+)\s*[\,=]?", line)
if match:
self.store_type(line_no, "symbol", match.group(1))
if "}" in line:
is_enum = False
continue
match = re.match(r"^\s*#\s*define\s+([\w_]+)\s+_IO", line)
if match:
self.store_type(line_no, "ioctl", match.group(1),
replace_underscores=False)
continue
match = re.match(r"^\s*#\s*define\s+([\w_]+)(\s+|$)", line)
if match:
self.store_type(line_no, "define", match.group(1))
continue
match = re.match(r"^\s*typedef\s+([_\w][\w\d_]+)\s+(.*)\s+([_\w][\w\d_]+);",
line)
if match:
name = match.group(2).strip()
symbol = match.group(3)
self.store_type(line_no, "typedef", symbol, ref_name=name)
continue
for re_enum in self.RE_ENUMS:
match = re_enum.match(line)
if match:
self.store_type(line_no, "enum", match.group(1))
is_enum = True
break
for re_struct in self.RE_STRUCTS:
match = re_struct.match(line)
if match:
self.store_type(line_no, "struct", match.group(1))
break
self.apply_exceptions()
def debug_print(self):
"""
Print debug information containing the replacement rules per symbol.
To make easier to check, group them per type.
"""
if not self.debug:
return
for c_type, refs in self.symbols.items():
if not refs: # Skip empty dictionaries
continue
print(f"{c_type}:")
for symbol, (ref, ln) in sorted(refs.items()):
print(f" #{ln:<5d} {symbol} -> {ref}")
print()
def gen_output(self):
"""Write the formatted output to a file."""
# Avoid extra blank lines
text = re.sub(r"\s+$", "", self.data) + "\n"
text = re.sub(r"\n\s+\n", "\n\n", text)
# Escape Sphinx special characters
text = re.sub(r"([\_\`\*\<\>\&\\\\:\/\|\%\$\#\{\}\~\^])", r"\\\1", text)
# Source uAPI files may have special notes. Use bold font for them
text = re.sub(r"DEPRECATED", "**DEPRECATED**", text)
# Delimiters to catch the entire symbol after escaped
start_delim = r"([ \n\t\(=\*\@])"
end_delim = r"(\s|,|\\=|\\:|\;|\)|\}|\{)"
# Process all reference types
for ref_dict in self.symbols.values():
for symbol, (replacement, _) in ref_dict.items():
symbol = re.escape(re.sub(r"([\_\`\*\<\>\&\\\\:\/])", r"\\\1", symbol))
text = re.sub(fr'{start_delim}{symbol}{end_delim}',
fr'\1{replacement}\2', text)
# Remove "\ " where not needed: before spaces and at the end of lines
text = re.sub(r"\\ ([\n ])", r"\1", text)
text = re.sub(r" \\ ", " ", text)
return text
def gen_toc(self):
"""
Create a list of symbols to be part of a TOC contents table.
"""
text = []
# Sort symbol types per description
symbol_descriptions = []
for k, v in self.DEF_SYMBOL_TYPES.items():
symbol_descriptions.append((v['description'], k))
symbol_descriptions.sort()
# Process each category
for description, c_type in symbol_descriptions:
refs = self.symbols[c_type]
if not refs: # Skip empty categories
continue
text.append(f"{description}")
text.append("-" * len(description))
text.append("")
# Sort symbols alphabetically
for symbol, (ref, ln) in sorted(refs.items()):
text.append(f"- LINENO_{ln}: {ref}")
text.append("") # Add empty line between categories
return "\n".join(text)
def write_output(self, file_in: str, file_out: str, toc: bool):
"""
Write a ReST output file.
"""
title = os.path.basename(file_in)
if toc:
text = self.gen_toc()
else:
text = self.gen_output()
with open(file_out, "w", encoding="utf-8", errors="backslashreplace") as f:
f.write(".. -*- coding: utf-8; mode: rst -*-\n\n")
f.write(f"{title}\n")
f.write("=" * len(title) + "\n\n")
if not toc:
f.write(".. parsed-literal::\n\n")
f.write(text)
linux-7.2.3/tools/lib/python/kdoc/python_version.py
+190
View File
@@ -0,0 +1,190 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0-or-later
# Copyright (c) 2017-2025 Mauro Carvalho Chehab <mchehab+huawei@kernel.org>
"""
Handle Python version check logic.
Not all Python versions are supported by scripts. Yet, on some cases,
like during documentation build, a newer version of python could be
available.
This class allows checking if the minimal requirements are followed.
Better than that, PythonVersion.check_python() not only checks the minimal
requirements, but it automatically switches to a the newest available
Python version if present.
"""
import os
import re
import subprocess
import shlex
import sys
from glob import glob
from textwrap import indent
class PythonVersion:
"""
Ancillary methods that checks for missing dependencies for different
types of types, like binaries, python modules, rpm deps, etc.
"""
def __init__(self, version):
"""
Ïnitialize self.version tuple from a version string.
"""
self.version = self.parse_version(version)
@staticmethod
def parse_version(version):
"""
Convert a major.minor.patch version into a tuple.
"""
return tuple(int(x) for x in version.split("."))
@staticmethod
def ver_str(version):
"""
Returns a version tuple as major.minor.patch.
"""
return ".".join([str(x) for x in version])
@staticmethod
def cmd_print(cmd, max_len=80):
"""
Outputs a command line, repecting maximum width.
"""
cmd_line = []
for w in cmd:
w = shlex.quote(w)
if cmd_line:
if not max_len or len(cmd_line[-1]) + len(w) < max_len:
cmd_line[-1] += " " + w
continue
else:
cmd_line[-1] += " \\"
cmd_line.append(w)
else:
cmd_line.append(w)
return "\n ".join(cmd_line)
def __str__(self):
"""
Return a version tuple as major.minor.patch from self.version.
"""
return self.ver_str(self.version)
@staticmethod
def get_python_version(cmd):
"""
Get python version from a Python binary. As we need to detect if
are out there newer python binaries, we can't rely on sys.release here.
"""
kwargs = {}
if sys.version_info < (3, 7):
kwargs['universal_newlines'] = True
else:
kwargs['text'] = True
result = subprocess.run([cmd, "--version"],
stdout = subprocess.PIPE,
stderr = subprocess.PIPE,
**kwargs, check=False)
version = result.stdout.strip()
match = re.search(r"(\d+\.\d+\.\d+)", version)
if match:
return PythonVersion.parse_version(match.group(1))
print(f"Can't parse version {version}")
return (0, 0, 0)
@staticmethod
def find_python(min_version):
"""
Detect if are out there any python 3.xy version newer than the
current one.
Note: this routine is limited to up to 2 digits for python3. We
may need to update it one day, hopefully on a distant future.
"""
patterns = [
"python3.[0-9][0-9]",
"python3.[0-9]",
]
python_cmd = []
# Seek for a python binary newer than min_version
for path in os.getenv("PATH", "").split(":"):
for pattern in patterns:
for cmd in glob(os.path.join(path, pattern)):
if os.path.isfile(cmd) and os.access(cmd, os.X_OK):
version = PythonVersion.get_python_version(cmd)
if version >= min_version:
python_cmd.append((version, cmd))
return sorted(python_cmd, reverse=True)
@staticmethod
def check_python(min_version, show_alternatives=False, bail_out=False,
success_on_error=False):
"""
Check if the current python binary satisfies our minimal requirement
for Sphinx build. If not, re-run with a newer version if found.
"""
cur_ver = sys.version_info[:3]
if cur_ver >= min_version:
ver = PythonVersion.ver_str(cur_ver)
return
python_ver = PythonVersion.ver_str(cur_ver)
available_versions = PythonVersion.find_python(min_version)
if not available_versions:
print(f"ERROR: Python version {python_ver} is not supported anymore\n")
print(" Can't find a new version. This script may fail")
return
script_path = os.path.abspath(sys.argv[0])
# Check possible alternatives
if available_versions:
new_python_cmd = available_versions[0][1]
else:
new_python_cmd = None
if show_alternatives and available_versions:
print("You could run, instead:")
for _, cmd in available_versions:
args = [cmd, script_path] + sys.argv[1:]
cmd_str = indent(PythonVersion.cmd_print(args), " ")
print(f"{cmd_str}\n")
if bail_out:
msg = f"Python {python_ver} not supported. Bailing out"
if success_on_error:
print(msg, file=sys.stderr)
sys.exit(0)
else:
sys.exit(msg)
print(f"Python {python_ver} not supported. Changing to {new_python_cmd}")
# Restart script using the newer version
args = [new_python_cmd, script_path] + sys.argv[1:]
try:
os.execv(new_python_cmd, args)
except OSError as e:
sys.exit(f"Failed to restart with {new_python_cmd}: {e}")
linux-7.2.3/tools/lib/python/kdoc/xforms_lists.py
+159
View File
@@ -0,0 +1,159 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
# Copyright(c) 2026: Mauro Carvalho Chehab <mchehab@kernel.org>.
import re
from kdoc.kdoc_re import KernRe
from kdoc.c_lex import CMatch, CTokenizer
struct_args_pattern = r"([^,)]+)"
class CTransforms:
"""
Data class containing a long set of transformations to turn
structure member prefixes, and macro invocations and variables
into something we can parse and generate kdoc for.
"""
#
# NOTE:
# Due to performance reasons, place CMatch rules before KernRe,
# as this avoids running the C parser every time.
#
#: Transforms for structs and unions.
struct_xforms = [
(CMatch("__attribute__"), ""),
(CMatch("__aligned"), ""),
(CMatch("__counted_by"), ""),
(CMatch("__counted_by_(le|be)"), ""),
(CMatch("__counted_by_ptr"), ""),
(CMatch("__guarded_by"), ""),
(CMatch("__pt_guarded_by"), ""),
(CMatch("__packed"), ""),
(CMatch("CRYPTO_MINALIGN_ATTR"), ""),
(CMatch("__private"), ""),
(CMatch("__rcu"), ""),
(CMatch("____cacheline_aligned_in_smp"), ""),
(CMatch("____cacheline_aligned"), ""),
(CMatch("__cacheline_group_(?:begin|end)"), ""),
(CMatch("__ETHTOOL_DECLARE_LINK_MODE_MASK"), r"DECLARE_BITMAP(\1, __ETHTOOL_LINK_MODE_MASK_NBITS)"),
(CMatch("DECLARE_PHY_INTERFACE_MASK",),r"DECLARE_BITMAP(\1, PHY_INTERFACE_MODE_MAX)"),
(CMatch("DECLARE_BITMAP"), r"unsigned long \1[BITS_TO_LONGS(\2)]"),
(CMatch("DECLARE_HASHTABLE"), r"unsigned long \1[1 << ((\2) - 1)]"),
(CMatch("DECLARE_KFIFO"), r"\2 *\1"),
(CMatch("DECLARE_KFIFO_PTR"), r"\2 *\1"),
(CMatch("(?:__)?DECLARE_FLEX_ARRAY"), r"\1 \2[]"),
(CMatch("DEFINE_DMA_UNMAP_ADDR"), r"dma_addr_t \1"),
(CMatch("DEFINE_DMA_UNMAP_LEN"), r"__u32 \1"),
(CMatch("VIRTIO_DECLARE_FEATURES"), r"union { u64 \1; u64 \1_array[VIRTIO_FEATURES_U64S]; }"),
(CMatch("__SYSFS_FUNCTION_ALTERNATIVE"), r"union { \1+ }"),
(CMatch("__attribute__"), ""),
#
# Macro __struct_group() creates an union with an anonymous
# and a non-anonymous struct, depending on the parameters. We only
# need one of those at kernel-doc, as we won't be documenting the same
# members twice.
#
(CMatch("struct_group"), r"struct { \2+ };"),
(CMatch("struct_group_attr"), r"struct { \3+ };"),
(CMatch("struct_group_tagged"), r"struct { \3+ };"),
(CMatch("__struct_group"), r"struct { \4+ };"),
]
#: Transforms for function prototypes.
function_xforms = [
(CMatch("static"), ""),
(CMatch("extern"), ""),
(CMatch("asmlinkage"), ""),
(CMatch("inline"), ""),
(CMatch("__inline__"), ""),
(CMatch("__inline"), ""),
(CMatch("__always_inline"), ""),
(CMatch("noinline"), ""),
(CMatch("__FORTIFY_INLINE"), ""),
(CMatch("__init"), ""),
(CMatch("__init_or_module"), ""),
(CMatch("__exit"), ""),
(CMatch("__deprecated"), ""),
(CMatch("__flatten"), ""),
(CMatch("__meminit"), ""),
(CMatch("__must_check"), ""),
(CMatch("__weak"), ""),
(CMatch("__sched"), ""),
(CMatch("__always_unused"), ""),
(CMatch("__maybe_unused"), ""),
(CMatch("__printf"), ""),
(CMatch("__(?:re)?alloc_size"), ""),
(CMatch("__diagnose_as"), ""),
(CMatch("DECL_BUCKET_PARAMS"), r"\1, \2"),
(CMatch("__cond_acquires"), ""),
(CMatch("__cond_releases"), ""),
(CMatch("__acquires"), ""),
(CMatch("__releases"), ""),
(CMatch("__must_hold"), ""),
(CMatch("__must_not_hold"), ""),
(CMatch("__must_hold_shared"), ""),
(CMatch("__cond_acquires_shared"), ""),
(CMatch("__acquires_shared"), ""),
(CMatch("__releases_shared"), ""),
(CMatch("__no_context_analysis"), ""),
(CMatch("__attribute_const__"), ""),
(CMatch("__attribute__"), ""),
(CMatch("STATIC_IFN_KUNIT"), ""),
(CMatch("INLINE_IFN_KUNIT"), ""),
#
# HACK: this is similar to process_export() hack. It is meant to
# drop _noproof from function name. See for instance:
# ahash_request_alloc kernel-doc declaration at include/crypto/hash.h.
#
(KernRe("_noprof"), ""),
]
#: Transforms for variable prototypes.
var_xforms = [
(CMatch("__read_mostly"), ""),
(CMatch("__ro_after_init"), ""),
(CMatch("__guarded_by"), ""),
(CMatch("__pt_guarded_by"), ""),
(CMatch("LIST_HEAD"), r"struct list_head \1"),
(CMatch("DECLARE_PER_CPU"), r"\1 \2[PER_CPU]; }"),
(KernRe(r"(?://.*)$"), ""),
(KernRe(r"(?:/\*.*\*/)"), ""),
(KernRe(r";$"), ""),
]
#: Transforms main dictionary used at apply_transforms().
xforms = {
"struct": struct_xforms,
"func": function_xforms,
"var": var_xforms,
}
def apply(self, xforms_type, source):
"""
Apply a set of transforms to a block of source.
As tokenizer is used here, this function also remove comments
at the end.
"""
if xforms_type not in self.xforms:
return source
if isinstance(source, str):
source = CTokenizer(source)
for search, subst in self.xforms[xforms_type]:
#
# KernRe only accept strings.
#
if isinstance(search, KernRe):
source = str(source)
source = search.sub(subst, source)
return str(source)
linux-7.2.3/tools/lib/python/unittest_helper.py
Executable
+363
View File
@@ -0,0 +1,363 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
# Copyright(c) 2025-2026: Mauro Carvalho Chehab <mchehab@kernel.org>.
#
# pylint: disable=C0103,R0912,R0914,E1101
"""
Provides helper functions and classes execute python unit tests.
Those help functions provide a nice colored output summary of each
executed test and, when a test fails, it shows the different in diff
format when running in verbose mode, like::
$ tools/unittests/nested_match.py -v
...
Traceback (most recent call last):
File "/new_devel/docs/tools/unittests/nested_match.py", line 69, in test_count_limit
self.assertEqual(replaced, "bar(a); bar(b); foo(c)")
~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AssertionError: 'bar(a) foo(b); foo(c)' != 'bar(a); bar(b); foo(c)'
- bar(a) foo(b); foo(c)
? ^^^^
+ bar(a); bar(b); foo(c)
? ^^^^^
...
It also allows filtering what tests will be executed via ``-k`` parameter.
Typical usage is to do::
from unittest_helper import run_unittest
...
if __name__ == "__main__":
run_unittest(__file__)
If passing arguments is needed, on a more complex scenario, it can be
used like on this example::
from unittest_helper import TestUnits, run_unittest
...
env = {'sudo': ""}
...
if __name__ == "__main__":
runner = TestUnits()
base_parser = runner.parse_args()
base_parser.add_argument('--sudo', action='store_true',
help='Enable tests requiring sudo privileges')
args = base_parser.parse_args()
# Update module-level flag
if args.sudo:
env['sudo'] = "1"
# Run tests with customized arguments
runner.run(__file__, parser=base_parser, args=args, env=env)
"""
import argparse
import atexit
import os
import re
import unittest
import sys
from unittest.mock import patch
class Summary(unittest.TestResult):
"""
Overrides ``unittest.TestResult`` class to provide a nice colored
summary. When in verbose mode, displays actual/expected difference in
unified diff format.
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
#: Dictionary to store organized test results.
self.test_results = {}
#: max length of the test names.
self.max_name_length = 0
def startTest(self, test):
super().startTest(test)
test_id = test.id()
parts = test_id.split(".")
# Extract module, class, and method names
if len(parts) >= 3:
module_name = parts[-3]
else:
module_name = ""
if len(parts) >= 2:
class_name = parts[-2]
else:
class_name = ""
method_name = parts[-1]
# Build the hierarchical structure
if module_name not in self.test_results:
self.test_results[module_name] = {}
if class_name not in self.test_results[module_name]:
self.test_results[module_name][class_name] = []
# Track maximum test name length for alignment
display_name = f"{method_name}:"
self.max_name_length = max(len(display_name), self.max_name_length)
def _record_test(self, test, status):
test_id = test.id()
parts = test_id.split(".")
if len(parts) >= 3:
module_name = parts[-3]
else:
module_name = ""
if len(parts) >= 2:
class_name = parts[-2]
else:
class_name = ""
method_name = parts[-1]
self.test_results[module_name][class_name].append((method_name, status))
def addSuccess(self, test):
super().addSuccess(test)
self._record_test(test, "OK")
def addFailure(self, test, err):
super().addFailure(test, err)
self._record_test(test, "FAIL")
def addError(self, test, err):
super().addError(test, err)
self._record_test(test, "ERROR")
def addSkip(self, test, reason):
super().addSkip(test, reason)
self._record_test(test, f"SKIP ({reason})")
def printResults(self, verbose):
"""
Print results using colors if tty.
"""
# Check for ANSI color support
use_color = sys.stdout.isatty()
COLORS = {
"OK": "\033[32m", # Green
"FAIL": "\033[31m", # Red
"SKIP": "\033[1;33m", # Yellow
"PARTIAL": "\033[33m", # Orange
"EXPECTED_FAIL": "\033[36m", # Cyan
"reset": "\033[0m", # Reset to default terminal color
}
if not use_color:
for c in COLORS:
COLORS[c] = ""
# Calculate maximum test name length
if not self.test_results:
return
try:
lengths = []
for module in self.test_results.values():
for tests in module.values():
for test_name, _ in tests:
lengths.append(len(test_name) + 1) # +1 for colon
max_length = max(lengths) + 2 # Additional padding
except ValueError:
sys.exit("Test list is empty")
# Print results
for module_name, classes in self.test_results.items():
if verbose:
print(f"{module_name}:")
for class_name, tests in classes.items():
if verbose:
print(f" {class_name}:")
for test_name, status in tests:
if not verbose and status in [ "OK", "EXPECTED_FAIL" ]:
continue
# Get base status without reason for SKIP
if status.startswith("SKIP"):
status_code = status.split()[0]
else:
status_code = status
color = COLORS.get(status_code, "")
print(
f" {test_name + ':':<{max_length}}{color}{status}{COLORS['reset']}"
)
if verbose:
print()
# Print summary
print(f"\nRan {self.testsRun} tests", end="")
if hasattr(self, "timeTaken"):
print(f" in {self.timeTaken:.3f}s", end="")
print()
if not self.wasSuccessful():
print(f"\n{COLORS['FAIL']}FAILED (", end="")
failures = getattr(self, "failures", [])
errors = getattr(self, "errors", [])
if failures:
print(f"failures={len(failures)}", end="")
if errors:
if failures:
print(", ", end="")
print(f"errors={len(errors)}", end="")
print(f"){COLORS['reset']}")
def flatten_suite(suite):
"""Flatten test suite hierarchy."""
tests = []
for item in suite:
if isinstance(item, unittest.TestSuite):
tests.extend(flatten_suite(item))
else:
tests.append(item)
return tests
class TestUnits:
"""
Helper class to set verbosity level.
This class discover test files, import its unittest classes and
executes the test on it.
"""
def parse_args(self):
"""Returns a parser for command line arguments."""
parser = argparse.ArgumentParser(description="Test runner with regex filtering")
parser.add_argument("-v", "--verbose", action="count", default=1)
parser.add_argument("-q", "--quiet", action="store_true")
parser.add_argument("-f", "--failfast", action="store_true")
parser.add_argument("-k", "--keyword",
help="Regex pattern to filter test methods")
return parser
def run(self, caller_file=None, pattern=None,
suite=None, parser=None, args=None, env=None):
"""
Execute all tests from the unity test file.
It contains several optional parameters:
``caller_file``:
- name of the file that contains test.
typical usage is to place __file__ at the caller test, e.g.::
if __name__ == "__main__":
TestUnits().run(__file__)
``pattern``:
- optional pattern to match multiple file names. Defaults
to basename of ``caller_file``.
``suite``:
- an unittest suite initialized by the caller using
``unittest.TestLoader().discover()``.
``parser``:
- an argparse parser. If not defined, this helper will create
one.
``args``:
- an ``argparse.Namespace`` data filled by the caller.
``env``:
- environment variables that will be passed to the test suite
At least ``caller_file`` or ``suite`` must be used, otherwise a
``TypeError`` will be raised.
"""
if not args:
if not parser:
parser = self.parse_args()
args = parser.parse_args()
if not caller_file and not suite:
raise TypeError("Either caller_file or suite is needed at TestUnits")
if args.quiet:
verbose = 0
else:
verbose = args.verbose
if not env:
env = os.environ.copy()
env["VERBOSE"] = f"{verbose}"
patcher = patch.dict(os.environ, env)
patcher.start()
# ensure it gets stopped after
atexit.register(patcher.stop)
if verbose >= 2:
unittest.TextTestRunner(verbosity=verbose).run = lambda suite: suite
# Load ONLY tests from the calling file
if not suite:
if not pattern:
pattern = caller_file
loader = unittest.TestLoader()
suite = loader.discover(start_dir=os.path.dirname(caller_file),
pattern=os.path.basename(caller_file))
# Flatten the suite for environment injection
tests_to_inject = flatten_suite(suite)
# Filter tests by method name if -k specified
if args.keyword:
try:
pattern = re.compile(args.keyword)
filtered_suite = unittest.TestSuite()
for test in tests_to_inject: # Use the pre-flattened list
method_name = test.id().split(".")[-1]
if pattern.search(method_name):
filtered_suite.addTest(test)
suite = filtered_suite
except re.error as e:
sys.stderr.write(f"Invalid regex pattern: {e}\n")
sys.exit(1)
else:
# Maintain original suite structure if no keyword filtering
suite = unittest.TestSuite(tests_to_inject)
if verbose >= 2:
resultclass = None
else:
resultclass = Summary
runner = unittest.TextTestRunner(verbosity=args.verbose,
resultclass=resultclass,
failfast=args.failfast)
result = runner.run(suite)
if resultclass:
result.printResults(verbose)
sys.exit(not result.wasSuccessful())
def run_unittest(fname):
"""
Basic usage of TestUnits class.
Use it when there's no need to pass any extra argument to the tests
with. The recommended way is to place this at the end of each
unittest module::
if __name__ == "__main__":
run_unittest(__file__)
"""
TestUnits().run(fname)