Source code for CIME.case.case_test

"""
Run a testcase.
case_test is a member of class Case from case.py
"""

from CIME.XML.standard_module_setup import *
from CIME.utils import expect, find_system_test, find_proc_id
from CIME.SystemTests.system_tests_common import *
from CIME.status import append_testlog

import sys, signal


def _iter_signal_names():
    for signame in [
        item
        for item in dir(signal)
        if item.startswith("SIG") and not item.startswith("SIG_")
    ]:
        yield signame


def _signal_handler(signum, _):
    name = "Unknown"
    for signame in _iter_signal_names():
        if signum == getattr(signal, signame):
            name = signame

    # Terminate children
    proc_ids = find_proc_id(children_only=True)
    for proc_id in proc_ids:
        try:
            os.kill(proc_id, signal.SIGKILL)
        except OSError:
            # If the batch system killed the entire process group, these
            # processes might already be dying
            pass

    # Throw an exception so SystemTest infrastructure can handle this error
    expect(False, "Job killed due to receiving signal {:d} ({})".format(signum, name))


def _set_up_signal_handlers():
    """
    Add handles for all signals that might be used to abort a test

    We need to handle a wide variety due to different implementations of the
    timeout mechanism for different batch systems.
    """
    for signame in ["SIGINT", "SIGTERM", "SIGXCPU", "SIGUSR1", "SIGUSR2"]:
        signum = getattr(signal, signame)
        signal.signal(signum, _signal_handler)


[docs] def case_test(self, testname=None, reset=False, skip_pnl=False): if testname is None: testname = self.get_value("TESTCASE") expect(testname is not None, "testname argument not resolved") logging.warning("Running test for {}".format(testname)) _set_up_signal_handlers() try: # The following line can throw exceptions if the testname is # not found or the test constructor throws. We need to be # sure to leave TestStatus in the appropriate state if that # happens. test = find_system_test(testname, self)(self) except BaseException: caseroot = self.get_value("CASEROOT") with TestStatus(test_dir=caseroot) as ts: ts.set_status(RUN_PHASE, TEST_FAIL_STATUS, comments="failed to initialize") append_testlog(str(sys.exc_info()[1])) raise if reset: logger.info("Reset test to initial conditions and exit") # pylint: disable=protected-access test._resetup_case(RUN_PHASE) return True success = test.run(skip_pnl=skip_pnl) return success