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

How to use partition_tests method in stestr

Best Python code snippet using stestr_python

concurrencytest.py

Source:concurrencytest.py

...40        :return: An iterable of TestCase-like objects which can each have41        run(result) called on them to feed tests to result.42        """43        result = []44        test_blocks = partition_tests(suite, concurrency_num)45        # Clear the tests from the original suite so it doesn't keep them alive46        suite._tests[:] = []47        for process_tests in test_blocks:48            process_suite = unittest.TestSuite(process_tests)49            # Also clear each split list so new suite has only reference50            process_tests[:] = []51            c2pread, c2pwrite = os.pipe()52            pid = os.fork()53            if pid == 0:54                try:55                    stream = os.fdopen(c2pwrite, 'wb')56                    os.close(c2pread)57                    # Leave stderr and stdout open so we can see test noise58                    # Close stdin so that the child goes away if it decides to59                    # read from stdin (otherwise its a roulette to see what60                    # child actually gets keystrokes for pdb etc).61                    sys.stdin.close()62                    subunit_result = AutoTimingTestResultDecorator(63                        TestProtocolClient(stream)64                    )65                    process_suite.run(subunit_result)66                except:67                    # Try and report traceback on stream, but exit with error68                    # even if stream couldn't be created or something else69                    # goes wrong.  The traceback is formatted to a string and70                    # written in one go to avoid interleaving lines from71                    # multiple failing children.72                    try:73                        stream.write(traceback.format_exc())74                    finally:75                        os._exit(1)76                os._exit(0)77            else:78                os.close(c2pwrite)79                stream = os.fdopen(c2pread, 'rb')80                test = ProtocolTestCase(stream)81                result.append(test)82        return result83    return do_fork84def partition_tests(suite, count):85    """Partition suite into count lists of tests."""86    # This just assigns tests in a round-robin fashion.  On one hand this87    # splits up blocks of related tests that might run faster if they shared88    # resources, but on the other it avoids assigning blocks of slow tests to89    # just one partition.  So the slowest partition shouldn't be much slower90    # than the fastest.91    partitions = [list() for _ in range(count)]92    tests = iterate_tests(suite)93    for partition, test in zip(cycle(partitions), tests):94        partition.append(test)95    return partitions96if __name__ == '__main__':97    import time98    class SampleTestCase(unittest.TestCase):...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.