eric7/Testing/Interfaces/PytestRunner.py

# -*- coding: utf-8 -*-

# Copyright (c) 2022 Detlev Offenbach <detlev@die-offenbachs.de>
#

"""
Module implementing the test runner script for the 'pytest' framework.
"""

import json
import os
import sys
import time

sys.path.insert(
    2,
    os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
)


class GetPluginVersionsPlugin():
    """
    Class implementing a pytest plugin to extract the version info of all
    installed plugins.
    """
    def __init__(self):
        """
        Constructor
        """
        super().__init__()
        
        self.versions = []
    
    def pytest_cmdline_main(self, config):
        """
        Public method called for performing the main command line action.
        
        @param config pytest config object
        @type Config
        """
        pluginInfo = config.pluginmanager.list_plugin_distinfo()
        if pluginInfo:
            for _plugin, dist in pluginInfo:
                self.versions.append({
                    "name": dist.project_name,
                    "version": dist.version
                })
    
    def getVersions(self):
        """
        Public method to get the assembled list of plugin versions.
        
        @return list of collected plugin versions
        @rtype list of dict
        """
        return self.versions


class EricPlugin():
    """
    Class implementing a pytest plugin which reports the data in a format
    suitable for the PytestExecutor.
    """
    def __init__(self, writer):
        """
        Constructor
        
        @param writer reference to the object to write the results to
        @type EricJsonWriter
        """
        self.__writer = writer
        
        self.__testsRun = 0
    
    def __initializeReportData(self):
        """
        Private method to initialize attributes for data collection.
        """
        self.__status = '---'
        self.__duration = 0
        self.__report = []
        self.__reportPhase = ""
        self.__sections = []
        self.__hadError = False
        self.__wasSkipped = False
        self.__wasXfail = False
    
    def pytest_report_header(self, config, startdir):
        """
        Public method called by pytest before any reporting.
        
        @param config reference to the configuration object
        @type Config
        @param startdir starting directory
        @type LocalPath
        """
        self.__writer.write({
            "event": "config",
            "root": str(config.rootdir)
        })
    
    def pytest_collectreport(self, report):
        """
        Public method called by pytest after the tests have been collected.
        
        @param report reference to the report object
        @type CollectReport
        """
        if report.outcome == "failed":
            self.__writer.write({
                "event": "collecterror",
                "nodeid": report.nodeid,
                "report": str(report.longrepr),
            })
    
    def pytest_itemcollected(self, item):
        """
        Public malled by pytest after a test item has been collected.
        
        @param item reference to the collected test item
        @type Item
        """
        self.__writer.write({
            "event": "collected",
            "nodeid": item.nodeid,
            "name": item.name,
        })
    
    def pytest_runtest_logstart(self, nodeid, location):
        """
        Public method called by pytest before running a test.
        
        @param nodeid node id of the test item
        @type str
        @param location tuple containing the file name, the line number and
            the test name
        @type tuple of (str, int, str)
        """
        self.__testsRun += 1
        
        self.__writer.write({
            "event": "starttest",
            "nodeid": nodeid,
        })
        
        self.__initializeReportData()
    
    def pytest_runtest_logreport(self, report):
        """
        Public method called by pytest when a test phase (setup, call and
            teardown) has been completed.
        
        @param report reference to the test report object
        @type TestReport
        """
        if report.when == "call":
            self.__status = report.outcome
            self.__duration = report.duration
        else:
            if report.outcome == "failed":
                self.__hadError = True
            elif report.outcome == "skipped":
                self.__wasSkipped = True
        
        if hasattr(report, "wasxfail"):
            self.__wasXfail = True
            self.__report.append(report.wasxfail)
            self.__reportPhase = report.when
        
        self.__sections = report.sections
        
        if report.longrepr:
            self.__reportPhase = report.when
            if (
                hasattr(report.longrepr, "reprcrash") and
                report.longrepr.reprcrash is not None
            ):
                self.__report.append(
                    report.longrepr.reprcrash.message)
            if isinstance(report.longrepr, tuple):
                self.__report.append(report.longrepr[2])
            elif isinstance(report.longrepr, str):
                self.__report.append(report.longrepr)
            else:
                self.__report.append(str(report.longrepr))
    
    def pytest_runtest_logfinish(self, nodeid, location):
        """
        Public method called by pytest after a test has been completed.
        
        @param nodeid node id of the test item
        @type str
        @param location tuple containing the file name, the line number and
            the test name
        @type tuple of (str, int, str)
        """
        if self.__wasXfail:
            self.__status = (
                "xpassed"
                if self.__status == "passed" else
                "xfailed"
            )
        elif self.__wasSkipped:
            self.__status = "skipped"
        
        data = {
            "event": "result",
            "status": self.__status,
            "with_error": self.__hadError,
            "sections": self.__sections,
            "duration_s": self.__duration,
            "nodeid": nodeid,
            "filename": location[0],
            "linenumber": location[1],
            "report_phase": self.__reportPhase,
        }
        if self.__report:
            messageLines = self.__report[0].rstrip().splitlines()
            data["message"] = messageLines[0]
        data["report"] = "\n".join(self.__report)
        
        self.__writer.write(data)
    
    def pytest_sessionstart(self, session):
        """
        Public method called by pytest before performing collection and
        entering the run test loop.
        
        @param session reference to the session object
        @type Session
        """
        self.__totalStartTime = time.monotonic_ns()
        self.__testsRun = 0
    
    def pytest_sessionfinish(self, session, exitstatus):
        """
        Public method called by pytest after the whole test run finished.
        
        @param session reference to the session object
        @type Session
        @param exitstatus exit status
        @type int or ExitCode
        """
        stopTime = time.monotonic_ns()
        duration = (stopTime - self.__totalStartTime) / 1_000_000_000   # s
        
        self.__writer.write({
            "event": "finished",
            "duration_s": duration,
            "tests": self.__testsRun,
        })


def getVersions():
    """
    Function to determine the framework version and versions of all available
    plugins.
    """
    try:
        import pytest
        versions = {
            "name": "pytest",
            "version": pytest.__version__,
            "plugins": [],
        }
        
        # --capture=sys needed on Windows to avoid
        # ValueError: saved filedescriptor not valid anymore
        plugin = GetPluginVersionsPlugin()
        pytest.main(['--version', '--capture=sys'], plugins=[plugin])
        versions["plugins"] = plugin.getVersions()
    except ImportError:
        versions = {}
    
    print(json.dumps(versions))
    sys.exit(0)


if __name__ == '__main__':
    command = sys.argv[1]
    if command == "installed":
        try:
            import pytest           # __IGNORE_WARNING__
            sys.exit(0)
        except ImportError:
            sys.exit(1)
    
    elif command == "versions":
        getVersions()
    
    elif command == "runtest":
        import pytest
        from EricNetwork.EricJsonStreamWriter import EricJsonWriter
        writer = EricJsonWriter(sys.argv[2], int(sys.argv[3]))
        pytest.main(sys.argv[4:], plugins=[EricPlugin(writer)])
        writer.close()
        sys.exit(0)
    
    sys.exit(42)

#
# eflag: noqa = M801