Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use pytest_testnodedown method in pytest-cov

Best Python code snippet using pytest-cov

test_dsession.py

Source:test_dsession.py

1from py.__.test.dist.dsession import DSession2from py.__.test import outcome3import py4XSpec = py.execnet.XSpec5def run(item, node, excinfo=None):6    runner = item.config.pluginmanager.getplugin("runner")7    rep = runner.ItemTestReport(item=item, 8        excinfo=excinfo, when="call")9    rep.node = node10    return rep 11class MockNode:12    def __init__(self):13        self.sent = []14    def sendlist(self, items):15        self.sent.append(items)16    def shutdown(self):17        self._shutdown=True18def dumpqueue(queue):19    while queue.qsize():20        print queue.get()21class TestDSession:22    def test_add_remove_node(self, testdir):23        item = testdir.getitem("def test_func(): pass")24        node = MockNode()25        rep = run(item, node)26        session = DSession(item.config)27        assert not session.node2pending28        session.addnode(node)29        assert len(session.node2pending) == 130        session.senditems_load([item])31        pending = session.removenode(node)32        assert pending == [item]33        assert item not in session.item2nodes34        l = session.removenode(node)35        assert not l 36    def test_senditems_each_and_receive_with_two_nodes(self, testdir):37        item = testdir.getitem("def test_func(): pass")38        node1 = MockNode()39        node2 = MockNode()40        session = DSession(item.config)41        session.addnode(node1)42        session.addnode(node2)43        session.senditems_each([item])44        assert session.node2pending[node1] == [item]45        assert session.node2pending[node2] == [item]46        assert node1 in session.item2nodes[item]47        assert node2 in session.item2nodes[item]48        session.removeitem(item, node1)49        assert session.item2nodes[item] == [node2]50        session.removeitem(item, node2)51        assert not session.node2pending[node1] 52        assert not session.item2nodes53    def test_senditems_load_and_receive_one_node(self, testdir):54        item = testdir.getitem("def test_func(): pass")55        node = MockNode()56        rep = run(item, node)57        session = DSession(item.config)58        session.addnode(node)59        session.senditems_load([item])  60        assert session.node2pending[node] == [item]61        assert session.item2nodes[item] == [node]62        session.removeitem(item, node)63        assert not session.node2pending[node] 64        assert not session.item2nodes65    def test_triggertesting_collect(self, testdir):66        modcol = testdir.getmodulecol("""67            def test_func():68                pass69        """)70        session = DSession(modcol.config)71        session.triggertesting([modcol])72        name, args, kwargs = session.queue.get(block=False)73        assert name == 'pytest_collectreport'74        report = kwargs['report']  75        assert len(report.result) == 176    def test_triggertesting_item(self, testdir):77        item = testdir.getitem("def test_func(): pass")78        session = DSession(item.config)79        node1 = MockNode()80        node2 = MockNode()81        session.addnode(node1)82        session.addnode(node2)83        session.triggertesting([item] * (session.MAXITEMSPERHOST*2 + 1))84        sent1 = node1.sent[0]85        sent2 = node2.sent[0]86        assert sent1 == [item] * session.MAXITEMSPERHOST87        assert sent2 == [item] * session.MAXITEMSPERHOST88        assert session.node2pending[node1] == sent189        assert session.node2pending[node2] == sent290        name, args, kwargs = session.queue.get(block=False)91        assert name == "pytest_rescheduleitems"92        assert kwargs['items'] == [item]93    def test_keyboardinterrupt(self, testdir):94        item = testdir.getitem("def test_func(): pass")95        session = DSession(item.config)96        def raise_(timeout=None): raise KeyboardInterrupt()97        session.queue.get = raise_98        exitstatus = session.loop([])99        assert exitstatus == outcome.EXIT_INTERRUPTED100    def test_internalerror(self, testdir):101        item = testdir.getitem("def test_func(): pass")102        session = DSession(item.config)103        def raise_(): raise ValueError()104        session.queue.get = raise_105        exitstatus = session.loop([])106        assert exitstatus == outcome.EXIT_INTERNALERROR107    def test_rescheduleevent(self, testdir):108        item = testdir.getitem("def test_func(): pass")109        session = DSession(item.config)110        node = MockNode()111        session.addnode(node)112        loopstate = session._initloopstate([])113        session.queueevent("pytest_rescheduleitems", items=[item])114        session.loop_once(loopstate)115        # check that RescheduleEvents are not immediately116        # rescheduled if there are no nodes117        assert loopstate.dowork == False 118        session.queueevent(None)119        session.loop_once(loopstate)120        session.queueevent(None)121        session.loop_once(loopstate)122        assert node.sent == [[item]]123        session.queueevent("pytest_runtest_logreport", report=run(item, node))124        session.loop_once(loopstate)125        assert loopstate.shuttingdown 126        assert not loopstate.testsfailed 127    def test_no_node_remaining_for_tests(self, testdir):128        item = testdir.getitem("def test_func(): pass")129        # setup a session with one node130        session = DSession(item.config)131        node = MockNode()132        session.addnode(node)133       134        # setup a HostDown event135        session.queueevent("pytest_testnodedown", node=node, error=None)136        loopstate = session._initloopstate([item])137        loopstate.dowork = False138        session.loop_once(loopstate)139        dumpqueue(session.queue)140        assert loopstate.exitstatus == outcome.EXIT_NOHOSTS141    def test_removeitem_from_failing_teardown(self, testdir):142        # teardown reports only come in when they signal a failure143        # internal session-management should basically ignore them 144        # XXX probably it'S best to invent a new error hook for 145        # teardown/setup related failures146        modcol = testdir.getmodulecol("""147            def test_one(): 148                pass 149            def teardown_function(function):150                assert 0151        """)152        item1, = modcol.collect()153        # setup a session with two nodes154        session = DSession(item1.config)155        node1, node2 = MockNode(), MockNode()156        session.addnode(node1)157        session.addnode(node2)158      159        # have one test pending for a node that goes down 160        session.senditems_each([item1])161        nodes = session.item2nodes[item1]162        class rep:163            failed = True164            item = item1165            node = nodes[0]166            when = "call"167        session.queueevent("pytest_runtest_logreport", report=rep)168        reprec = testdir.getreportrecorder(session)169        print session.item2nodes170        loopstate = session._initloopstate([])171        assert len(session.item2nodes[item1]) == 2172        session.loop_once(loopstate)173        assert len(session.item2nodes[item1]) == 1174        rep.when = "teardown"175        session.queueevent("pytest_runtest_logreport", report=rep)176        session.loop_once(loopstate)177        assert len(session.item2nodes[item1]) == 1178    def test_testnodedown_causes_reschedule_pending(self, testdir):179        modcol = testdir.getmodulecol("""180            def test_crash(): 181                assert 0182            def test_fail(): 183                x184        """)185        item1, item2 = modcol.collect()186        # setup a session with two nodes187        session = DSession(item1.config)188        node1, node2 = MockNode(), MockNode()189        session.addnode(node1)190        session.addnode(node2)191      192        # have one test pending for a node that goes down 193        session.senditems_load([item1, item2])194        node = session.item2nodes[item1] [0]195        item1.config.option.dist = "load"196        session.queueevent("pytest_testnodedown", node=node, error="xyz")197        reprec = testdir.getreportrecorder(session)198        print session.item2nodes199        loopstate = session._initloopstate([])200        session.loop_once(loopstate)201        assert loopstate.colitems == [item2] # do not reschedule crash item202        rep = reprec.matchreport(names="pytest_runtest_logreport")203        assert rep.failed204        assert rep.item == item1205        assert str(rep.longrepr).find("crashed") != -1206        #assert str(testrep.longrepr).find(node.gateway.spec) != -1207    def test_testnodeready_adds_to_available(self, testdir):208        item = testdir.getitem("def test_func(): pass")209        # setup a session with two nodes210        session = DSession(item.config)211        node1 = MockNode()212        session.queueevent("pytest_testnodeready", node=node1)213        loopstate = session._initloopstate([item])214        loopstate.dowork = False215        assert len(session.node2pending) == 0216        session.loop_once(loopstate)217        assert len(session.node2pending) == 1218    def runthrough(self, item, excinfo=None):219        session = DSession(item.config)220        node = MockNode()221        session.addnode(node)222        loopstate = session._initloopstate([item])223        session.queueevent(None)224        session.loop_once(loopstate)225        assert node.sent == [[item]]226        ev = run(item, node, excinfo=excinfo) 227        session.queueevent("pytest_runtest_logreport", report=ev)228        session.loop_once(loopstate)229        assert loopstate.shuttingdown  230        session.queueevent("pytest_testnodedown", node=node, error=None)231        session.loop_once(loopstate)232        dumpqueue(session.queue)233        return session, loopstate.exitstatus 234    def test_exit_completed_tests_ok(self, testdir):235        item = testdir.getitem("def test_func(): pass")236        session, exitstatus = self.runthrough(item)237        assert exitstatus == outcome.EXIT_OK238    def test_exit_completed_tests_fail(self, testdir):239        item = testdir.getitem("def test_func(): 0/0")240        session, exitstatus = self.runthrough(item, excinfo="fail")241        assert exitstatus == outcome.EXIT_TESTSFAILED242    def test_exit_on_first_failing(self, testdir):243        modcol = testdir.getmodulecol("""244            def test_fail(): 245                assert 0246            def test_pass(): 247                pass248        """)249        modcol.config.option.exitfirst = True250        session = DSession(modcol.config)251        node = MockNode()252        session.addnode(node)253        items = modcol.config.hook.pytest_make_collect_report(collector=modcol).result254        # trigger testing  - this sends tests to the node255        session.triggertesting(items)256        # run tests ourselves and produce reports 257        ev1 = run(items[0], node, "fail")258        ev2 = run(items[1], node, None)259        session.queueevent("pytest_runtest_logreport", report=ev1) # a failing one260        session.queueevent("pytest_runtest_logreport", report=ev2)261        # now call the loop262        loopstate = session._initloopstate(items)263        session.loop_once(loopstate)264        assert loopstate.testsfailed265        assert loopstate.shuttingdown266    def test_shuttingdown_filters(self, testdir):267        item = testdir.getitem("def test_func(): pass")268        session = DSession(item.config)269        node = MockNode()270        session.addnode(node)271        loopstate = session._initloopstate([])272        loopstate.shuttingdown = True273        reprec = testdir.getreportrecorder(session)274        session.queueevent("pytest_runtest_logreport", report=run(item, node))275        session.loop_once(loopstate)276        assert not reprec.getcalls("pytest_testnodedown")277        session.queueevent("pytest_testnodedown", node=node, error=None)278        session.loop_once(loopstate)279        assert reprec.getcall('pytest_testnodedown').node == node280    def test_filteritems(self, testdir):281        modcol = testdir.getmodulecol("""282            def test_fail(): 283                assert 0284            def test_pass(): 285                pass286        """)287        session = DSession(modcol.config)288        modcol.config.option.keyword = "nothing"289        dsel = session.filteritems([modcol])290        assert dsel == [modcol] 291        items = modcol.collect()292        hookrecorder = testdir.getreportrecorder(session).hookrecorder293        remaining = session.filteritems(items)294        assert remaining == []295        296        event = hookrecorder.getcalls("pytest_deselected")[-1]297        assert event.items == items 298        modcol.config.option.keyword = "test_fail"299        remaining = session.filteritems(items)300        assert remaining == [items[0]]301        event = hookrecorder.getcalls("pytest_deselected")[-1]302        assert event.items == [items[1]]303    def test_testnodedown_shutdown_after_completion(self, testdir):304        item = testdir.getitem("def test_func(): pass")305        session = DSession(item.config)306        node = MockNode()307        session.addnode(node)308        session.senditems_load([item])309        session.queueevent("pytest_runtest_logreport", report=run(item, node))310        loopstate = session._initloopstate([])311        session.loop_once(loopstate)312        assert node._shutdown is True313        assert loopstate.exitstatus is None, "loop did not wait for testnodedown"314        assert loopstate.shuttingdown 315        session.queueevent("pytest_testnodedown", node=node, error=None)316        session.loop_once(loopstate)317        assert loopstate.exitstatus == 0318    def test_nopending_but_collection_remains(self, testdir):319        modcol = testdir.getmodulecol("""320            def test_fail(): 321                assert 0322            def test_pass(): 323                pass324        """)325        session = DSession(modcol.config)326        node = MockNode()327        session.addnode(node)328        colreport = modcol.config.hook.pytest_make_collect_report(collector=modcol)329        item1, item2 = colreport.result330        session.senditems_load([item1])331        # node2pending will become empty when the loop sees the report 332        rep = run(item1, node)333        session.queueevent("pytest_runtest_logreport", report=run(item1, node)) 334        # but we have a collection pending335        session.queueevent("pytest_collectreport", report=colreport) 336        loopstate = session._initloopstate([])337        session.loop_once(loopstate)338        assert loopstate.exitstatus is None, "loop did not care for collection report"339        assert not loopstate.colitems 340        session.loop_once(loopstate)341        assert loopstate.colitems == colreport.result342        assert loopstate.exitstatus is None, "loop did not care for colitems"343    def test_dist_some_tests(self, testdir):344        p1 = testdir.makepyfile(test_one="""345            def test_1(): 346                pass347            def test_x():348                import py349                py.test.skip("aaa")350            def test_fail():351                assert 0352        """)353        config = testdir.parseconfig('-d', p1, '--tx=popen')354        dsession = DSession(config)355        hookrecorder = testdir.getreportrecorder(config).hookrecorder356        dsession.main([config.getfsnode(p1)])357        rep = hookrecorder.popcall("pytest_runtest_logreport").report358        assert rep.passed359        rep = hookrecorder.popcall("pytest_runtest_logreport").report360        assert rep.skipped361        rep = hookrecorder.popcall("pytest_runtest_logreport").report362        assert rep.failed363        # see that the node is really down 364        node = hookrecorder.popcall("pytest_testnodedown").node365        assert node.gateway.spec.popen366        #XXX eq.geteventargs("pytest_sessionfinish")367def test_collected_function_causes_remote_skip(testdir):368    sub = testdir.mkpydir("testing")369    sub.join("test_module.py").write(py.code.Source("""370        import py371        path = py.path.local(%r)372        if path.check():373            path.remove()374        else:375            py.test.skip("remote skip")376        def test_func(): 377            pass378        def test_func2(): 379            pass380    """ % str(sub.ensure("somefile"))))381    result = testdir.runpytest('-v', '--dist=each', '--tx=popen')382    result.stdout.fnmatch_lines([383        "*2 skipped*"384    ])385def test_teardownfails_one_function(testdir):386    p = testdir.makepyfile("""387        def test_func(): 388            pass389        def teardown_function(function):390            assert 0391    """)392    result = testdir.runpytest(p, '--dist=each', '--tx=popen')393    result.stdout.fnmatch_lines([394        "*def teardown_function(function):*", 395        "*1 passed*1 error*"...

pytest_cov.py

Source:pytest_cov.py

...60        Mark this hook as optional in case xdist is not installed.61        """62        self.cov_controller.configure_node(node)63    pytest_configure_node.optionalhook = True64    def pytest_testnodedown(self, node, error):65        """Delegate to our implementation.66        Mark this hook as optional in case xdist is not installed.67        """68        self.cov_controller.testnodedown(node, error)69    pytest_testnodedown.optionalhook = True70    def pytest_sessionfinish(self, session, exitstatus):71        """Delegate to our implementation."""72        self.cov_controller.finish()73    def pytest_terminal_summary(self, terminalreporter):74        """Delegate to our implementation."""75        self.cov_controller.summary(terminalreporter._tw)76def pytest_funcarg__cov(request):77    """A pytest funcarg that provides access to the underlying coverage object."""78    # Check with hasplugin to avoid getplugin exception in older pytest....

test_xdist.py

Source:test_xdist.py

...24    assert fake_session.config.workeroutput == {"pytest_durations": measurements}25def test_pytest_sessionfinish_noxdist(fake_session, instance, measurements):26    instance.pytest_sessionfinish(fake_session, 0)27    assert isinstance(fake_session.config.workeroutput["pytest_durations"], Mock)28def test_pytest_testnodedown(fake_node, instance, measurements):29    fake_node.workeroutput = {"pytest_durations": measurements}30    instance.pytest_testnodedown(fake_node, None)31    assert instance.measurements == measurements32def test_pytest_testnodedown_noxdist(fake_node, instance, measurements):33    instance.pytest_testnodedown(fake_node, None)...

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.