Source code for schrodinger.test.pytest.reporter

"""
Schrodinger-specific display of pytest output.
"""
import gc
import glob
import itertools
import os
import signal
import subprocess
import sys
import traceback

import _pytest.runner
import psutil
import pytest

import pymmlibs

from . import _i_am_buildbot
from . import startup



[docs]def terminal_summary(terminalreporter):
    """
    Improve output of summary after tests. Needs to run before the default
    impementation.
    """
    monkeypatch_terminal_summary(terminalreporter)
    if _i_am_buildbot():
        print_all_owners(terminalreporter)
    if terminalreporter.config.option.tbstyle != "no":
        print_killed_tests(terminalreporter)







[docs]def get_core_dump_signal_codes():
    """
    Return codes for signal that may produce a core file
    """
    return (
        signal.SIGABRT,
        signal.SIGBUS,  # not defined on windows
        signal.SIGFPE,
        signal.SIGILL,
        signal.SIGSEGV,
    )







[docs]def monkeypatch_terminal_summary(terminalreporter):
    """
    Monkeypatch display of terminal summary. Intended to run first
    in a pytest_terminal_summary hook.
    """
    if _i_am_buildbot():
        # 80 is a crazy-small width, and doesn't allow clear display of test
        # names. This is the default width without a tty.
        if terminalreporter._tw.fullwidth == 80:
            terminalreporter._tw.fullwidth = 130

    # Display warnings as the first output, then prevent it from getting
    # redisplayed. Helps with buildbot emails which display last 20 lines.
    terminalreporter.summary_warnings()
    terminalreporter.summary_warnings = lambda: None

    # Monkey-patch _getfailureheadline to display test names in a way that
    # allows copy/paste of names to rerun.
    # terminalreporter can't just be subclassed because there are ordering
    # issues with the plugins

    def getfailureheadline(rep):
        if hasattr(rep, 'nodeid') and hasattr(rep, 'location'):
            return terminalreporter._locationline(rep.nodeid, *rep.location)
        else:
            return super()._getfailureheadline(rep)

    terminalreporter._getfailureheadline = getfailureheadline







[docs]def sessionfinish(session, exitstatus):
    """If there is uncollectable garbage, report about it and exit non-zero."""
    gc.collect()
    if gc.garbage:
        terminalreporter = session.config.pluginmanager.get_plugin(
            'terminalreporter')
        terminalreporter.write_sep("=", "Memory Leak!", bold=True, red=True)
        msg = "FAILURE:"
        msg += " There is a memory leak in the python modules that must be"
        msg += " fixed."
        msg += " Uncollectable garbage: ({} items):".format(len(gc.garbage))
        terminalreporter.write_line(msg)
        for item in gc.garbage:
            terminalreporter.write_line(f'  * {item}')
        if not exitstatus:
            session.exitstatus = len(gc.garbage)

    # Fault handler forgets to close stderr. Done in sessionfinish so that it
    # runs on all xdist processes.
    try:
        session.config.fault_handler_stderr.close()
    except AttributeError:
        pass




[docs]def unconfigure(config):
    """
    At the end of the py.test process, report:
    * memory use
    * all processes belonging to current user (for user buildbot)

    This is done as late as possible so that parallel py.tests from pytest-xdist
    won't interfere.
    """
    if config.option.verbose:
        terminalreporter = config.pluginmanager.get_plugin('terminalreporter')
        terminalreporter.write_sep("=", "Final Structure Count", bold=True)
        ct_total = 0
        if pymmlibs.mmct_refcount() > 0:
            ct_total = pymmlibs.mmct_get_ct_total()
        line = "      {} {}".format("Structures in use:", ct_total)
        terminalreporter.write_line(line)

        mem = psutil.Process().memory_info()
        terminalreporter.write_sep("=", "Final Memory Usage", bold=True)
        line = "{:>10} {} mb / VMS: {} mb".format("RSS:", mem.rss / 1000000,
                                                  mem.vms / 1000000)
        terminalreporter.write_line(line)




[docs]def collectreport(report):
    if not report.passed:
        _add_owners_to_report(build_hook(report), report)




[docs]def runtest_logreport(report):
    """
    The makereport hook doesn't run for crashed tests, add owners here.
    """
    if not report.passed and not hasattr(report, 'owners'):
        _add_owners_to_report(build_hook(report), report)




[docs]def build_hook(report):
    """
    If test import fails or if the test crashes, build a hook based on the
    path to the file.
    """
    path = startup.CURRENT_SESSION.rootdir.join(report.nodeid)
    return startup.CURRENT_SESSION.pluginmanager.get_plugin(
        'session').gethookproxy(path)




[docs]def format_captured_exceptions(exceptions):
    """
    Formats exceptions given as (type, value, traceback) into a string
    suitable to display as a test failure.
    """
    message = ('Uncaught exception raised to top level exception handler:\n'
               '  (Often caused by an exception raised in a PyQt slot)\n')
    message += '_' * 80 + '\n'
    for (exc_type, value, tback) in exceptions:
        message += ''.join(traceback.format_tb(tback)) + '\n'
        message += f'{exc_type.__name__}: {value}\n'
        message += '_' * 80 + '\n'
    return message




[docs]def log_exceptions(item):
    if hasattr(item, 'original_excepthook'):
        sys.excepthook = item.original_excepthook
        if item.exceptions_found:
            message = "".join(format_captured_exceptions(item.exceptions_found))
            pytest.fail(message, pytrace=False)



def _add_owners_to_report(hook, report):
    owners = hook.pytest_test_owners()
    if isinstance(owners, str):
        owners = (owners,)
    setattr(report, 'owners', owners)
    if owners:
        report.sections.append(('owners: {}'.format(', '.join(owners)), ''))



[docs]def runtest_makereport(item, call):
    """Add the test owner to each report."""
    report = _pytest.runner.pytest_runtest_makereport(item, call)
    if not report.passed:
        _add_owners_to_report(item.ihook, report)
        tmpdir = getattr(item, 'schro_tmp_cwd', None)
        if tmpdir:
            report.sections.append(('Execution directory', tmpdir))
    return report