| import argparse |
| import concurrent.futures |
| import json |
| import logging |
| import os |
| import re |
| import subprocess |
| import time |
| from enum import Enum |
| from typing import List, NamedTuple, Optional, Pattern |
| |
| |
| LINTER_CODE = "CMAKE" |
| |
| |
| class LintSeverity(str, Enum): |
| ERROR = "error" |
| WARNING = "warning" |
| ADVICE = "advice" |
| DISABLED = "disabled" |
| |
| |
| class LintMessage(NamedTuple): |
| path: Optional[str] |
| line: Optional[int] |
| char: Optional[int] |
| code: str |
| severity: LintSeverity |
| name: str |
| original: Optional[str] |
| replacement: Optional[str] |
| description: Optional[str] |
| |
| |
| # CMakeLists.txt:901: Lines should be <= 80 characters long [linelength] |
| RESULTS_RE: Pattern[str] = re.compile( |
| r"""(?mx) |
| ^ |
| (?P<file>.*?): |
| (?P<line>\d+): |
| \s(?P<message>.*) |
| \s(?P<code>\[.*\]) |
| $ |
| """ |
| ) |
| |
| |
| def run_command( |
| args: List[str], |
| ) -> "subprocess.CompletedProcess[bytes]": |
| logging.debug("$ %s", " ".join(args)) |
| start_time = time.monotonic() |
| try: |
| return subprocess.run( |
| args, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE, |
| ) |
| finally: |
| end_time = time.monotonic() |
| logging.debug("took %dms", (end_time - start_time) * 1000) |
| |
| |
| def check_file( |
| filename: str, |
| config: str, |
| ) -> List[LintMessage]: |
| try: |
| proc = run_command( |
| ["cmakelint", f"--config={config}", filename], |
| ) |
| except OSError as err: |
| return [ |
| LintMessage( |
| path=None, |
| line=None, |
| char=None, |
| code=LINTER_CODE, |
| severity=LintSeverity.ERROR, |
| name="command-failed", |
| original=None, |
| replacement=None, |
| description=(f"Failed due to {err.__class__.__name__}:\n{err}"), |
| ) |
| ] |
| stdout = str(proc.stdout, "utf-8").strip() |
| return [ |
| LintMessage( |
| path=match["file"], |
| name=match["code"], |
| description=match["message"], |
| line=int(match["line"]), |
| char=None, |
| code=LINTER_CODE, |
| severity=LintSeverity.ERROR, |
| original=None, |
| replacement=None, |
| ) |
| for match in RESULTS_RE.finditer(stdout) |
| ] |
| |
| |
| if __name__ == "__main__": |
| parser = argparse.ArgumentParser( |
| description="cmakelint runner", |
| fromfile_prefix_chars="@", |
| ) |
| parser.add_argument( |
| "--config", |
| required=True, |
| help="location of cmakelint config", |
| ) |
| parser.add_argument( |
| "filenames", |
| nargs="+", |
| help="paths to lint", |
| ) |
| |
| args = parser.parse_args() |
| |
| with concurrent.futures.ThreadPoolExecutor( |
| max_workers=os.cpu_count(), |
| thread_name_prefix="Thread", |
| ) as executor: |
| futures = { |
| executor.submit( |
| check_file, |
| filename, |
| args.config, |
| ): filename |
| for filename in args.filenames |
| } |
| for future in concurrent.futures.as_completed(futures): |
| try: |
| for lint_message in future.result(): |
| print(json.dumps(lint_message._asdict()), flush=True) |
| except Exception: |
| logging.critical('Failed at "%s".', futures[future]) |
| raise |