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 from mozlint import result
  13 from mozlint.pathutils import expand_exclusions
  14
  15 here = os.path.abspath(os.path.dirname(__file__))
  16
  17
  18 def default_bindir():
  19     # We use sys.prefix to find executables as that gets modified with
  20     # virtualenv's activate_this.py, whereas sys.executable doesn't.
  21     if platform.system() == "Windows":
  22         return os.path.join(sys.prefix, "Scripts")
  23     else:
  24         return os.path.join(sys.prefix, "bin")
  25
  26
  27 def get_black_version(binary):
  28     """
  29     Returns found binary's version
  30     """
  31     try:
  32         output = subprocess.check_output(
  33             [binary, "--version"],
  34             stderr=subprocess.STDOUT,
  35             universal_newlines=True,
  36         )
  37     except subprocess.CalledProcessError as e:
  38         output = e.output
  39     try:
  40         # Accept `black.EXE, version ...` on Windows.
  41         # for old version of black, the output is
  42         # black, version 21.4b2
  43         # From black 21.11b1, the output is like
  44         # black, 21.11b1 (compiled: no)
  45         return re.match(r"black.*,( version)? (\S+)", output)[2]
  46     except TypeError as e:
  47         print("Could not parse the version '{}'".format(output))
  48         print("Error: {}".format(e))
  49
  50
  51 def parse_issues(config, output, paths, *, log):
  52     would_reformat = re.compile("^would reformat (.*)$", re.I)
  53     reformatted = re.compile("^reformatted (.*)$", re.I)
  54     cannot_reformat = re.compile("^error: cannot format (.*?): (.*)$", re.I)
  55     results = []
  56     for l in output.split(b"\n"):
  57         line = l.decode("utf-8").rstrip("\r\n")
  58         if line.startswith("All done!") or line.startswith("Oh no!"):
  59             break
  60
  61         match = would_reformat.match(line)
  62         if match:
  63             res = {"path": match.group(1), "level": "error"}
  64             results.append(result.from_config(config, **res))
  65             continue
  66
  67         match = reformatted.match(line)
  68         if match:
  69             res = {"path": match.group(1), "level": "warning", "message": "reformatted"}
  70             results.append(result.from_config(config, **res))
  71             continue
  72
  73         match = cannot_reformat.match(line)
  74         if match:
  75             res = {"path": match.group(1), "level": "error", "message": match.group(2)}
  76             results.append(result.from_config(config, **res))
  77             continue
  78
  79         log.debug(f"Unhandled line: {line}")
  80     return results
  81
  82
  83 def run_process(config, cmd):
  84     orig = signal.signal(signal.SIGINT, signal.SIG_IGN)
  85     proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
  86     signal.signal(signal.SIGINT, orig)
  87     try:
  88         output, _ = proc.communicate()
  89         proc.wait()
  90     except KeyboardInterrupt:
  91         proc.kill()
  92
  93     return output
  94
  95
  96 def run_black(config, paths, fix=None, *, log, virtualenv_bin_path):
  97     fixed = 0
  98     binary = os.path.join(virtualenv_bin_path or default_bindir(), "black")
  99
 100     log.debug("Black version {}".format(get_black_version(binary)))
 101
 102     cmd_args = [binary]
 103     if not fix:
 104         cmd_args.append("--check")
 105
 106     base_command = cmd_args + paths
 107     log.debug("Command: {}".format(" ".join(base_command)))
 108     output = parse_issues(config, run_process(config, base_command), paths, log=log)
 109
 110     # black returns an issue for fixed files as well
 111     for eachIssue in output:
 112         if eachIssue.message == "reformatted":
 113             fixed += 1
 114
 115     return {"results": output, "fixed": fixed}
 116
 117
 118 def lint(paths, config, fix=None, **lintargs):
 119     files = list(expand_exclusions(paths, config, lintargs["root"]))
 120
 121     return run_black(
 122         config,
 123         files,
 124         fix=fix,
 125         log=lintargs["log"],
 126         virtualenv_bin_path=lintargs.get("virtualenv_bin_path"),
 127     )