import CIME.compare_namelists, CIME.simple_compare
from CIME.utils import append_status, EnvironmentContext
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
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()
if (compare_tests in [[], None] or CIME.utils.match_any(test_name, compare_tests)):
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):
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
if nl_do_compare or do_compare:
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:
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)
return all_pass_or_skip