KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run pytest-cov automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
