tools/lint/python/black.py

   1 # This Source Code Form is subject to the terms of the Mozilla Public
   2 # License, v. 2.0. If a copy of the MPL was not distributed with this
   3 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
   4
   5 import os
   6 import platform
   7 import re
   8 import signal
   9 import subprocess
  10 import sys
  11
  12 import mozpack.path as mozpath
  13 from mozfile import which
  14 from mozlint import result
  15 from mozlint.pathutils import expand_exclusions
  16 from mozprocess import ProcessHandler
  17
  18 here = os.path.abspath(os.path.dirname(__file__))
  19 BLACK_REQUIREMENTS_PATH = os.path.join(here, "black_requirements.txt")
  20
  21 BLACK_INSTALL_ERROR = """
  22 Unable to install correct version of black
  23 Try to install it manually with:
  24     $ pip install -U --require-hashes -r {}
  25 """.strip().format(
  26     BLACK_REQUIREMENTS_PATH
  27 )
  28
  29
  30 def default_bindir():
  31     # We use sys.prefix to find executables as that gets modified with
  32     # virtualenv's activate_this.py, whereas sys.executable doesn't.
  33     if platform.system() == "Windows":
  34         return os.path.join(sys.prefix, "Scripts")
  35     else:
  36         return os.path.join(sys.prefix, "bin")
  37
  38
  39 def get_black_version(binary):
  40     """
  41     Returns found binary's version
  42     """
  43     try:
  44         output = subprocess.check_output(
  45             [binary, "--version"],
  46             stderr=subprocess.STDOUT,
  47             universal_newlines=True,
  48         )
  49     except subprocess.CalledProcessError as e:
  50         output = e.output
  51     try:
  52         # Accept `black.EXE, version ...` on Windows.
  53         # for old version of black, the output is
  54         # black, version 21.4b2
  55         # From black 21.11b1, the output is like
  56         # black, 21.11b1 (compiled: no)
  57         return re.match(r"black.*,( version)? (\S+)", output)[2]
  58     except TypeError as e:
  59         print("Could not parse the version '{}'".format(output))
  60         print("Error: {}".format(e))
  61
  62
  63 def parse_issues(config, output, paths, *, log):
  64     would_reformat = re.compile("^would reformat (.*)$", re.I)
  65     reformatted = re.compile("^reformatted (.*)$", re.I)
  66     cannot_reformat = re.compile("^error: cannot format (.*?): (.*)$", re.I)
  67     results = []
  68     for line in output:
  69         line = line.decode("utf-8")
  70         if line.startswith("All done!") or line.startswith("Oh no!"):
  71             break
  72
  73         match = would_reformat.match(line)
  74         if match:
  75             res = {"path": match.group(1), "level": "error"}
  76             results.append(result.from_config(config, **res))
  77             continue
  78
  79         match = reformatted.match(line)
  80         if match:
  81             res = {"path": match.group(1), "level": "warning", "message": "reformatted"}
  82             results.append(result.from_config(config, **res))
  83             continue
  84
  85         match = cannot_reformat.match(line)
  86         if match:
  87             res = {"path": match.group(1), "level": "error", "message": match.group(2)}
  88             results.append(result.from_config(config, **res))
  89             continue
  90
  91         log.debug(f"Unhandled line: {line}")
  92     return results
  93
  94
  95 class BlackProcess(ProcessHandler):
  96     def __init__(self, config, *args, **kwargs):
  97         self.config = config
  98         kwargs["stream"] = False
  99         ProcessHandler.__init__(self, *args, **kwargs)
 100
 101     def run(self, *args, **kwargs):
 102         orig = signal.signal(signal.SIGINT, signal.SIG_IGN)
 103         ProcessHandler.run(self, *args, **kwargs)
 104         signal.signal(signal.SIGINT, orig)
 105
 106
 107 def run_process(config, cmd):
 108     proc = BlackProcess(config, cmd)
 109     proc.run()
 110     try:
 111         proc.wait()
 112     except KeyboardInterrupt:
 113         proc.kill()
 114
 115     return proc.output
 116
 117
 118 def setup(root, **lintargs):
 119     log = lintargs["log"]
 120     virtualenv_bin_path = lintargs.get("virtualenv_bin_path")
 121     # Using `which` searches multiple directories and handles `.exe` on Windows.
 122     binary = which("black", path=(virtualenv_bin_path, default_bindir()))
 123
 124     if binary and os.path.exists(binary):
 125         binary = mozpath.normsep(binary)
 126         log.debug("Looking for black at {}".format(binary))
 127         version = get_black_version(binary)
 128         versions = [
 129             line.split()[0].strip()
 130             for line in open(BLACK_REQUIREMENTS_PATH).readlines()
 131             if line.startswith("black==")
 132         ]
 133         if ["black=={}".format(version)] == versions:
 134             log.debug("Black is present with expected version {}".format(version))
 135             return 0
 136         else:
 137             log.debug("Black is present but unexpected version {}".format(version))
 138
 139     log.debug("Black needs to be installed or updated")
 140     virtualenv_manager = lintargs["virtualenv_manager"]
 141     try:
 142         virtualenv_manager.install_pip_requirements(BLACK_REQUIREMENTS_PATH, quiet=True)
 143     except subprocess.CalledProcessError:
 144         print(BLACK_INSTALL_ERROR)
 145         return 1
 146
 147
 148 def run_black(config, paths, fix=None, *, log, virtualenv_bin_path):
 149     fixed = 0
 150     binary = os.path.join(virtualenv_bin_path or default_bindir(), "black")
 151
 152     log.debug("Black version {}".format(get_black_version(binary)))
 153
 154     cmd_args = [binary]
 155     if not fix:
 156         cmd_args.append("--check")
 157
 158     base_command = cmd_args + paths
 159     log.debug("Command: {}".format(" ".join(base_command)))
 160     output = parse_issues(config, run_process(config, base_command), paths, log=log)
 161
 162     # black returns an issue for fixed files as well
 163     for eachIssue in output:
 164         if eachIssue.message == "reformatted":
 165             fixed += 1
 166
 167     return {"results": output, "fixed": fixed}
 168
 169
 170 def lint(paths, config, fix=None, **lintargs):
 171     files = list(expand_exclusions(paths, config, lintargs["root"]))
 172
 173     return run_black(
 174         config,
 175         files,
 176         fix=fix,
 177         log=lintargs["log"],
 178         virtualenv_bin_path=lintargs.get("virtualenv_bin_path"),
 179     )