[None][test] CBTS coverage_utils: pre-commit lint fixes

Squashed cleanup pass to satisfy the repo's pre-commit hooks:

* coverage_audit.py: add `r` prefix to module docstring (D301), split
  summary/description on two helper docstrings (D205), fix spelling
  flagged by codespell.
* stability_diff.py: split docstring summary/description (D205),
  replace `print(); return` semicolon-joined pair (E702).
* sitecustomize.py + stability_diff.py: ruff-format auto-fixes
  (single-line f-strings, import grouping, trailing commas).
* cbts_plugin.py: minor reformatting from ruff-format.

Pure formatting / docstring cleanup -- no behavior change.

Signed-off-by: Ivy Zhang <25222398+crazydemo@users.noreply.github.com>

Author: crazydemo

jenkins/scripts/cbts/coverage_utils/cbts_plugin.py

@@ -40,6 +40,7 @@
 containing this module on ``PYTHONPATH`` (typically same dir as
 ``sitecustomize.py``).
 """
+
 import hashlib
 import inspect
 import os
@@ -72,8 +73,9 @@ def install_mpi_pool_patch(*, raise_on_refactor=True):
         the resulting coverage gap.
     """
     try:
-        import tensorrt_llm.llmapi.mpi_session as _ms
         from mpi4py.futures import MPIPoolExecutor  # noqa: F401
+
+        import tensorrt_llm.llmapi.mpi_session as _ms
     except ImportError:
         return False
 
@@ -96,16 +98,15 @@ def install_mpi_pool_patch(*, raise_on_refactor=True):
     def _patched_start_mpi_pool(self):
         """Widened env whitelist so COVERAGE_* and PYTHON* reach workers."""
         import sys as _sys
+
         from mpi4py.futures import MPIPoolExecutor as _MPE
 
         assert not self.mpi_pool, "MPI session already started"
-        env = {
-            k: v
-            for k, v in os.environ.items()
-            if k.startswith(_ENV_WHITELIST_PREFIXES)
-        }
+        env = {k: v for k, v in os.environ.items() if k.startswith(_ENV_WHITELIST_PREFIXES)}
         self.mpi_pool = _MPE(
-            max_workers=self.n_workers, path=_sys.path, env=env,
+            max_workers=self.n_workers,
+            path=_sys.path,
+            env=env,
         )
 
     setattr(_patched_start_mpi_pool, _PATCHED_MARKER, True)
@@ -152,9 +153,7 @@ def pytest_runtest_protocol(item, nextitem):  # noqa: D401 - pytest hook
     # vs "plugin ran but data was lost". Filename = sha1(nodeid) to dodge
     # any path-escape issues from "::" / "/" in nodeids.
     os.makedirs(SENTINEL_DIR, exist_ok=True)
-    sentinel = os.path.join(
-        SENTINEL_DIR, hashlib.sha1(nodeid.encode()).hexdigest()
-    )
+    sentinel = os.path.join(SENTINEL_DIR, hashlib.sha1(nodeid.encode()).hexdigest())
     Path(sentinel).touch()
 
     cov = coverage.Coverage.current()

jenkins/scripts/cbts/coverage_utils/coverage_audit.py

@@ -11,7 +11,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-"""Post-run audit for CBTS Layer C per-test coverage data quality.
+r"""Post-run audit for CBTS Layer C per-test coverage data quality.
 
 Non-blocking report mode (Stage 1 of the rollout): emits a per-test CSV/JSON
 report plus a summary, and only exits non-zero on *hard* pipeline-plumbing
@@ -85,6 +85,7 @@
         --out-csv   ${JOB_WORKSPACE}/cbts_audit.csv \\
         --out-json  ${JOB_WORKSPACE}/cbts_audit.json
 """
+
 import argparse
 import ast
 import csv
@@ -139,28 +140,31 @@ def _is_integration_test(test_id: str) -> bool:
 @dataclass
 class TestRecord:
     test_id: str
-    junit_status: str           # passed / failed / error / skipped / NO_JUNIT_MATCH / unknown
-    cov_status: str             # OK / MAIN_ONLY / EMPTY / MISSING
-    n_files: int                # total product files (source filter already restricts to tensorrt_llm/)
-    n_engine_files: int         # subset under _torch/pyexecutor or _torch/models -- the worker-side signal
-    n_engine_body_lines: int    # lines in engine files MINUS def/class/decorator/import lines
-                                # (shell-style hits get covered just by importing; this metric
-                                # tells us whether engine function bodies were actually executed)
-    context_key: str            # matched coverage context (may differ from test_id)
+    junit_status: str  # passed / failed / error / skipped / NO_JUNIT_MATCH / unknown
+    cov_status: str  # OK / MAIN_ONLY / EMPTY / MISSING
+    n_files: int  # total product files (source filter already restricts to tensorrt_llm/)
+    n_engine_files: int  # subset under _torch/pyexecutor or _torch/models -- the worker-side signal
+    n_engine_body_lines: int  # lines in engine files MINUS def/class/decorator/import lines
+    # (shell-style hits get covered just by importing; this metric
+    # tells us whether engine function bodies were actually executed)
+    context_key: str  # matched coverage context (may differ from test_id)
 
 
 # --- junit -----------------------------------------------------------------
 
+
 def _junit_class_part(classname: str, file_attr: str) -> str:
-    """Strip the dotted module prefix from a junit ``classname`` when the
-    ``file`` attribute is present."""
+    """Strip the dotted module prefix from a junit ``classname``.
+
+    Only does this when the ``file`` attribute is present.
+    """
     if not classname or not file_attr:
         return ""
     mod_dotted = file_attr.replace("/", ".").rsplit(".py", 1)[0]
     if classname == mod_dotted:
         return ""
     if classname.startswith(mod_dotted + "."):
-        return classname[len(mod_dotted) + 1:]
+        return classname[len(mod_dotted) + 1 :]
     return classname
 
 
@@ -189,9 +193,7 @@ def _junit_to_nodeid(classname: str, name: str, file_attr: str) -> str:
     if not classname:
         return name
     parts = classname.split(".")
-    first_class_idx = next(
-        (i for i, p in enumerate(parts) if p[:1].isupper()), None
-    )
+    first_class_idx = next((i for i, p in enumerate(parts) if p[:1].isupper()), None)
     if first_class_idx is None:
         file_path = "/".join(parts) + ".py"
         return f"{file_path}::{name}"
@@ -230,15 +232,17 @@ def load_junit(path: Path) -> dict:
 
 # --- coverage db -----------------------------------------------------------
 
+
 @functools.lru_cache(maxsize=None)
 def _decl_import_lines(file_path: str) -> frozenset:
-    """Lines that are def/class/decorator/import -- 'shell' code that's
-    counted as covered just by importing the module.
+    """Lines that are def/class/decorator/import -- 'shell' code.
+
+    These count as covered just by importing the module.
 
     For any file we can't parse (missing, syntax error, encoding), return
     an empty frozenset, which means we'll count ALL covered lines as
     body -- conservative for audit purposes (we'd rather over-count body
-    coverage than spuriously deflate it for unparseable files).
+    coverage than spuriously deflate it for unparsable files).
     """
     try:
         with open(file_path) as f:
@@ -294,6 +298,7 @@ def load_contexts(db_path: Path) -> dict:
 
 # --- matching --------------------------------------------------------------
 
+
 def match_test_to_context(test_id: str, contexts: dict) -> str:
     """Find the context for ``test_id``. Returns ``""`` if no match."""
     if test_id in contexts:
@@ -348,8 +353,7 @@ def classify(ctx_info, has_sentinel, junit_status="unknown", test_id="") -> str:
         return "TEST_FAILED"
     if ctx_info["n_engine"] == 0:
         return "MAIN_ONLY"
-    if (_is_integration_test(test_id)
-            and not ctx_info.get("hit_main_executor", False)):
+    if _is_integration_test(test_id) and not ctx_info.get("hit_main_executor", False):
         return "SUBPROCESS_LOST"
     return "OK"
 
@@ -372,6 +376,7 @@ def _nodeid_sentinel_name(nodeid: str) -> str:
 
 # --- save error scan -------------------------------------------------------
 
+
 def count_save_errors(log_path: Path) -> int:
     count = 0
     with open(log_path, errors="replace") as f:
@@ -383,18 +388,33 @@ def count_save_errors(log_path: Path) -> int:
 
 # --- outputs ---------------------------------------------------------------
 
+
 def write_csv(path: Path, records):
     with open(path, "w", newline="") as f:
         w = csv.writer(f)
-        w.writerow([
-            "test_id", "junit_status", "cov_status", "n_files",
-            "n_engine_files", "n_engine_body_lines", "context_key",
-        ])
+        w.writerow(
+            [
+                "test_id",
+                "junit_status",
+                "cov_status",
+                "n_files",
+                "n_engine_files",
+                "n_engine_body_lines",
+                "context_key",
+            ]
+        )
         for r in records:
-            w.writerow([
-                r.test_id, r.junit_status, r.cov_status, r.n_files,
-                r.n_engine_files, r.n_engine_body_lines, r.context_key,
-            ])
+            w.writerow(
+                [
+                    r.test_id,
+                    r.junit_status,
+                    r.cov_status,
+                    r.n_files,
+                    r.n_engine_files,
+                    r.n_engine_body_lines,
+                    r.context_key,
+                ]
+            )
 
 
 def write_json(path: Path, records, save_errs, db_path):
@@ -413,10 +433,16 @@ def write_json(path: Path, records, save_errs, db_path):
 # every run; the rest are only shown when non-zero, to keep the summary
 # uncluttered when the relevant signal isn't present in this run.
 _REPORTED_STATUSES = (
-    "OK", "SUBPROCESS_LOST", "MAIN_ONLY", "EMPTY",
+    "OK",
+    "SUBPROCESS_LOST",
+    "MAIN_ONLY",
+    "EMPTY",
     "TEST_FAILED",
-    "NO_PRODUCT_CODE", "FAILED_EARLY",
-    "MISSING", "MISSING_LOST", "MISSING_NO_PLUGIN",
+    "NO_PRODUCT_CODE",
+    "FAILED_EARLY",
+    "MISSING",
+    "MISSING_LOST",
+    "MISSING_NO_PLUGIN",
 )
 _ALWAYS_SHOW = frozenset({"OK", "SUBPROCESS_LOST", "MAIN_ONLY", "EMPTY"})
 
@@ -441,38 +467,41 @@ def print_summary(records, save_errs):
         if not examples:
             continue
         n_show = min(5, len(examples))
-        print(f"\nFirst {n_show} {status} tests "
-              f"(of {len(examples)}):")
+        print(f"\nFirst {n_show} {status} tests (of {len(examples)}):")
         for r in examples[:n_show]:
-            print(f"  {r.test_id}  "
-                  f"(files={r.n_files}, engine={r.n_engine_files}, "
-                  f"body_lines={r.n_engine_body_lines})")
+            print(
+                f"  {r.test_id}  "
+                f"(files={r.n_files}, engine={r.n_engine_files}, "
+                f"body_lines={r.n_engine_body_lines})"
+            )
 
 
 # --- main ------------------------------------------------------------------
 
+
 def main(argv=None):
     p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
-    p.add_argument("--db", type=Path, required=True,
-                   help="combined .coverage.<stage> DB")
-    p.add_argument("--junit", type=Path,
-                   help="pytest junit.xml (enables MISSING detection)")
-    p.add_argument("--pytest-log", type=Path,
-                   help="pytest stdout/stderr log (scanned for save errors)")
-    p.add_argument("--sentinel-dir", type=Path,
-                   help="directory of per-test sentinel files written by "
-                        "cbts_plugin (enables MISSING_LOST vs "
-                        "MISSING_NO_PLUGIN split)")
-    p.add_argument("--out-csv", type=Path,
-                   help="per-test CSV report (recommended for triage)")
-    p.add_argument("--out-json", type=Path,
-                   help="per-test JSON report (recommended for trend metrics)")
+    p.add_argument("--db", type=Path, required=True, help="combined .coverage.<stage> DB")
+    p.add_argument("--junit", type=Path, help="pytest junit.xml (enables MISSING detection)")
+    p.add_argument(
+        "--pytest-log", type=Path, help="pytest stdout/stderr log (scanned for save errors)"
+    )
+    p.add_argument(
+        "--sentinel-dir",
+        type=Path,
+        help="directory of per-test sentinel files written by "
+        "cbts_plugin (enables MISSING_LOST vs "
+        "MISSING_NO_PLUGIN split)",
+    )
+    p.add_argument("--out-csv", type=Path, help="per-test CSV report (recommended for triage)")
+    p.add_argument(
+        "--out-json", type=Path, help="per-test JSON report (recommended for trend metrics)"
+    )
     args = p.parse_args(argv)
 
     # ---- Hard signals on the DB ----
     if not args.db.exists() or args.db.stat().st_size == 0:
-        print(f"FATAL: coverage DB {args.db} missing or empty",
-              file=sys.stderr)
+        print(f"FATAL: coverage DB {args.db} missing or empty", file=sys.stderr)
         return 2
 
     contexts = load_contexts(args.db)
@@ -507,22 +536,20 @@ def main(argv=None):
             continue  # NOT_RUN -- not a coverage gap
         ctx_name = match_test_to_context(test_id, contexts)
         ctx_info = contexts.get(ctx_name) if ctx_name else None
-        has_sentinel = (
-            _nodeid_sentinel_name(test_id) in sentinels
-            if sentinels_in_use else None
-        )
+        has_sentinel = _nodeid_sentinel_name(test_id) in sentinels if sentinels_in_use else None
         if ctx_name:
             matched_contexts.add(ctx_name)
-        records.append(TestRecord(
-            test_id=test_id,
-            junit_status=status,
-            cov_status=classify(ctx_info, has_sentinel,
-                                junit_status=status, test_id=test_id),
-            n_files=ctx_info["n_files"] if ctx_info else 0,
-            n_engine_files=ctx_info["n_engine"] if ctx_info else 0,
-            n_engine_body_lines=ctx_info["n_engine_body"] if ctx_info else 0,
-            context_key=ctx_name,
-        ))
+        records.append(
+            TestRecord(
+                test_id=test_id,
+                junit_status=status,
+                cov_status=classify(ctx_info, has_sentinel, junit_status=status, test_id=test_id),
+                n_files=ctx_info["n_files"] if ctx_info else 0,
+                n_engine_files=ctx_info["n_engine"] if ctx_info else 0,
+                n_engine_body_lines=ctx_info["n_engine_body"] if ctx_info else 0,
+                context_key=ctx_name,
+            )
+        )
 
     # Contexts with no matching junit row: still report them. Parametrized
     # tests, multi-call hooks, and fixture-level captures can land here.
@@ -531,15 +558,17 @@ def main(argv=None):
     for ctx_name, ctx_info in contexts.items():
         if ctx_name in matched_contexts:
             continue
-        records.append(TestRecord(
-            test_id=ctx_name,
-            junit_status="NO_JUNIT_MATCH" if junit_tests else "unknown",
-            cov_status=classify(ctx_info, has_sentinel=None, test_id=ctx_name),
-            n_files=ctx_info["n_files"],
-            n_engine_files=ctx_info["n_engine"],
-            n_engine_body_lines=ctx_info["n_engine_body"],
-            context_key=ctx_name,
-        ))
+        records.append(
+            TestRecord(
+                test_id=ctx_name,
+                junit_status="NO_JUNIT_MATCH" if junit_tests else "unknown",
+                cov_status=classify(ctx_info, has_sentinel=None, test_id=ctx_name),
+                n_files=ctx_info["n_files"],
+                n_engine_files=ctx_info["n_engine"],
+                n_engine_body_lines=ctx_info["n_engine_body"],
+                context_key=ctx_name,
+            )
+        )
 
     # ---- Outputs ----
     if args.out_csv: