<s> from __future__ import ( absolute_import , division , generators , nested_scopes , print_function , <EOL> unicode_literals , with_statement ) <EOL> import re <EOL> from collections import namedtuple <EOL> from pants . backend . python . targets . python_target import PythonTarget <EOL> from pants . backend . python . tasks . python_task import PythonTask <EOL> from pants . base . build_environment import get_buildroot <EOL> from pants . base . exceptions import TaskError <EOL> from pants . option . custom_types import file_option <EOL> from pants . contrib . python . checks . tasks . checkstyle . common import CheckSyntaxError , Nit , PythonFile <EOL> from pants . contrib . python . checks . tasks . checkstyle . file_excluder import FileExcluder <EOL> from pants . contrib . python . checks . tasks . checkstyle . register_plugins import register_plugins <EOL> _NOQA_LINE_SEARCH = re . compile ( r'<STR_LIT>' ) . search <EOL> _NOQA_FILE_SEARCH = re . compile ( r'<STR_LIT>' ) . search <EOL> class LintPlugin ( namedtuple ( '<STR_LIT>' , [ '<STR_LIT:name>' , '<STR_LIT>' ] ) ) : <EOL> def skip ( self ) : <EOL> return self . subsystem . global_instance ( ) . get_options ( ) . skip <EOL> def checker ( self , python_file ) : <EOL> return self . subsystem . global_instance ( ) . get_plugin ( python_file ) <EOL> def line_contains_noqa ( line ) : <EOL> return _NOQA_LINE_SEARCH ( line ) is not None <EOL> def noqa_file_filter ( python_file ) : <EOL> return any ( _NOQA_FILE_SEARCH ( line ) is not None for line in python_file . lines ) <EOL> class PythonCheckStyleTask ( PythonTask ) : <EOL> _PYTHON_SOURCE_EXTENSION = '<STR_LIT>' <EOL> _plugins = [ ] <EOL> _subsystems = tuple ( ) <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( PythonCheckStyleTask , self ) . __init__ ( * args , ** kwargs ) <EOL> self . _plugins = [ plugin for plugin in self . _plugins if not plugin . skip ( ) ] <EOL> self . options = self . get_options ( ) <EOL> self . excluder = FileExcluder ( self . options . suppress , self . context . log ) <EOL> @ classmethod <EOL> def global_subsystems ( cls ) : <EOL> return super ( PythonTask , cls ) . global_subsystems ( ) + cls . _subsystems <EOL> @ classmethod <EOL> def register_options ( cls , register ) : <EOL> super ( PythonCheckStyleTask , cls ) . register_options ( register ) <EOL> register ( '<STR_LIT>' , fingerprint = True , default = '<STR_LIT>' , type = str , <EOL> help = '<STR_LIT>' ) <EOL> register ( '<STR_LIT>' , fingerprint = True , type = bool , <EOL> help = '<STR_LIT>' ) <EOL> register ( '<STR_LIT>' , type = bool , <EOL> help = '<STR_LIT>' ) <EOL> register ( '<STR_LIT>' , fingerprint = True , type = file_option , default = None , <EOL> help = '<STR_LIT>' ) <EOL> register ( '<STR_LIT>' , fingerprint = True , default = True , type = bool , <EOL> help = '<STR_LIT>' ) <EOL> @ classmethod <EOL> def supports_passthru_args ( cls ) : <EOL> return True <EOL> def _is_checked ( self , target ) : <EOL> return isinstance ( target , PythonTarget ) and target . has_sources ( self . _PYTHON_SOURCE_EXTENSION ) <EOL> @ classmethod <EOL> def clear_plugins ( cls ) : <EOL> """<STR_LIT>""" <EOL> cls . _plugins = [ ] <EOL> @ classmethod <EOL> def register_plugin ( cls , name , subsystem ) : <EOL> """<STR_LIT>""" <EOL> plugin = LintPlugin ( name = name , subsystem = subsystem ) <EOL> cls . _plugins . append ( plugin ) <EOL> cls . _subsystems += ( plugin . subsystem , ) <EOL> def get_nits ( self , filename ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> python_file = PythonFile . parse ( filename , root = get_buildroot ( ) ) <EOL> except CheckSyntaxError as e : <EOL> yield e . as_nit ( ) <EOL> return <EOL> if noqa_file_filter ( python_file ) : <EOL> return <EOL> if self . options . suppress : <EOL> check_plugins = [ plugin for plugin in self . _plugins <EOL> if self . excluder . should_include ( filename , plugin . name ) ] <EOL> else : <EOL> check_plugins = self . _plugins <EOL> for plugin in check_plugins : <EOL> for i , nit in enumerate ( plugin . checker ( python_file ) ) : <EOL> if i == <NUM_LIT:0> : <EOL> self . context . log . debug ( '<STR_LIT>' . format ( plugin . name , filename ) ) <EOL> if not nit . has_lines_to_display : <EOL> yield nit <EOL> continue <EOL> if all ( not line_contains_noqa ( line ) for line in nit . lines ) : <EOL> yield nit <EOL> def check_file ( self , filename ) : <EOL> """<STR_LIT>""" <EOL> log_threshold = Nit . SEVERITY . get ( self . options . severity , Nit . COMMENT ) <EOL> failure_count = <NUM_LIT:0> <EOL> fail_threshold = Nit . WARNING if self . options . strict else Nit . ERROR <EOL> for i , nit in enumerate ( self . get_nits ( filename ) ) : <EOL> if i == <NUM_LIT:0> : <EOL> print ( ) <EOL> if nit . severity >= log_threshold : <EOL> print ( '<STR_LIT>' . format ( nit = nit ) ) <EOL> if nit . severity >= fail_threshold : <EOL> failure_count += <NUM_LIT:1> <EOL> return failure_count <EOL> def checkstyle ( self , sources ) : <EOL> """<STR_LIT>""" <EOL> failure_count = <NUM_LIT:0> <EOL> for filename in sources : <EOL> failure_count += self . check_file ( filename ) <EOL> if failure_count > <NUM_LIT:0> and self . options . fail : <EOL> raise TaskError ( '<STR_LIT>' . format ( failure_count ) , exit_code = <NUM_LIT:1> ) <EOL> return failure_count <EOL> def execute ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . options . skip : <EOL> return <EOL> with self . invalidated ( self . context . targets ( self . _is_checked ) ) as invalidation_check : <EOL> sources = self . calculate_sources ( [ vt . target for vt in invalidation_check . invalid_vts ] ) <EOL> if sources : <EOL> return self . checkstyle ( sources ) <EOL> def calculate_sources ( self , targets ) : <EOL> """<STR_LIT>""" <EOL> sources = set ( ) <EOL> for target in targets : <EOL> sources . update ( <EOL> source for source in target . sources_relative_to_buildroot ( ) <EOL> if source . endswith ( self . _PYTHON_SOURCE_EXTENSION ) <EOL> ) <EOL> return sources <EOL> register_plugins ( PythonCheckStyleTask ) </s>
