| # Copyright 2022 The Pigweed Authors |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); you may not |
| # use this file except in compliance with the License. You may obtain a copy of |
| # the License at |
| # |
| # https://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, 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. |
| """Keep specified lists sorted.""" |
| |
| import argparse |
| import dataclasses |
| import difflib |
| import logging |
| import os |
| from pathlib import Path |
| import re |
| import sys |
| from typing import ( |
| Callable, |
| Collection, |
| List, |
| Optional, |
| Pattern, |
| Sequence, |
| Tuple, |
| Union, |
| ) |
| |
| import pw_cli |
| from . import cli, git_repo, presubmit, tools |
| |
| _COLOR = pw_cli.color.colors() |
| _LOG: logging.Logger = logging.getLogger(__name__) |
| |
| # Ignore a whole section. Please do not change the order of these lines. |
| _START = re.compile(r'keep-sorted: (begin|start)', re.IGNORECASE) |
| _END = re.compile(r'keep-sorted: (stop|end)', re.IGNORECASE) |
| _IGNORE_CASE = re.compile(r'ignore-case', re.IGNORECASE) |
| _ALLOW_DUPES = re.compile(r'allow-dupes', re.IGNORECASE) |
| _IGNORE_PREFIX = re.compile(r'ignore-prefix=(\S+)', re.IGNORECASE) |
| _STICKY_COMMENTS = re.compile(r'sticky-comments=(\S+)', re.IGNORECASE) |
| |
| # Only include these literals here so keep_sorted doesn't try to reorder later |
| # test lines. |
| START, END = """ |
| keep-sorted: start |
| keep-sorted: end |
| """.strip().splitlines() |
| |
| |
| @dataclasses.dataclass |
| class KeepSortedContext: |
| paths: List[Path] |
| fix: bool |
| failed: bool = False |
| |
| def fail(self, |
| description: str = '', |
| path: Optional[Path] = None, |
| line: Optional[int] = None) -> None: |
| if not self.fix: |
| self.failed = True |
| |
| line_part: str = '' |
| if line is not None: |
| line_part = f'{line}:' |
| |
| log = _LOG.error |
| if self.fix: |
| log = _LOG.warning |
| |
| if path: |
| log('%s:%s %s', path, line_part, description) |
| else: |
| log('%s', description) |
| |
| |
| class KeepSortedParsingError(presubmit.PresubmitFailure): |
| pass |
| |
| |
| @dataclasses.dataclass |
| class _Line: |
| value: str = '' |
| sticky_comments: Sequence[str] = () |
| |
| @property |
| def full(self): |
| return ''.join((*self.sticky_comments, self.value)) |
| |
| def __lt__(self, other): |
| if not isinstance(other, _Line): |
| return NotImplemented |
| if self.value != other.value: |
| return self.value < other.value |
| return self.sticky_comments < other.sticky_comments |
| |
| |
| @dataclasses.dataclass |
| class _Block: |
| ignore_case: bool = False |
| allow_dupes: bool = False |
| ignored_prefixes: Sequence[str] = dataclasses.field(default_factory=list) |
| sticky_comments: Tuple[str, ...] = () |
| start_line_number: int = -1 |
| start_line: str = '' |
| end_line: str = '' |
| lines: List[str] = dataclasses.field(default_factory=list) |
| |
| |
| class _FileSorter: |
| def __init__(self, ctx: Union[presubmit.PresubmitContext, |
| KeepSortedContext], path: Path): |
| self.ctx = ctx |
| self.path: Path = path |
| self.all_lines: List[str] = [] |
| self.changed: bool = False |
| |
| def _process_block(self, block: _Block) -> Sequence[str]: |
| raw_lines: List[str] = block.lines |
| lines: List[_Line] = [] |
| |
| if block.sticky_comments: |
| comments: List[str] = [] |
| for raw_line in raw_lines: |
| if raw_line.lstrip().startswith(block.sticky_comments): |
| _LOG.debug('found sticky %s', raw_line.strip()) |
| comments.append(raw_line) |
| else: |
| _LOG.debug('non-sticky %s', raw_line.strip()) |
| line = _Line(raw_line, tuple(comments)) |
| _LOG.debug('line %s', line) |
| lines.append(line) |
| comments = [] |
| if comments: |
| self.ctx.fail( |
| f'sticky comment at end of block: {comments[0].strip()}', |
| self.path, block.start_line_number) |
| |
| else: |
| lines = [_Line(x) for x in block.lines] |
| |
| if not block.allow_dupes: |
| lines = list({x.full: x for x in lines}.values()) |
| |
| StrLinePair = Tuple[str, _Line] |
| sort_key_funcs: List[Callable[[StrLinePair], StrLinePair]] = [] |
| |
| if block.ignored_prefixes: |
| |
| def strip_ignored_prefixes(val): |
| """Remove one ignored prefix from val, if present.""" |
| wo_white = val[0].lstrip() |
| white = val[0][0:-len(wo_white)] |
| for prefix in block.ignored_prefixes: |
| if wo_white.startswith(prefix): |
| return (f'{white}{wo_white[len(prefix):]}', val[1]) |
| return (val[0], val[1]) |
| |
| sort_key_funcs.append(strip_ignored_prefixes) |
| |
| if block.ignore_case: |
| sort_key_funcs.append(lambda val: (val[0].lower(), val[1])) |
| |
| def sort_key(line): |
| vals = (line.value, line) |
| for sort_key_func in sort_key_funcs: |
| vals = sort_key_func(vals) |
| return vals |
| |
| for val in lines: |
| _LOG.debug('For sorting: %r => %r', val, sort_key(val)) |
| |
| sorted_lines = sorted(lines, key=sort_key) |
| raw_sorted_lines: List[str] = [] |
| for line in sorted_lines: |
| raw_sorted_lines.extend(line.sticky_comments) |
| raw_sorted_lines.append(line.value) |
| |
| if block.lines != raw_sorted_lines: |
| self.changed = True |
| self.ctx.fail('keep-sorted block is not sorted', self.path, |
| block.start_line_number) |
| _LOG.info(' %s', block.start_line.rstrip()) |
| diff = difflib.Differ() |
| for dline in diff.compare( |
| [x.rstrip() for x in block.lines], |
| [x.rstrip() for x in raw_sorted_lines], |
| ): |
| if dline.startswith('-'): |
| dline = _COLOR.red(dline) |
| elif dline.startswith('+'): |
| dline = _COLOR.green(dline) |
| _LOG.info(dline) |
| _LOG.info(' %s', block.end_line.rstrip()) |
| |
| return raw_sorted_lines |
| |
| def _parse_file(self, ins): |
| block: Optional[_Block] = None |
| |
| for i, line in enumerate(ins, start=1): |
| if block: |
| if _START.search(line): |
| raise KeepSortedParsingError( |
| f'found {line.strip()!r} inside keep-sorted block', |
| self.path, i) |
| |
| if _END.search(line): |
| _LOG.debug('Found end line %d %r', i, line) |
| block.end_line = line |
| self.all_lines.extend(self._process_block(block)) |
| block = None |
| self.all_lines.append(line) |
| |
| else: |
| _LOG.debug('Adding to block line %d %r', i, line) |
| block.lines.append(line) |
| |
| elif start_match := _START.search(line): |
| _LOG.debug('Found start line %d %r', i, line) |
| |
| block = _Block() |
| |
| block.ignore_case = bool(_IGNORE_CASE.search(line)) |
| _LOG.debug('ignore_case: %s', block.ignore_case) |
| |
| block.allow_dupes = bool(_ALLOW_DUPES.search(line)) |
| _LOG.debug('allow_dupes: %s', block.allow_dupes) |
| |
| match = _IGNORE_PREFIX.search(line) |
| if match: |
| block.ignored_prefixes = match.group(1).split(',') |
| |
| # We want to check the longest prefixes first, in case one |
| # prefix is a prefix of another prefix. |
| block.ignored_prefixes.sort(key=lambda x: (-len(x), x)) |
| _LOG.debug('ignored_prefixes: %r', block.ignored_prefixes) |
| |
| match = _STICKY_COMMENTS.search(line) |
| if match: |
| if match.group(1) == 'no': |
| block.sticky_comments = () |
| else: |
| block.sticky_comments = tuple( |
| match.group(1).split(',')) |
| else: |
| prefix = line[:start_match.start()].strip() |
| if prefix and len(prefix) <= 3: |
| block.sticky_comments = (prefix, ) |
| _LOG.debug('sticky_comments: %s', block.sticky_comments) |
| |
| block.start_line = line |
| block.start_line_number = i |
| self.all_lines.append(line) |
| |
| remaining = line[start_match.end():].strip() |
| remaining = _IGNORE_CASE.sub('', remaining, count=1).strip() |
| remaining = _ALLOW_DUPES.sub('', remaining, count=1).strip() |
| remaining = _IGNORE_PREFIX.sub('', remaining, count=1).strip() |
| remaining = _STICKY_COMMENTS.sub('', remaining, |
| count=1).strip() |
| if remaining.strip(): |
| raise KeepSortedParsingError( |
| f'unrecognized directive on keep-sorted line: ' |
| f'{remaining}', self.path, i) |
| |
| elif _END.search(line): |
| raise KeepSortedParsingError( |
| f'found {line.strip()!r} outside keep-sorted block', |
| self.path, i) |
| |
| else: |
| self.all_lines.append(line) |
| |
| if block: |
| raise KeepSortedParsingError( |
| f'found EOF while looking for "{END}"', self.path) |
| |
| def sort(self) -> None: |
| """Check for unsorted keep-sorted blocks.""" |
| _LOG.debug('Evaluating path %s', self.path) |
| try: |
| with self.path.open() as ins: |
| _LOG.debug('Processing %s', self.path) |
| self._parse_file(ins) |
| |
| except UnicodeDecodeError: |
| # File is not text, like a gif. |
| _LOG.debug('File %s is not a text file', self.path) |
| |
| def write(self, path: Optional[Path] = None) -> None: |
| if not self.changed: |
| return |
| if not path: |
| path = self.path |
| with path.open('w') as outs: |
| outs.writelines(self.all_lines) |
| _LOG.info('Applied keep-sorted changes to %s', path) |
| |
| |
| def _print_howto_fix(paths: Sequence[Path]) -> None: |
| def path_relative_to_cwd(path): |
| try: |
| return Path(path).resolve().relative_to(Path.cwd().resolve()) |
| except ValueError: |
| return Path(path).resolve() |
| |
| message = (f' pw keep-sorted --fix {path_relative_to_cwd(path)}' |
| for path in paths) |
| _LOG.warning('To sort these blocks, run:\n\n%s\n', '\n'.join(message)) |
| |
| |
| def _process_files( |
| ctx: Union[presubmit.PresubmitContext, |
| KeepSortedContext]) -> Sequence[Path]: |
| fix = getattr(ctx, 'fix', False) |
| changed_paths = [] |
| |
| for path in ctx.paths: |
| if path.is_symlink() or path.is_dir(): |
| continue |
| |
| try: |
| sorter = _FileSorter(ctx, path) |
| |
| sorter.sort() |
| if sorter.changed: |
| changed_paths.append(path) |
| if fix: |
| sorter.write() |
| |
| except KeepSortedParsingError as exc: |
| ctx.fail(str(exc)) |
| |
| return changed_paths |
| |
| |
| @presubmit.check(name='keep_sorted') |
| def presubmit_check(ctx: presubmit.PresubmitContext) -> None: |
| """Presubmit check that ensures specified lists remain sorted.""" |
| |
| changed_paths = _process_files(ctx) |
| |
| if changed_paths: |
| _print_howto_fix(changed_paths) |
| |
| |
| def parse_args() -> argparse.Namespace: |
| """Creates an argument parser and parses arguments.""" |
| |
| parser = argparse.ArgumentParser(description=__doc__) |
| cli.add_path_arguments(parser) |
| parser.add_argument('--fix', |
| action='store_true', |
| help='Apply fixes in place.') |
| |
| return parser.parse_args() |
| |
| |
| def keep_sorted_in_repo(paths: Collection[Union[Path, str]], fix: bool, |
| exclude: Collection[Pattern[str]], base: str) -> int: |
| """Checks or fixes keep-sorted blocks for files in a Git repo.""" |
| |
| files = [Path(path).resolve() for path in paths if os.path.isfile(path)] |
| repo = git_repo.root() if git_repo.is_repo() else None |
| |
| # Implement a graceful fallback in case the tracking branch isn't available. |
| if (base == git_repo.TRACKING_BRANCH_ALIAS |
| and not git_repo.tracking_branch(repo)): |
| _LOG.warning( |
| 'Failed to determine the tracking branch, using --base HEAD~1 ' |
| 'instead of listing all files') |
| base = 'HEAD~1' |
| |
| # If this is a Git repo, list the original paths with git ls-files or diff. |
| if repo: |
| project_root = Path(pw_cli.env.pigweed_environment().PW_PROJECT_ROOT) |
| _LOG.info( |
| 'Sorting %s', |
| git_repo.describe_files(repo, Path.cwd(), base, paths, exclude, |
| project_root)) |
| |
| # Add files from Git and remove duplicates. |
| files = sorted( |
| set(tools.exclude_paths(exclude, git_repo.list_files(base, paths))) |
| | set(files)) |
| elif base: |
| _LOG.critical( |
| 'A base commit may only be provided if running from a Git repo') |
| return 1 |
| |
| ctx = KeepSortedContext(paths=files, fix=fix) |
| changed_paths = _process_files(ctx) |
| |
| if not fix and changed_paths: |
| _print_howto_fix(changed_paths) |
| |
| return int(ctx.failed) |
| |
| |
| def main() -> int: |
| return keep_sorted_in_repo(**vars(parse_args())) |
| |
| |
| if __name__ == '__main__': |
| pw_cli.log.install(logging.INFO) |
| sys.exit(main()) |
