1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
|
"""Class to support running various linters in a common framework."""
from typing import Dict, List, Optional
import difflib
import logging
import os
import re
import site
import subprocess
import sys
import threading
import pkg_resources
from . import base
def _check_version(linter, cmd_path, args):
# type: (base.LinterBase, List[str], List[str]) -> bool
"""Check if the given linter has the correct version."""
try:
cmd = cmd_path + args
logging.info(str(cmd))
process_handle = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
output, stderr = process_handle.communicate()
decoded_output = output.decode("utf-8")
if process_handle.returncode:
logging.info(
"Version check failed for [%s], return code '%d'."
"Standard Output:\n%s\nStandard Error:\n%s",
cmd,
process_handle.returncode,
decoded_output,
stderr,
)
pattern = r"\b(?:(%s) )?(?P<version>\S+)\b" % (linter.cmd_name)
required_version = pkg_resources.parse_version(linter.required_version)
match = re.search(pattern, decoded_output)
if match:
found_version = match.group("version")
else:
found_version = "0.0"
if pkg_resources.parse_version(found_version) < required_version:
logging.info(
"Linter %s has wrong version for '%s'. Expected >= '%s',"
"Standard Output:\n'%s'\nStandard Error:\n%s",
linter.cmd_name,
cmd,
required_version,
decoded_output,
stderr,
)
return False
except OSError as os_error:
# The WindowsError exception is thrown if the command is not found.
# We catch OSError since WindowsError does not exist on non-Windows platforms.
logging.info("Version check command [%s] failed: %s", cmd, os_error)
return False
return True
def _find_linter(linter, config_dict):
# type: (base.LinterBase, Dict[str,str]) -> Optional[base.LinterInstance]
"""
Look for a linter command with the required version.
Return a LinterInstance with the location of the linter binary if a linter binary with the
matching version is found. None otherwise.
"""
if linter.cmd_name in config_dict and config_dict[linter.cmd_name] is not None:
cmd = [config_dict[linter.cmd_name]]
# If the user specified a tool location, we do not search any further
if _check_version(linter, cmd, linter.get_lint_version_cmd_args()):
return base.LinterInstance(linter, cmd)
return None
# Search for tool
# 1. In the same directory as the interpreter
# 2. Check user base -- i.e. site.USERBASE. With "pip install --user" puts files
# 3. The current path
# 4. In '/opt/mongodbtoolchain/v4/bin' if virtualenv is set up.
python_dir = os.path.dirname(sys.executable)
if sys.platform == "win32":
# On Windows, these scripts are installed in %PYTHONDIR%\scripts like
# 'C:\python\python310\scripts', and have .exe extensions.
python_dir = os.path.join(python_dir, "scripts")
cmd_str = os.path.join(python_dir, linter.cmd_name)
cmd_str += ".exe"
cmd = [cmd_str]
else:
# On Mac and with Homebrew, check for the binaries in /usr/local instead of sys.executable.
if sys.platform == "darwin" and python_dir.startswith("/usr/local/opt"):
python_dir = "/usr/local/bin"
# On Linux, these scripts are installed in %PYTHONDIR%\bin like
# '/opt/mongodbtoolchain/v4/bin', but they may point to the wrong interpreter.
cmd_str = os.path.join(python_dir, linter.cmd_name)
# Check for executables before assuming they are python scripts
cmd = [cmd_str]
if _check_version(linter, cmd, linter.get_lint_version_cmd_args()):
return base.LinterInstance(linter, cmd)
cmd = [sys.executable] + cmd
# Check 1: interpreter location or for linters that ignore current interpreter.
if _check_version(linter, cmd, linter.get_lint_version_cmd_args()):
return base.LinterInstance(linter, cmd)
logging.info(
"First version check failed for linter '%s', trying a different location.", linter.cmd_name
)
# Check 2: Check USERBASE
cmd = [os.path.join(site.getuserbase(), "bin", linter.cmd_name)]
if _check_version(linter, cmd, linter.get_lint_version_cmd_args()):
return base.LinterInstance(linter, cmd)
# Check 3: current path
cmd = [linter.cmd_name]
if _check_version(linter, cmd, linter.get_lint_version_cmd_args()):
return base.LinterInstance(linter, cmd)
# Check 4: When a virtualenv is setup the linter modules are not installed, so we need
# to use the linters installed in '/opt/mongodbtoolchain/v4/bin'.
cmd = [sys.executable, os.path.join("/opt/mongodbtoolchain/v4/bin", linter.cmd_name)]
if _check_version(linter, cmd, linter.get_lint_version_cmd_args()):
return base.LinterInstance(linter, cmd)
return None
def find_linters(linter_list, config_dict):
# type: (List[base.LinterBase], Dict[str,str]) -> List[base.LinterInstance]
"""Find the location of all linters."""
linter_instances = [] # type: List[base.LinterInstance]
for linter in linter_list:
linter_instance = _find_linter(linter, config_dict)
if not linter_instance:
logging.error(
"""\
Could not find the correct version of linter '%s', expected '%s'. Check your
PATH environment variable or re-run with --verbose for more information.
To fix, install the needed python modules for Python 3.x:
python3 -m poetry install --no-root --sync
These commands are typically available via packages with names like python-pip,
or python3-pip. See your OS documentation for help.
""",
linter.cmd_name,
linter.required_version,
)
return None
linter_instances.append(linter_instance)
return linter_instances
class LintRunner(object):
"""Run a linter and print results in a thread safe manner."""
def __init__(self):
# type: () -> None
"""Create a Lint Runner."""
self.print_lock = threading.Lock()
def _safe_print(self, line):
# type: (str) -> None
"""
Print a line of text under a lock.
Take a lock to ensure diffs do not get mixed when printed to the screen.
"""
with self.print_lock:
print(line)
def run_lint(self, linter: base.LinterInstance, file_name: str, mongo_path: str) -> bool:
"""Run the specified linter for the file."""
linter_args = linter.linter.get_lint_cmd_args(file_name)
cmd = linter.cmd_path + linter_args
logging.debug(" ".join(cmd))
no_lint_errors = self.run(cmd)
return no_lint_errors
def run_fix(self, linter: base.LinterInstance, file_name: str, mongo_path: str) -> bool:
"""Run the specified lint fixes for the file."""
linter_args = linter.linter.get_fix_cmd_args(file_name)
cmd = linter.cmd_path + linter_args
logging.debug(" ".join(cmd))
no_lint_errors = self.run(cmd)
return no_lint_errors
def run(self, cmd):
# type: (List[str]) -> bool
"""Check the specified cmd succeeds."""
logging.debug(str(cmd))
try:
subprocess.check_output(cmd).decode("utf-8")
except subprocess.CalledProcessError as cpe:
self._safe_print("CMD [%s] failed:\n%s" % (" ".join(cmd), cpe.output.decode("utf-8")))
return False
return True
|
