KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use pytest_runtestloop method in Pytest

Best Python code snippet using pytest

scheduler.py

Source:scheduler.py

...8level. For instance, all performance tests will run sequentially9(i.e. concurrency=1), since they rely on the availability of the full host10resources, in order to make accurate measurements. Additionally, other tests11may be restricted to running sequentially, if they are per se12concurrency-unsafe. See `PytestScheduler.pytest_runtestloop()`.13Scheduling is achieved by overriding the pytest run loop (i.e.14`pytest_runtestloop()`), and splitting the test session item list across15multiple `fork()`ed worker processes. Since no user code is run before16`pytest_runtestloop()`, each worker becomes a pytest session itself.17Reporting is disabled for worker process, each worker sending its results18back to the main / server process, via an IPC pipe, for aggregation.19"""20import multiprocessing as mp21import os22import re23import sys24from random import random25from select import select26from time import sleep27import pytest28from _pytest.main import ExitCode29from . import mpsing  # pylint: disable=relative-beyond-top-level30class PytestScheduler(mpsing.MultiprocessSingleton):31    """A pretty custom test execution scheduler."""32    def __init__(self):33        """Initialize the scheduler.34        Not to be called directly, since this is a singleton. Use35        `PytestScheduler.instance()` to get the scheduler object.36        """37        super().__init__()38        self._mp_singletons = [self]39        self.session = None40    def register_mp_singleton(self, mp_singleton):41        """Register a multi-process singleton object.42        Since the scheduler will be handling the main testing loop, it needs43        to be aware of any multi-process singletons that must be serviced44        during the test run (i.e. polled and allowed to handle method45        execution in the server context).46        """47        self._mp_singletons.append(mp_singleton)48    @staticmethod49    def do_pytest_addoption(parser):50        """Pytest hook. Add concurrency command line option."""51        avail_cpus = len(os.sched_getaffinity(0))52        # Defaulting to a third of the available (logical) CPUs sounds like a53        # good enough plan.54        default = max(1, int(avail_cpus / 3))55        parser.addoption(56            "--concurrency",57            "--concurrency",58            dest="concurrency",59            action="store",60            type=int,61            default=default,62            help="Concurrency level (max number of worker processes to spawn)."63        )64    def pytest_sessionstart(self, session):65        """Pytest hook. Called at pytest session start.66        This will execute in the server context (before the tests are67        executed).68        """69        self.session = session70    def pytest_runtest_logreport(self, report):71        """Pytest hook. Called whenever a new test report is ready.72        This will execute in the worker / child context.73        """74        self._add_report(report)75    def pytest_runtestloop(self, session):76        """Pytest hook. The main test scheduling and running loop.77        Called in the server process context.78        """79        # Don't run tests on test discovery80        if session.config.option.collectonly:81            return True82        # max_concurrency = self.session.config.option.concurrency83        schedule = [84            {85                # Performance batch: tests that measure performance, and need86                # to be run in a non-cuncurrent environment.87                'name': 'performance',88                'concurrency': 1,89                'patterns': [...

spydist.py

Source:spydist.py

...81        self.thread = threading.Thread(target=self.server.start)82        self.thread.start()83    def pytest_sessionfinish(self, session):84        debug("master: pytest_sessionfinish", session)85    def pytest_runtestloop(self):86        if wa.start_slaves_from_master:87            slaves_init(self.logs_path)88        try:89            conn = rpyc.connect("127.0.0.1", self.port)90            while 1:91                if not getattr(conn.root, "has_pending")():92                    break93                debug("master: pytest_runtestloop")94                time.sleep(5)95        except KeyboardInterrupt:96            trace("master: interrupted")97            getattr(conn.root, "shutdown")()98            time.sleep(5)99        os._exit(0)100    def pytest_terminal_summary(self, terminalreporter):101        debug("master: pytest_terminal_summary", terminalreporter)102class BatchSlave(object):103    def __init__(self, config, logs_path):104        self.config = config105        self.items = []106        self.logs_path = logs_path107    @pytest.mark.trylast108    def pytest_sessionstart(self, session):109        debug("slave: pytest_sessionstart", session)110    def pytest_sessionfinish(self, session):111        debug("slave: pytest_sessionfinish", session)112    @pytest.hookimpl(trylast=True)113    def pytest_collection_modifyitems(self, session, config, items):114        debug("slave: pytest_collection_modifyitems", session, config, items)115        self.items = items116    def pytest_runtestloop(self):117        def search_nodeid(entries, nodeid):118            for ent in entries:119                if nodeid == ent.nodeid:120                    return ent121            return None122        def finish_test(item):123            getattr(conn.root, "finish_test")(item.nodeid)124        def get_test(entries):125            while 1:126                nodeid = getattr(conn.root, "get_test")()127                if not nodeid:128                    break129                item = search_nodeid(entries, nodeid)130                if item:...

pydev_runfiles_pytest.py

Source:pydev_runfiles_pytest.py

...91        if hasattr(config.option, 'numprocesses'):92            if config.option.numprocesses:93                self._using_xdist = True94                pydev_runfiles_xml_rpc.notifyTestRunFinished('Unable to show results (py.test xdist plugin not compatible with PyUnit view)')95    def pytest_runtestloop(self, session):96        if self._using_xdist:97            #Yes, we don't have the hooks we'd need to show the results in the pyunit view...98            #Maybe the plugin maintainer may be able to provide these additional hooks?99            return None100        101        #This mock will make all file representations to be printed as Pydev expects, 102        #so that hyperlinks are properly created in errors. Note that we don't unmock it!103        self._MockFileRepresentation()104        105        #Based on the default run test loop: _pytest.session.pytest_runtestloop106        #but getting the times we need, reporting the number of tests found and notifying as each107        #test is run.108        109        start_total = time.time()...

plugin.py

Source:plugin.py

...27        action="store_true",28        dest="find_dependencies_internal",29        help="""For internal use only""",30    )31def pytest_runtestloop(session):32    if session.config.getoption("find_dependencies_internal"):33        return run_tests(session)34    if not session.config.getoption("find_dependencies"):35        return pytest_main.pytest_runtestloop(session)36    if len(session.items) == 1:37        print("Only one test collected: ignoring option --find-dependencies")38        restore_verbosity(session.config)39        return pytest_main.pytest_runtestloop(session)40    if (session.testsfailed and41            not session.config.option.continue_on_collection_errors):42        restore_verbosity(session.config)43        raise session.Interrupted(44            "%d errors during collection" % session.testsfailed)45    DependencyFinder(session).find_dependencies()46    return True47def restore_verbosity(config):48    verbosity = 049    if hasattr(config, "initial_args"):50        for arg in config.initial_args:51            if arg.startswith("-v"):52                verbosity = len(arg) - 153                break...

Pytest Tutorial

Looking for an in-depth tutorial around pytest? LambdaTest covers the detailed pytest tutorial that has everything related to the pytest, from setting up the pytest framework to automation testing. Delve deeper into pytest testing by exploring advanced use cases like parallel testing, pytest fixtures, parameterization, executing multiple test cases from a single file, and more.

Chapters

What is pytest
Pytest installation: Want to start pytest from scratch? See how to install and configure pytest for Python automation testing.
Run first test with pytest framework: Follow this step-by-step tutorial to write and run your first pytest script.
Parallel testing with pytest: A hands-on guide to parallel testing with pytest to improve the scalability of your test automation.
Generate pytest reports: Reports make it easier to understand the results of pytest-based test runs. Learn how to generate pytest reports.
Pytest Parameterized tests: Create and run your pytest scripts while avoiding code duplication and increasing test coverage with parameterization.
Pytest Fixtures: Check out how to implement pytest fixtures for your end-to-end testing needs.
Execute Multiple Test Cases: Explore different scenarios for running multiple test cases in pytest from a single file.
Stop Test Suite after N Test Failures: See how to stop your test suite after n test failures in pytest using the @pytest.mark.incremental decorator and maxfail command-line option.