Source code for CIME.code_checker

"""
Libraries for checking python code with pylint
"""

from CIME.XML.standard_module_setup import *

from CIME.utils import run_cmd, run_cmd_no_fail, expect, get_cime_root, is_python_executable, get_cime_default_driver

from multiprocessing.dummy import Pool as ThreadPool
#pylint: disable=import-error
from distutils.spawn import find_executable

logger = logging.getLogger(__name__)

###############################################################################
def _run_pylint(on_file, interactive):
###############################################################################
    pylint = find_executable("pylint")

    cmd_options = " --disable=I,C,R,logging-not-lazy,wildcard-import,unused-wildcard-import"
    cmd_options += ",fixme,broad-except,bare-except,eval-used,exec-used,global-statement"
    cmd_options += ",logging-format-interpolation,no-name-in-module"
    cimeroot = get_cime_root()

    if "scripts/Tools" in on_file:
        cmd_options +=",relative-import"

    # add init-hook option
    cmd_options += " --init-hook='sys.path.extend((\"%s\",\"%s\",\"%s\",\"%s\"))'"%\
        (os.path.join(cimeroot,"scripts","lib"),
         os.path.join(cimeroot,"scripts","Tools"),
         os.path.join(cimeroot,"scripts","fortran_unit_testing","python"),
         os.path.join(cimeroot,"src","drivers","nuopc","cime_config","runseq"))

    cmd = "%s %s %s" % (pylint, cmd_options, on_file)
    logger.debug("pylint command is %s"%cmd)
    stat, out, err = run_cmd(cmd, verbose=False, from_dir=cimeroot)
    if stat != 0:
        if interactive:
            logger.info("File %s has pylint problems, please fix\n    Use command: %s" % (on_file, cmd))
            logger.info(out + "\n" + err)
        return (on_file, out + "\n" + err)
    else:
        if interactive:
            logger.info("File %s has no pylint problems" % on_file)
        return (on_file, "")

###############################################################################
def _matches(file_path, file_ends):
###############################################################################
    for file_end in file_ends:
        if file_path.endswith(file_end):
            return True

    return False

###############################################################################
def _should_pylint_skip(filepath):
###############################################################################
    # TODO - get rid of this
    list_of_directories_to_ignore = ("xmlconvertors", "pointclm", "point_clm", "tools", "machines", "apidocs", "doc")
    for dir_to_skip in list_of_directories_to_ignore:
        if dir_to_skip + "/" in filepath:
            return True
        if filepath == "scripts/lib/six.py":
            return True
        # intended to be temporary, file needs update
        if filepath.endswith("archive_metadata") or filepath.endswith("pgn.py"):
            return True

    return False

###############################################################################
[docs]def get_all_checkable_files(): ############################################################################### cimeroot = get_cime_root() all_git_files = run_cmd_no_fail("git ls-files", from_dir=cimeroot, verbose=False).splitlines() if get_cime_default_driver() == "nuopc": nuopc_git_files = [] try: nuopc_git_files = run_cmd_no_fail("git ls-files", from_dir=os.path.join(cimeroot,"src","drivers","nuopc"), verbose=False).splitlines() except: logger.warning("No nuopc driver found in source") all_git_files.extend([os.path.join("src","drivers","nuopc",_file) for _file in nuopc_git_files]) files_to_test = [item for item in all_git_files if ((item.endswith(".py") or is_python_executable(os.path.join(cimeroot, item))) and not _should_pylint_skip(item))] return files_to_test
###############################################################################
[docs]def check_code(files, num_procs=10, interactive=False): ############################################################################### """ Check all python files in the given directory Returns True if all files had no problems """ # Get list of files to check, we look to see if user-provided file argument # is a valid file, if not, we search the repo for a file with similar name. files_to_check = [] if files: repo_files = get_all_checkable_files() for filearg in files: if os.path.exists(filearg): files_to_check.append(os.path.abspath(filearg)) else: found = False for repo_file in repo_files: if repo_file.endswith(filearg): found = True files_to_check.append(repo_file) # could have multiple matches if not found: logger.warning("Could not find file matching argument '%s'" % filearg) else: # Check every python file files_to_check = get_all_checkable_files() if "scripts/lib/six.py" in files_to_check: files_to_check.remove("scripts/lib/six.py") logger.info("Not checking contributed file six.py") expect(len(files_to_check) > 0, "No matching files found") # No point in using more threads than files if len(files_to_check) < num_procs: num_procs = len(files_to_check) pool = ThreadPool(num_procs) results = pool.map(lambda x : _run_pylint(x, interactive), files_to_check) pool.close() pool.join() return dict(results)