Coverage for C:\Repos\ekr-pylint\pylint\lint\run.py: 24%

106 statements  

« prev     ^ index     » next       coverage.py v6.4, created at 2022-05-24 10:21 -0500

1# Licensed under the GPL: https://www.gnu.org/licenses/old-licenses/gpl-2.0.html 

2# For details: https://github.com/PyCQA/pylint/blob/main/LICENSE 

3# Copyright (c) https://github.com/PyCQA/pylint/blob/main/CONTRIBUTORS.txt 

4 

5from __future__ import annotations 

6 

7import os 

8import sys 

9import warnings 

10from collections.abc import Sequence 

11from pathlib import Path 

12from typing import Any 

13 

14from pylint import config 

15from pylint.config.config_initialization import _config_initialization 

16from pylint.config.exceptions import ArgumentPreprocessingError 

17from pylint.config.utils import _preprocess_options 

18from pylint.constants import full_version 

19from pylint.lint.base_options import _make_run_options 

20from pylint.lint.pylinter import PyLinter 

21from pylint.reporters.base_reporter import BaseReporter 

22 

23try: 

24 import multiprocessing 

25 from multiprocessing import synchronize # noqa pylint: disable=unused-import 

26except ImportError: 

27 multiprocessing = None # type: ignore[assignment] 

28 

29 

30def _query_cpu() -> int | None: 

31 """Try to determine number of CPUs allotted in a docker container. 

32 

33 This is based on discussion and copied from suggestions in 

34 https://bugs.python.org/issue36054. 

35 """ 

36 cpu_quota, avail_cpu = None, None 

37 

38 if Path("/sys/fs/cgroup/cpu/cpu.cfs_quota_us").is_file(): 

39 with open("/sys/fs/cgroup/cpu/cpu.cfs_quota_us", encoding="utf-8") as file: 

40 # Not useful for AWS Batch based jobs as result is -1, but works on local linux systems 

41 cpu_quota = int(file.read().rstrip()) 

42 

43 if ( 

44 cpu_quota 

45 and cpu_quota != -1 

46 and Path("/sys/fs/cgroup/cpu/cpu.cfs_period_us").is_file() 

47 ): 

48 with open("/sys/fs/cgroup/cpu/cpu.cfs_period_us", encoding="utf-8") as file: 

49 cpu_period = int(file.read().rstrip()) 

50 # Divide quota by period and you should get num of allotted CPU to the container, rounded down if fractional. 

51 avail_cpu = int(cpu_quota / cpu_period) 

52 elif Path("/sys/fs/cgroup/cpu/cpu.shares").is_file(): 

53 with open("/sys/fs/cgroup/cpu/cpu.shares", encoding="utf-8") as file: 

54 cpu_shares = int(file.read().rstrip()) 

55 # For AWS, gives correct value * 1024. 

56 avail_cpu = int(cpu_shares / 1024) 

57 return avail_cpu 

58 

59 

60def _cpu_count() -> int: 

61 """Use sched_affinity if available for virtualized or containerized environments.""" 

62 cpu_share = _query_cpu() 

63 cpu_count = None 

64 sched_getaffinity = getattr(os, "sched_getaffinity", None) 

65 # pylint: disable=not-callable,using-constant-test,useless-suppression 

66 if sched_getaffinity: 

67 cpu_count = len(sched_getaffinity(0)) 

68 elif multiprocessing: 

69 cpu_count = multiprocessing.cpu_count() 

70 else: 

71 cpu_count = 1 

72 if cpu_share is not None: 

73 return min(cpu_share, cpu_count) 

74 return cpu_count 

75 

76 

77UNUSED_PARAM_SENTINEL = object() 

78 

79 

80class Run: 

81 """Helper class to use as main for pylint with 'run(*sys.argv[1:])'.""" 

82 

83 LinterClass = PyLinter 

84 option_groups = ( 

85 ( 

86 "Commands", 

87 "Options which are actually commands. Options in this \ 

88group are mutually exclusive.", 

89 ), 

90 ) 

91 

92 def __init__( 

93 self, 

94 args: Sequence[str], 

95 reporter: BaseReporter | None = None, 

96 exit: bool = True, # pylint: disable=redefined-builtin 

97 do_exit: Any = UNUSED_PARAM_SENTINEL, 

98 ) -> None: 

99 # Immediately exit if user asks for version 

100 if "--version" in args: 

101 print(full_version) 

102 sys.exit(0) 

103 

104 self._rcfile: str | None = None 

105 self._output: str | None = None 

106 self._plugins: list[str] = [] 

107 self.verbose: bool = False 

108 

109 # Pre-process certain options and remove them from args list 

110 try: 

111 args = _preprocess_options(self, args) 

112 except ArgumentPreprocessingError as ex: 

113 print(ex, file=sys.stderr) 

114 sys.exit(32) 

115 

116 # Determine configuration file 

117 if self._rcfile is None: 

118 default_file = next(config.find_default_config_files(), None) 

119 if default_file: 

120 self._rcfile = str(default_file) 

121 

122 self.linter = linter = self.LinterClass( 

123 _make_run_options(self), 

124 option_groups=self.option_groups, 

125 pylintrc=self._rcfile, 

126 ) 

127 # register standard checkers 

128 linter.load_default_plugins() 

129 # load command line plugins 

130 linter.load_plugin_modules(self._plugins) 

131 

132 linter.disable("I") 

133 linter.enable("c-extension-no-member") 

134 

135 args = _config_initialization( 

136 linter, args, reporter, config_file=self._rcfile, verbose_mode=self.verbose 

137 ) 

138 

139 if linter.config.jobs < 0: 

140 print( 

141 f"Jobs number ({linter.config.jobs}) should be greater than or equal to 0", 

142 file=sys.stderr, 

143 ) 

144 sys.exit(32) 

145 if linter.config.jobs > 1 or linter.config.jobs == 0: 

146 if multiprocessing is None: 

147 print( 

148 "Multiprocessing library is missing, fallback to single process", 

149 file=sys.stderr, 

150 ) 

151 linter.set_option("jobs", 1) 

152 elif linter.config.jobs == 0: 

153 linter.config.jobs = _cpu_count() 

154 

155 if self._output: 

156 try: 

157 with open(self._output, "w", encoding="utf-8") as output: 

158 linter.reporter.out = output 

159 linter.check(args) 

160 score_value = linter.generate_reports() 

161 except OSError as ex: 

162 print(ex, file=sys.stderr) 

163 sys.exit(32) 

164 else: 

165 linter.check(args) 

166 score_value = linter.generate_reports() 

167 

168 if do_exit is not UNUSED_PARAM_SENTINEL: 

169 warnings.warn( 

170 "do_exit is deprecated and it is going to be removed in a future version.", 

171 DeprecationWarning, 

172 ) 

173 exit = do_exit 

174 

175 if exit: 

176 if linter.config.exit_zero: 

177 sys.exit(0) 

178 elif linter.any_fail_on_issues(): 

179 # We need to make sure we return a failing exit code in this case. 

180 # So we use self.linter.msg_status if that is non-zero, otherwise we just return 1. 

181 sys.exit(self.linter.msg_status or 1) 

182 elif score_value is not None: 

183 if score_value >= linter.config.fail_under: 

184 sys.exit(0) 

185 else: 

186 # We need to make sure we return a failing exit code in this case. 

187 # So we use self.linter.msg_status if that is non-zero, otherwise we just return 1. 

188 sys.exit(self.linter.msg_status or 1) 

189 else: 

190 sys.exit(self.linter.msg_status)