Search code examples
pythonnose

Python nosetests: add** hooks of custom plugin do not called when tests are run in multiple processes


Trying to run nose tests in parallel and want to make a summary report at the end of tests run.
Want to use nose custom plugin to collect all test execution data and aggregate report at the end of the run-in report method.
This code runs tests in parallel and uses a custom plugin and calls it's report method at the end of a run but add** hooks are not called during execution and data is not collected for aggregation.

Can anybody tell please why addSuccess, addError, and addFailure methods are skipped during tests run?

If I remove these options --processes=16, --process-timeout=360 then the hooks in the plugin works fine but the whole point in this code is a parallel test run so I can not remove them.

import nose
import os
from nose.plugins import Plugin


class ReportPlugin(Plugin):
    name = 'reportplugin'
    score = 2500

    def options(self, parser, env):
        super(ReportPlugin, self).options(parser, env)

    def configure(self, options, conf):
        super(ReportPlugin, self).configure(options, conf)
        self.enabled = True

    def addSuccess(self, test, capt=None):
        print('*** SUCCESS')
        self.handler("SUCCESS", test)

    def addError(self, test, err, capt=None):
        print('*** ERROR')
        self.handler("ERROR", test)

    def addFailure(self, test, err, capt=None, tb_info=None):
        print('*** FAILURE')
        self.handler("FAILURE", test)

    def handler(self, status, test):
        # handle test run, collect status in multiprocessing manager

    def report(self, stream):
        # make nice report


TODO_TESTS = 'test_file.py'
path_to_tests = os.path.join(os.path.dirname(__file__), TODO_TESTS)

if __name__ == '__main__':
    nose.run(
        argv=['nosetests', '--processes=16', '--process-timeout=360', '--nocapture', '--verbose',
              '--stop', '--with-xunitmp', '--xunitmp-file=report.xml', path_to_tests],
        addplugins=[ReportPlugin()],
    )

Solution

  • Solved this problem after loading plugin entry point to nose.plugins.0.10 namespace

    def load_entry_point(namespace, entry_name, class_path):
        """Function to load temporary entry point. Need to load plugins to nose."""
        distribution = pkg_resources.Distribution(__file__)
        entry_point = pkg_resources.EntryPoint.parse(f'{entry_name} = {class_path}', dist=distribution)
        distribution._ep_map = {namespace: {entry_name: entry_point}}
        pkg_resources.working_set.add(distribution)
    
    
    load_entry_point('nose.plugins.0.10', 'multiprocess_report', 'run_tests:MultiprocessReport')
    
    
    # ... setting test args for multiprocessing here.
    
    nose.run(argv=test_run_args,)