Best Python code snippet using pytest-mozwebqa_python
test_txnode.py
Source:test_txnode.py  
1import py2from py.__.test.dist.txnode import TXNode3class EventQueue:4    def __init__(self, registry, queue=None):5        if queue is None:6            queue = py.std.Queue.Queue()7        self.queue = queue8        registry.register(self)9    def geteventargs(self, eventname, timeout=2.0):10        events = []11        while 1:12            try:13                eventcall = self.queue.get(timeout=timeout)14            except py.std.Queue.Empty:15                #print "node channel", self.node.channel16                #print "remoteerror", self.node.channel._getremoteerror()17                print "seen events", events18                raise IOError("did not see %r events" % (eventname))19            else:20                name, args, kwargs = eventcall 21                assert isinstance(name, str)22                if name == eventname:23                    if args:24                        return args25                    return kwargs26                events.append(name)27                if name == "pytest_internalerror":28                    print str(kwargs["excrepr"])29class MySetup:30    def __init__(self, request):31        self.id = 032        self.request = request33    def geteventargs(self, eventname, timeout=2.0):34        eq = EventQueue(self.config.pluginmanager, self.queue)35        return eq.geteventargs(eventname, timeout=timeout)36    def makenode(self, config=None):37        if config is None:38            config = py.test.config._reparse([])39        self.config = config40        self.queue = py.std.Queue.Queue()41        self.xspec = py.execnet.XSpec("popen")42        self.gateway = py.execnet.makegateway(self.xspec)43        self.id += 144        self.gateway.id = str(self.id)45        self.node = TXNode(self.gateway, self.config, putevent=self.queue.put)46        assert not self.node.channel.isclosed()47        return self.node 48    def xfinalize(self):49        if hasattr(self, 'node'):50            gw = self.node.gateway51            print "exiting:", gw52            gw.exit()53def pytest_funcarg__mysetup(request):54    mysetup = MySetup(request)55    #pyfuncitem.addfinalizer(mysetup.finalize)56    return mysetup57def test_node_hash_equality(mysetup):58    node = mysetup.makenode()59    node2 = mysetup.makenode()60    assert node != node261    assert node == node62    assert not (node != node)63class TestMasterSlaveConnection:64    def test_crash_invalid_item(self, mysetup):65        node = mysetup.makenode()66        node.send(123) # invalid item 67        kwargs = mysetup.geteventargs("pytest_testnodedown")68        assert kwargs['node'] is node 69        assert str(kwargs['error']).find("AttributeError") != -170    def test_crash_killed(self, testdir, mysetup):71        if not hasattr(py.std.os, 'kill'):72            py.test.skip("no os.kill")73        item = testdir.getitem("""74            def test_func():75                import os76                os.kill(os.getpid(), 15)77        """)78        node = mysetup.makenode(item.config)79        node.send(item) 80        kwargs = mysetup.geteventargs("pytest_testnodedown")81        assert kwargs['node'] is node 82        assert str(kwargs['error']).find("Not properly terminated") != -183    def test_node_down(self, mysetup):84        node = mysetup.makenode()85        node.shutdown()86        kwargs = mysetup.geteventargs("pytest_testnodedown")87        assert kwargs['node'] is node 88        assert not kwargs['error']89        node.callback(node.ENDMARK)90        excinfo = py.test.raises(IOError, 91            "mysetup.geteventargs('testnodedown', timeout=0.01)")92    def test_send_on_closed_channel(self, testdir, mysetup):93        item = testdir.getitem("def test_func(): pass")94        node = mysetup.makenode(item.config)95        node.channel.close()96        py.test.raises(IOError, "node.send(item)")97        #ev = self.getcalls(pytest_internalerror)98        #assert ev.excinfo.errisinstance(IOError)99    def test_send_one(self, testdir, mysetup):100        item = testdir.getitem("def test_func(): pass")101        node = mysetup.makenode(item.config)102        node.send(item)103        kwargs = mysetup.geteventargs("pytest_runtest_logreport")104        rep = kwargs['report'] 105        assert rep.passed 106        print rep107        assert rep.item == item108    def test_send_some(self, testdir, mysetup):109        items = testdir.getitems("""110            def test_pass(): 111                pass112            def test_fail():113                assert 0114            def test_skip():115                import py116                py.test.skip("x")117        """)118        node = mysetup.makenode(items[0].config)119        for item in items:120            node.send(item)121        for outcome in "passed failed skipped".split():122            kwargs = mysetup.geteventargs("pytest_runtest_logreport")123            report = kwargs['report']124            assert getattr(report, outcome) 125        node.sendlist(items)126        for outcome in "passed failed skipped".split():127            rep = mysetup.geteventargs("pytest_runtest_logreport")['report']...pytest_pudb.py
Source:pytest_pudb.py  
...50        https://docs.pytest.org/en/latest/reference.html#_pytest.hookspec.pytest_exception_interact51        """52        self.disable_io_capture()53        _enter_pudb(node, call.excinfo, report)54    def pytest_internalerror(self, excrepr, excinfo):55        """56        Pytest plugin interface for internal errors handling57        https://docs.pytest.org/en/latest/reference.html#_pytest.hookspec.pytest_internalerror58        """59        for line in str(excrepr).split("\n"):60            sys.stderr.write("INTERNALERROR> {}\n".format(line))61            sys.stderr.flush()62        tb = _postmortem_traceback(excinfo)63        post_mortem(tb, excinfo)64    def _suspend_capture(self, capman, *args, **kwargs):65        if hasattr(capman, 'suspendcapture'):66            # pytest changed the suspend capture API since v3.3.167            # see: https://github.com/pytest-dev/pytest/pull/280168            # TODO: drop this case after pytest v3.3.1+ is minimal required...resultlog.py
Source:resultlog.py  
...72                assert report.skipped73                code = "S"74                longrepr = "%s:%d: %s" % report.longrepr75            self.log_outcome(report, code, longrepr)76    def pytest_internalerror(self, excrepr):77        reprcrash = getattr(excrepr, "reprcrash", None)78        path = getattr(reprcrash, "path", None)79        if path is None:80            path = "cwd:%s" % py.path.local()...conftest.py
Source:conftest.py  
...7    @pytest.hookimpl(tryfirst=True)8    def pytest_exception_interact(call):9        raise call.excinfo.value10    @pytest.hookimpl(tryfirst=True)11    def pytest_internalerror(excinfo):...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
