eric7/Testing/Interfaces/PytestRunner.py

branch
unittest
changeset 9089
b48a6d0f6309
parent 9066
a219ade50f7c
child 9192
a763d57e23bc
diff -r 4dc05dd170a5 -r b48a6d0f6309 eric7/Testing/Interfaces/PytestRunner.py
--- a/eric7/Testing/Interfaces/PytestRunner.py	Fri May 20 11:31:18 2022 +0200
+++ b/eric7/Testing/Interfaces/PytestRunner.py	Mon May 23 16:48:19 2022 +0200
@@ -8,9 +8,14 @@
 """
 
 import json
+import os
 import sys
+import time
 
-# TODO: implement 'pytest' support in PytestRunner
+sys.path.insert(
+    2,
+    os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+)
 
 
 class GetPluginVersionsPlugin():
@@ -51,13 +56,208 @@
         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               # __IGNORE_WARNING__
+        import pytest
         versions = {
             "name": "pytest",
             "version": pytest.__version__,
@@ -88,6 +288,14 @@
     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)
 
 #

eric ide

mercurial