"""
Runs workups.
Requires as input a `testscript.TestScript` object referring to a completed job.
Uses the outcome status of the TestScript object as well as the workup protocol
stored within the TestScript object to perform a workup and get a Pass/Fail
(True/False) result for the test.
Actual workup criteria are in outcomes.
@copyright: Schrodinger, Inc. All rights reserved.
"""
import glob
import importlib
import inspect
import os
import time
import traceback
from schrodinger.job.jobcontrol import timestamp_format
from schrodinger.job.queue import JobControlJob
from schrodinger.utils import fileutils
from schrodinger.utils import log
from . import common
from . import constants
from . import outcomes
from . import sysinfo
from .workups import workup_tools
logger = common.logger
TIMEOUT_KILLED_TEXT = "Failed due to timeout - killed by STU after %ds."
[docs]def workup_outcome(test, job_dir, registered_workups=None, job_dj_job=None):
"""
Runs the outcome_workup of a given script.
:param test: The script to be tested.
:type test: `testscripts.TestScript`
:param job_dir: The directory in which the job was executed
:type job_dir: str
:param registered_workups: Maps all valid workup function names to wrapped
workups.
:type registered_workups: dict | None
:param job_dj_job: Job that was run by the test, if any.
:type job_dj_job: job.queue.BaseJob | None
:return: Does the test pass or fail?
:rtype: bool
"""
test_id = test.id
if not test_id:
test_id = os.path.split(job_dir)[-1]
if job_dj_job and isinstance(job_dj_job, JobControlJob):
job = job_dj_job.getJob()
else:
job = None
try:
check_exit_status(test, job_dj_job)
with fileutils.chdir(job_dir):
check_workup(test,
test_id,
registered_workups=registered_workups,
job_dj_job=job_dj_job)
outcomes.job_status.license_check()
except (AssertionError, outcomes.failures.WorkupFailure) as err:
test.workup_messages = str(err)
test.failure_type = getattr(
err, 'failure_type', outcomes.failures.WorkupFailure.failure_type)
test.outcome = False
except (Exception, SystemExit) as err:
test.workup_messages = "Failure in workup:"
tb = traceback.format_exc()
if tb:
test.workup_messages += '\n' + tb
else:
test.workup_messages += ' ' + str(err)
test.failure_type = outcomes.failures.WorkupFailure.failure_type
test.outcome = False
else:
test.outcome = True
if test.workup_messages:
logger.error(test.workup_messages.rstrip('\n') + '\n')
[docs]def check_exit_status(test, job=None):
"""
Check that the exit status of the test matched expected. Either a good exit
status or a bad one can be expected, you can't enforce that an exit status
be "incorporated", for instance.
:param test: Test object
:type test: schrodinger.test.stu.testscripts.TestScript
:param job: job representation
:type job: schrodinger.job.queue.BaseJob
"""
job_succeeded = constants.JC_outcome_codes.get(test.exit_status, False)
msg = constants.BAD_STATUS_TEXT % test.exit_status
if job_succeeded and test.expect_job_failure:
# If failure is expected and the job actually succeeds, this is bad.
raise outcomes.failures.JobExpectedFailure(msg)
elif not job_succeeded:
if job and job.canceled_by_timeout:
msg = get_job_killed_message(test, job)
raise outcomes.failures.JobKilledFailure(msg)
elif test.expect_job_failure and test.exit_status not in (
constants.FAILED_TO_LAUNCH, constants.JOB_NOT_STARTED):
# With expect_job_failure, we expect the job to launch but not succeed.
pass
elif test.exit_status == 'fizzled':
raise outcomes.failures.JobFizzledFailure(msg)
elif test.exit_status == 'Failed to launch':
raise outcomes.failures.JobLaunchFailure(msg)
else:
raise outcomes.failures.JobDiedFailure(msg)
[docs]def discover_workups():
"""
Registers all workups in outcomes and outcomes.custom.*
"""
# empty namespace to act like a pseudo module
class _blank:
pass
w = {}
for name, item in outcomes.__dict__.items():
if inspect.isfunction(item):
w[name] = workup_tools.workup(item)
modules = glob.glob(os.path.dirname(outcomes.__file__) + "/custom/*.py")
custom = _blank()
for module_name in modules:
module_name = os.path.basename(module_name)[:-3]
module_path = 'schrodinger.test.stu.outcomes.custom.' + module_name
module = importlib.import_module(module_path)
mod = _blank()
for name, item in module.__dict__.items():
if inspect.isfunction(item):
if getattr(item, "wrapped", None):
setattr(mod, name, item)
else:
setattr(mod, name, workup_tools.workup(item))
setattr(custom, module_name, mod)
w['custom'] = custom
return w
[docs]def check_workup(test, test_id, registered_workups=None, job_dj_job=None):
"""
Evaluates the workup for a given test object. If the workup function fails,
raise a WorkupFailure. Otherwise, return None.
:param test: STU test object
:type test: schrodinger.test.stu.testscripts.TestScript
:param test_id: test ID for the STU test.
:type test_id: int
:raises outcomes.failures.WorkupImportFailure: if workupstr is not a valid function
:raises outcomes.failures.WorkupFailure: if the workup evaluates to False,
per the parameters of the workup
"""
workupstr = test.workupstr
if not workupstr:
return
if registered_workups is None:
local_vars = discover_workups()
else:
local_vars = registered_workups.copy()
try:
workup_tools.messages = []
local_vars['test'] = test
local_vars['job_dj_job'] = job_dj_job
try:
# This is necessary to access the job inside of workup functions
__builtins__['job_dj_job'] = job_dj_job
outcome = eval(workupstr, globals(), local_vars)
finally:
del __builtins__['job_dj_job']
# A NameError will occur when the user tries to use a workup function that
# is not defined.
except (NameError, ImportError) as e:
raise outcomes.failures.WorkupImportFailure(
f'Workup: {workupstr} from {os.getcwd()} caused the following error: {e}'
)
# if outcome is false (test did not pass)
if not outcome:
if workup_tools.messages:
failure_message = "\n\n".join(
workup_tools.messages +
[f'Workup: {workupstr} failed in {os.getcwd()}'])
workup_tools.messages = []
else:
failure_message = f'Workup: {workupstr} failed in {os.getcwd()}'
raise outcomes.failures.WorkupFailure(failure_message)
return
[docs]def print_summary(scripts):
"""
Prints the summary information at end of a execution.
:param scripts: a dictionary of TestScript objects. Each holds its own
results information.
:type scripts: dict
:return: Did all tests pass?
"""
if not scripts:
logger.setLevel(log.DEBUG)
logger.debug('No tests ran. Skipping run summary.')
return False
format = "%-7s %-16s %-14s %-12s %-8s %6s "
separator = '=' * len(format % ('', '', '', '', '', ''))
logger.info("\n" + separator)
logger.info("Execution Summary:\n")
#Print system information
if (sysinfo.REMOTE.host == sysinfo.LOCAL.host):
logger.info(sysinfo.TABLE_FORMAT("host:", sysinfo.LOCAL.name))
logger.info(sysinfo.LOCAL)
else:
logger.info(sysinfo.TABLE_FORMAT("local host:", sysinfo.LOCAL.name))
logger.info(sysinfo.LOCAL)
logger.info(sysinfo.TABLE_FORMAT("remote host:", sysinfo.REMOTE.host))
logger.info(sysinfo.REMOTE)
logger.warning(separator)
logger.warning(
format %
("Test ID", "Product", "Platform", "Exit Status", "Workup", "Timing"))
logger.warning(separator)
success = True
"""Did all tests pass?"""
#Print workup information for each script.
format = "%-7s %-16s %-14s %-12s %-8s %6.1d "
for script_id in sort_mixed_list(scripts):
product = scripts[script_id].product
if len(product) > 15:
product = product[:12] + "..."
if scripts[script_id].outcome is None:
poutcome = "Not Run"
elif scripts[script_id].outcome:
poutcome = "Success"
else:
poutcome = "Failure"
success = False
#platform should be based on the actual execution platform!
if scripts[script_id].useJC():
system = sysinfo.REMOTE
else:
system = sysinfo.LOCAL
#to protect against problems converting timing to a string.
timing = scripts[script_id].timing
if not timing:
timing = 0
##HERE IS THE PRINT STATEMENT
logger.warning(format %
(script_id, product, system.platform,
scripts[script_id].exit_status, poutcome, timing))
return success
[docs]def sort_mixed_list(items):
"""
Sort mixed list of ints and strs. Necessary for sorting lists of test ids
as these may be either kind of data.
"""
sorting_key = lambda x: str(x) if any(not isinstance(i, int)
for i in items) else x
return sorted(items, key=sorting_key)
[docs]def get_job_killed_message(test, job):
"""
Return a suitable message for a job killed by STU.
:param test: a STU Test
:type test: schrodinger.test.stu.testscripts.TestScript
:param job: job representation if JobControlJob
:type job: schrodinger.job.jobcontrol.Job or None
:rtype: str
"""
if test.timing <= 0:
return "Failed due to timeout - killed by STU while still on queue."
elif not isinstance(job, JobControlJob) or not job.getJob().QueueHost:
return TIMEOUT_KILLED_TEXT % job.duration
if not job.getJob().StartTime:
return (f"Job had timing of {test.timing=} but job has no StartTime. "
f"Please report this to BLDMGR\n{job.getJob().summary()}")
run_time = job.duration
total_time = int(
time.mktime(time.strptime(job.getJob().StopTime, timestamp_format)) -
time.mktime(time.strptime(job.getJob().LaunchTime, timestamp_format)))
queued_time = total_time - run_time
return f"Failed due to timeout - {queued_time}s on queue, {run_time}s running"