Source code for CIME.compare_test_results

import CIME.compare_namelists, CIME.simple_compare
from CIME.status import append_status
from CIME.utils import EnvironmentContext, parse_test_name
from CIME.test_status import *
from CIME.hist_utils import compare_baseline, get_ts_synopsis
from CIME.case import Case
from CIME.test_utils import get_test_status_files

import os, logging

###############################################################################
[docs] def append_status_cprnc_log(msg, logfile_name, test_dir): ############################################################################### try: append_status(msg, logfile_name, caseroot=test_dir) except IOError: pass
###############################################################################
[docs] def compare_namelists(case, baseline_name, baseline_root, logfile_name): ############################################################################### log_lvl = logging.getLogger().getEffectiveLevel() logging.disable(logging.CRITICAL) success = case.case_cmpgen_namelists( compare=True, compare_name=baseline_name, baseline_root=baseline_root, logfile_name=logfile_name, ) logging.getLogger().setLevel(log_lvl) return success
###############################################################################
[docs] def compare_history(case, baseline_name, baseline_root, log_id): ############################################################################### real_user = case.get_value("REALUSER") with EnvironmentContext(USER=real_user): baseline_full_dir = os.path.join( baseline_root, baseline_name, case.get_value("CASEBASEID") ) outfile_suffix = "{}.{}".format(baseline_name, log_id) try: result, comments = compare_baseline( case, baseline_dir=baseline_full_dir, outfile_suffix=outfile_suffix ) except IOError: result, comments = compare_baseline( case, baseline_dir=baseline_full_dir, outfile_suffix=None ) return result, comments
###############################################################################
[docs] def compare_test_results( baseline_name, baseline_root, test_root, compiler, test_id=None, compare_tests=None, namelists_only=False, hist_only=False, ): ############################################################################### """ Compares with baselines for all matching tests Outputs results for each test to stdout (one line per test); possible status codes are: PASS, FAIL, SKIP. (A SKIP denotes a test that did not make it to the run phase or a test for which the run phase did not pass: we skip baseline comparisons in this case.) In addition, creates files named compare.log.BASELINE_NAME.TIMESTAMP in each test directory, which contain more detailed output. Also creates *.cprnc.out.BASELINE_NAME.TIMESTAMP files in each run directory. Returns True if all tests generated either PASS or SKIP results, False if there was at least one FAIL result. """ test_status_files = get_test_status_files(test_root, compiler, test_id=test_id) # ID to use in the log file names, to avoid file name collisions with # earlier files that may exist. log_id = CIME.utils.get_timestamp() all_pass_or_skip = True compare_tests_counts = None if compare_tests: compare_tests_counts = dict( [(compare_test, 0) for compare_test in compare_tests] ) for test_status_file in test_status_files: test_dir = os.path.dirname(test_status_file) ts = TestStatus(test_dir=test_dir) test_name = ts.get_name() testopts = parse_test_name(test_name)[1] testopts = [] if testopts is None else testopts build_only = "B" in testopts if not compare_tests or CIME.utils.match_any(test_name, compare_tests_counts): if not hist_only: nl_compare_result = None nl_compare_comment = "" nl_result = ts.get_status(SETUP_PHASE) if nl_result is None: nl_compare_result = "SKIP" nl_compare_comment = "Test did not make it to setup phase" nl_do_compare = False else: nl_do_compare = True else: nl_do_compare = False detailed_comments = "" if not namelists_only and not build_only: compare_result = None compare_comment = "" run_result = ts.get_status(RUN_PHASE) if run_result is None: compare_result = "SKIP" compare_comment = "Test did not make it to run phase" do_compare = False elif run_result != TEST_PASS_STATUS: compare_result = "SKIP" compare_comment = "Run phase did not pass" do_compare = False else: do_compare = True else: do_compare = False with Case(test_dir) as case: if baseline_name is None: baseline_name = case.get_value("BASELINE_NAME_CMP") if not baseline_name: baseline_name = CIME.utils.get_current_branch( repo=CIME.utils.get_cime_root() ) if baseline_root is None: baseline_root = case.get_value("BASELINE_ROOT") logfile_name = "compare.log.{}.{}".format( baseline_name.replace("/", "_"), log_id ) append_status_cprnc_log( "Comparing against baseline with compare_test_results:\n" "Baseline: {}\n In baseline_root: {}".format( baseline_name, baseline_root ), logfile_name, test_dir, ) if nl_do_compare or do_compare: if nl_do_compare: nl_success = compare_namelists( case, baseline_name, baseline_root, logfile_name ) if nl_success: nl_compare_result = TEST_PASS_STATUS nl_compare_comment = "" else: nl_compare_result = TEST_FAIL_STATUS nl_compare_comment = "See {}/{}".format(test_dir, logfile_name) all_pass_or_skip = False if do_compare: success, detailed_comments = compare_history( case, baseline_name, baseline_root, log_id ) if success: compare_result = TEST_PASS_STATUS else: compare_result = TEST_FAIL_STATUS all_pass_or_skip = False compare_comment = get_ts_synopsis(detailed_comments) brief_result = "" if not hist_only: brief_result += "{} {} {} {}\n".format( nl_compare_result, test_name, NAMELIST_PHASE, nl_compare_comment ) if not namelists_only: brief_result += "{} {} {}".format( compare_result, test_name, BASELINE_PHASE ) if compare_comment: brief_result += " {}".format(compare_comment) brief_result += "\n" print(brief_result) append_status_cprnc_log(brief_result, logfile_name, test_dir) if detailed_comments: append_status_cprnc_log( "Detailed comments:\n" + detailed_comments, logfile_name, test_dir ) # Emit a warning if items in compare_tests did not match anything if compare_tests: for compare_test, compare_count in compare_tests_counts.items(): if compare_count == 0: logging.warning( """ compare test arg '{}' did not match any tests in test_root {} with compiler {} and test_id {}. It's possible that one of these arguments had a mistake (likely compiler or testid).""".format( compare_test, test_root, compiler, test_id ) ) return all_pass_or_skip