"""
Schrodinger-specific display of pytest output.
"""
import gc
import itertools
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 print_killed_tests(terminalreporter):
    """
    Print messages about c++ killed tests.
    """
    killed_reports = terminalreporter.getreports('killed')
    if not killed_reports:
        return
    terminalreporter.write_sep("=", "KILLED TESTS")
    for rep in killed_reports:
        if terminalreporter.config.option.tbstyle == "line":
            line = terminalreporter._getcrashline(rep)
            terminalreporter.write_line(line)
        else:
            msg = terminalreporter._getfailureheadline(rep)
            markup = {'red': True, 'bold': True}
            terminalreporter.write_sep("_", msg, **markup)
            terminalreporter._outrep_summary(rep) 
[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 print_all_owners(terminalreporter):
    """
    Print all owners of all killed or failed tests.
    """
    killed_reports = terminalreporter.getreports('killed')
    failed_reports = terminalreporter.getreports('failed')
    error_reports = terminalreporter.getreports('error')
    get_owners = lambda report: getattr(report, 'owners', tuple())
    bad_reports = itertools.chain(failed_reports, killed_reports, error_reports)
    owners = set(
        itertools.chain.from_iterable(
            [_f for _f in map(get_owners, bad_reports) if _f]))
    if owners:
        terminalreporter.write_sep("=", "Owners of failed tests", red=True)
        terminalreporter.write_line("Owners: " + ' '.join(owners)) 
[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 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 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