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

Copied to Clipboard!

How to use listener method in wpt

Best JavaScript code snippet using wpt

network.py

Source:network.py

...444        """445        event_name = "Network.dataReceived"446        if listener is None:447            future = self.client.loop.create_future()448            def _listener(event: Optional[Dict] = None) -> None:449                future.set_result(event)450            self.client.once(event_name, _listener)451            return future452        self.client.on(event_name, listener)453        return lambda: self.client.remove_listener(event_name, listener)454    def eventSourceMessageReceived(455        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None456    ) -> Any:457        """458        Fired when EventSource message is received.459        See `https://chromedevtools.github.io/devtools-protocol/tot/Network#event-eventSourceMessageReceived`460        :param listener: Optional listener function461        :return: If a listener was supplied the return value is a callable that462        will remove the supplied listener otherwise a future that resolves463        with the value of the event464        """465        event_name = "Network.eventSourceMessageReceived"466        if listener is None:467            future = self.client.loop.create_future()468            def _listener(event: Optional[Dict] = None) -> None:469                future.set_result(event)470            self.client.once(event_name, _listener)471            return future472        self.client.on(event_name, listener)473        return lambda: self.client.remove_listener(event_name, listener)474    def loadingFailed(475        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None476    ) -> Any:477        """478        Fired when HTTP request has failed to load.479        See `https://chromedevtools.github.io/devtools-protocol/tot/Network#event-loadingFailed`480        :param listener: Optional listener function481        :return: If a listener was supplied the return value is a callable that482        will remove the supplied listener otherwise a future that resolves483        with the value of the event484        """485        event_name = "Network.loadingFailed"486        if listener is None:487            future = self.client.loop.create_future()488            def _listener(event: Optional[Dict] = None) -> None:489                future.set_result(event)490            self.client.once(event_name, _listener)491            return future492        self.client.on(event_name, listener)493        return lambda: self.client.remove_listener(event_name, listener)494    def loadingFinished(495        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None496    ) -> Any:497        """498        Fired when HTTP request has finished loading.499        See `https://chromedevtools.github.io/devtools-protocol/tot/Network#event-loadingFinished`500        :param listener: Optional listener function501        :return: If a listener was supplied the return value is a callable that502        will remove the supplied listener otherwise a future that resolves503        with the value of the event504        """505        event_name = "Network.loadingFinished"506        if listener is None:507            future = self.client.loop.create_future()508            def _listener(event: Optional[Dict] = None) -> None:509                future.set_result(event)510            self.client.once(event_name, _listener)511            return future512        self.client.on(event_name, listener)513        return lambda: self.client.remove_listener(event_name, listener)514    def requestIntercepted(515        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None516    ) -> Any:517        """518        Details of an intercepted HTTP request, which must be either allowed, blocked, modified or519        mocked.520        See `https://chromedevtools.github.io/devtools-protocol/tot/Network#event-requestIntercepted`521        :param listener: Optional listener function522        :return: If a listener was supplied the return value is a callable that523        will remove the supplied listener otherwise a future that resolves524        with the value of the event525        """526        event_name = "Network.requestIntercepted"527        if listener is None:528            future = self.client.loop.create_future()529            def _listener(event: Optional[Dict] = None) -> None:530                future.set_result(event)531            self.client.once(event_name, _listener)532            return future533        self.client.on(event_name, listener)534        return lambda: self.client.remove_listener(event_name, listener)535    def requestServedFromCache(536        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None537    ) -> Any:538        """539        Fired if request ended up loading from cache.540        See `https://chromedevtools.github.io/devtools-protocol/tot/Network#event-requestServedFromCache`541        :param listener: Optional listener function542        :return: If a listener was supplied the return value is a callable that543        will remove the supplied listener otherwise a future that resolves544        with the value of the event545        """546        event_name = "Network.requestServedFromCache"547        if listener is None:548            future = self.client.loop.create_future()549            def _listener(event: Optional[Dict] = None) -> None:550                future.set_result(event)551            self.client.once(event_name, _listener)552            return future553        self.client.on(event_name, listener)554        return lambda: self.client.remove_listener(event_name, listener)555    def requestWillBeSent(556        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None557    ) -> Any:558        """559        Fired when page is about to send HTTP request.560        See `https://chromedevtools.github.io/devtools-protocol/tot/Network#event-requestWillBeSent`561        :param listener: Optional listener function562        :return: If a listener was supplied the return value is a callable that563        will remove the supplied listener otherwise a future that resolves564        with the value of the event565        """566        event_name = "Network.requestWillBeSent"567        if listener is None:568            future = self.client.loop.create_future()569            def _listener(event: Optional[Dict] = None) -> None:570                future.set_result(event)571            self.client.once(event_name, _listener)572            return future573        self.client.on(event_name, listener)574        return lambda: self.client.remove_listener(event_name, listener)575    def resourceChangedPriority(576        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None577    ) -> Any:578        """579        Fired when resource loading priority is changed580        See `https://chromedevtools.github.io/devtools-protocol/tot/Network#event-resourceChangedPriority`581        :param listener: Optional listener function582        :return: If a listener was supplied the return value is a callable that583        will remove the supplied listener otherwise a future that resolves584        with the value of the event585        """586        event_name = "Network.resourceChangedPriority"587        if listener is None:588            future = self.client.loop.create_future()589            def _listener(event: Optional[Dict] = None) -> None:590                future.set_result(event)591            self.client.once(event_name, _listener)592            return future593        self.client.on(event_name, listener)594        return lambda: self.client.remove_listener(event_name, listener)595    def signedExchangeReceived(596        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None597    ) -> Any:598        """599        Fired when a signed exchange was received over the network600        See `https://chromedevtools.github.io/devtools-protocol/tot/Network#event-signedExchangeReceived`601        :param listener: Optional listener function602        :return: If a listener was supplied the return value is a callable that603        will remove the supplied listener otherwise a future that resolves604        with the value of the event605        """606        event_name = "Network.signedExchangeReceived"607        if listener is None:608            future = self.client.loop.create_future()609            def _listener(event: Optional[Dict] = None) -> None:610                future.set_result(event)611            self.client.once(event_name, _listener)612            return future613        self.client.on(event_name, listener)614        return lambda: self.client.remove_listener(event_name, listener)615    def responseReceived(616        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None617    ) -> Any:618        """619        Fired when HTTP response is available.620        See `https://chromedevtools.github.io/devtools-protocol/tot/Network#event-responseReceived`621        :param listener: Optional listener function622        :return: If a listener was supplied the return value is a callable that623        will remove the supplied listener otherwise a future that resolves624        with the value of the event625        """626        event_name = "Network.responseReceived"627        if listener is None:628            future = self.client.loop.create_future()629            def _listener(event: Optional[Dict] = None) -> None:630                future.set_result(event)631            self.client.once(event_name, _listener)632            return future633        self.client.on(event_name, listener)634        return lambda: self.client.remove_listener(event_name, listener)635    def webSocketClosed(636        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None637    ) -> Any:638        """639        Fired when WebSocket is closed.640        See `https://chromedevtools.github.io/devtools-protocol/tot/Network#event-webSocketClosed`641        :param listener: Optional listener function642        :return: If a listener was supplied the return value is a callable that643        will remove the supplied listener otherwise a future that resolves644        with the value of the event645        """646        event_name = "Network.webSocketClosed"647        if listener is None:648            future = self.client.loop.create_future()649            def _listener(event: Optional[Dict] = None) -> None:650                future.set_result(event)651            self.client.once(event_name, _listener)652            return future653        self.client.on(event_name, listener)654        return lambda: self.client.remove_listener(event_name, listener)655    def webSocketCreated(656        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None657    ) -> Any:658        """659        Fired upon WebSocket creation.660        See `https://chromedevtools.github.io/devtools-protocol/tot/Network#event-webSocketCreated`661        :param listener: Optional listener function662        :return: If a listener was supplied the return value is a callable that663        will remove the supplied listener otherwise a future that resolves664        with the value of the event665        """666        event_name = "Network.webSocketCreated"667        if listener is None:668            future = self.client.loop.create_future()669            def _listener(event: Optional[Dict] = None) -> None:670                future.set_result(event)671            self.client.once(event_name, _listener)672            return future673        self.client.on(event_name, listener)674        return lambda: self.client.remove_listener(event_name, listener)675    def webSocketFrameError(676        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None677    ) -> Any:678        """679        Fired when WebSocket message error occurs.680        See `https://chromedevtools.github.io/devtools-protocol/tot/Network#event-webSocketFrameError`681        :param listener: Optional listener function682        :return: If a listener was supplied the return value is a callable that683        will remove the supplied listener otherwise a future that resolves684        with the value of the event685        """686        event_name = "Network.webSocketFrameError"687        if listener is None:688            future = self.client.loop.create_future()689            def _listener(event: Optional[Dict] = None) -> None:690                future.set_result(event)691            self.client.once(event_name, _listener)692            return future693        self.client.on(event_name, listener)694        return lambda: self.client.remove_listener(event_name, listener)695    def webSocketFrameReceived(696        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None697    ) -> Any:698        """699        Fired when WebSocket message is received.700        See `https://chromedevtools.github.io/devtools-protocol/tot/Network#event-webSocketFrameReceived`701        :param listener: Optional listener function702        :return: If a listener was supplied the return value is a callable that703        will remove the supplied listener otherwise a future that resolves704        with the value of the event705        """706        event_name = "Network.webSocketFrameReceived"707        if listener is None:708            future = self.client.loop.create_future()709            def _listener(event: Optional[Dict] = None) -> None:710                future.set_result(event)711            self.client.once(event_name, _listener)712            return future713        self.client.on(event_name, listener)714        return lambda: self.client.remove_listener(event_name, listener)715    def webSocketFrameSent(716        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None717    ) -> Any:718        """719        Fired when WebSocket message is sent.720        See `https://chromedevtools.github.io/devtools-protocol/tot/Network#event-webSocketFrameSent`721        :param listener: Optional listener function722        :return: If a listener was supplied the return value is a callable that723        will remove the supplied listener otherwise a future that resolves724        with the value of the event725        """726        event_name = "Network.webSocketFrameSent"727        if listener is None:728            future = self.client.loop.create_future()729            def _listener(event: Optional[Dict] = None) -> None:730                future.set_result(event)731            self.client.once(event_name, _listener)732            return future733        self.client.on(event_name, listener)734        return lambda: self.client.remove_listener(event_name, listener)735    def webSocketHandshakeResponseReceived(736        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None737    ) -> Any:738        """739        Fired when WebSocket handshake response becomes available.740        See `https://chromedevtools.github.io/devtools-protocol/tot/Network#event-webSocketHandshakeResponseReceived`741        :param listener: Optional listener function742        :return: If a listener was supplied the return value is a callable that743        will remove the supplied listener otherwise a future that resolves744        with the value of the event745        """746        event_name = "Network.webSocketHandshakeResponseReceived"747        if listener is None:748            future = self.client.loop.create_future()749            def _listener(event: Optional[Dict] = None) -> None:750                future.set_result(event)751            self.client.once(event_name, _listener)752            return future753        self.client.on(event_name, listener)754        return lambda: self.client.remove_listener(event_name, listener)755    def webSocketWillSendHandshakeRequest(756        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None757    ) -> Any:758        """759        Fired when WebSocket is about to initiate handshake.760        See `https://chromedevtools.github.io/devtools-protocol/tot/Network#event-webSocketWillSendHandshakeRequest`761        :param listener: Optional listener function762        :return: If a listener was supplied the return value is a callable that763        will remove the supplied listener otherwise a future that resolves764        with the value of the event765        """766        event_name = "Network.webSocketWillSendHandshakeRequest"767        if listener is None:768            future = self.client.loop.create_future()769            def _listener(event: Optional[Dict] = None) -> None:770                future.set_result(event)771            self.client.once(event_name, _listener)772            return future773        self.client.on(event_name, listener)...

interfaces.py

Source:interfaces.py

...22            # etc.23        # create a new pool with a listener24        p = QueuePool(..., listeners=[MyListener()])25        # add a listener after the fact26        p.add_listener(MyListener())27        # usage with create_engine()28        e = create_engine("url://", listeners=[MyListener()])29    All of the standard connection :class:`~sqlalchemy.pool.Pool` types can30    accept event listeners for key connection lifecycle events:31    creation, pool check-out and check-in.  There are no events fired32    when a connection closes.33    For any given DB-API connection, there will be one ``connect``34    event, `n` number of ``checkout`` events, and either `n` or `n - 1`35    ``checkin`` events.  (If a ``Connection`` is detached from its36    pool via the ``detach()`` method, it won't be checked back in.)37    These are low-level events for low-level objects: raw Python38    DB-API connections, without the conveniences of the SQLAlchemy39    ``Connection`` wrapper, ``Dialect`` services or ``ClauseElement``40    execution.  If you execute SQL through the connection, explicitly41    closing all cursors and other resources is recommended.42    Events also receive a ``_ConnectionRecord``, a long-lived internal43    ``Pool`` object that basically represents a "slot" in the44    connection pool.  ``_ConnectionRecord`` objects have one public45    attribute of note: ``info``, a dictionary whose contents are46    scoped to the lifetime of the DB-API connection managed by the47    record.  You can use this shared storage area however you like.48    There is no need to subclass ``PoolListener`` to handle events.49    Any class that implements one or more of these methods can be used50    as a pool listener.  The ``Pool`` will inspect the methods51    provided by a listener object and add the listener to one or more52    internal event queues based on its capabilities.  In terms of53    efficiency and function call overhead, you're much better off only54    providing implementations for the hooks you'll be using.55    """56    @classmethod57    def _adapt_listener(cls, self, listener):58        """Adapt a :class:`.PoolListener` to individual59        :class:`event.Dispatch` events.60        """61        listener = util.as_interface(listener,62                                     methods=('connect', 'first_connect',63                                              'checkout', 'checkin'))64        if hasattr(listener, 'connect'):65            event.listen(self, 'connect', listener.connect)66        if hasattr(listener, 'first_connect'):67            event.listen(self, 'first_connect', listener.first_connect)68        if hasattr(listener, 'checkout'):69            event.listen(self, 'checkout', listener.checkout)70        if hasattr(listener, 'checkin'):71            event.listen(self, 'checkin', listener.checkin)72    def connect(self, dbapi_con, con_record):73        """Called once for each new DB-API connection or Pool's ``creator()``.74        dbapi_con75          A newly connected raw DB-API connection (not a SQLAlchemy76          ``Connection`` wrapper).77        con_record78          The ``_ConnectionRecord`` that persistently manages the connection79        """80    def first_connect(self, dbapi_con, con_record):81        """Called exactly once for the first DB-API connection.82        dbapi_con83          A newly connected raw DB-API connection (not a SQLAlchemy84          ``Connection`` wrapper).85        con_record86          The ``_ConnectionRecord`` that persistently manages the connection87        """88    def checkout(self, dbapi_con, con_record, con_proxy):89        """Called when a connection is retrieved from the Pool.90        dbapi_con91          A raw DB-API connection92        con_record93          The ``_ConnectionRecord`` that persistently manages the connection94        con_proxy95          The ``_ConnectionFairy`` which manages the connection for the span of96          the current checkout.97        If you raise an ``exc.DisconnectionError``, the current98        connection will be disposed and a fresh connection retrieved.99        Processing of all checkout listeners will abort and restart100        using the new connection.101        """102    def checkin(self, dbapi_con, con_record):103        """Called when a connection returns to the pool.104        Note that the connection may be closed, and may be None if the105        connection has been invalidated.  ``checkin`` will not be called106        for detached connections.  (They do not return to the pool.)107        dbapi_con108          A raw DB-API connection109        con_record110          The ``_ConnectionRecord`` that persistently manages the connection111        """112class ConnectionProxy(object):113    """Allows interception of statement execution by Connections.114    .. note::115       :class:`.ConnectionProxy` is deprecated.   Please116       refer to :class:`.ConnectionEvents`.117    Either or both of the ``execute()`` and ``cursor_execute()``118    may be implemented to intercept compiled statement and119    cursor level executions, e.g.::120        class MyProxy(ConnectionProxy):121            def execute(self, conn, execute, clauseelement,122                        *multiparams, **params):123                print "compiled statement:", clauseelement124                return execute(clauseelement, *multiparams, **params)125            def cursor_execute(self, execute, cursor, statement,126                               parameters, context, executemany):127                print "raw statement:", statement128                return execute(cursor, statement, parameters, context)129    The ``execute`` argument is a function that will fulfill the default130    execution behavior for the operation.  The signature illustrated131    in the example should be used.132    The proxy is installed into an :class:`~sqlalchemy.engine.Engine` via133    the ``proxy`` argument::134        e = create_engine('someurl://', proxy=MyProxy())135    """136    @classmethod137    def _adapt_listener(cls, self, listener):138        def adapt_execute(conn, clauseelement, multiparams, params):139            def execute_wrapper(clauseelement, *multiparams, **params):140                return clauseelement, multiparams, params141            return listener.execute(conn, execute_wrapper,142                                    clauseelement, *multiparams,143                                    **params)144        event.listen(self, 'before_execute', adapt_execute)145        def adapt_cursor_execute(conn, cursor, statement,146                                 parameters, context, executemany):147            def execute_wrapper(148                cursor,149                statement,150                parameters,151                context,152            ):153                return statement, parameters154            return listener.cursor_execute(155                execute_wrapper,156                cursor,157                statement,158                parameters,159                context,160                executemany,161                )162        event.listen(self, 'before_cursor_execute', adapt_cursor_execute)163        def do_nothing_callback(*arg, **kw):164            pass165        def adapt_listener(fn):166            def go(conn, *arg, **kw):167                fn(conn, do_nothing_callback, *arg, **kw)168            return util.update_wrapper(go, fn)169        event.listen(self, 'begin', adapt_listener(listener.begin))170        event.listen(self, 'rollback',171                     adapt_listener(listener.rollback))172        event.listen(self, 'commit', adapt_listener(listener.commit))173        event.listen(self, 'savepoint',174                     adapt_listener(listener.savepoint))175        event.listen(self, 'rollback_savepoint',176                     adapt_listener(listener.rollback_savepoint))177        event.listen(self, 'release_savepoint',178                     adapt_listener(listener.release_savepoint))179        event.listen(self, 'begin_twophase',180                     adapt_listener(listener.begin_twophase))181        event.listen(self, 'prepare_twophase',182                     adapt_listener(listener.prepare_twophase))183        event.listen(self, 'rollback_twophase',184                     adapt_listener(listener.rollback_twophase))185        event.listen(self, 'commit_twophase',186                     adapt_listener(listener.commit_twophase))187    def execute(self, conn, execute, clauseelement, *multiparams, **params):188        """Intercept high level execute() events."""189        return execute(clauseelement, *multiparams, **params)190    def cursor_execute(self, execute, cursor, statement, parameters,191                       context, executemany):192        """Intercept low-level cursor execute() events."""193        return execute(cursor, statement, parameters, context)194    def begin(self, conn, begin):195        """Intercept begin() events."""196        return begin()197    def rollback(self, conn, rollback):198        """Intercept rollback() events."""199        return rollback()200    def commit(self, conn, commit):...

events.py

Source:events.py

...43        self.kwargs = kwargs44        self.removeMethod = removeMethod45        self.removeMethodArgs = (self,)46    def handleEvent(self, event):47        self.listener(event, *self.args, **self.kwargs)48    def unlisten(self):49        self.removeMethod(*self.removeMethodArgs)50def addEventListener(target, eventInterface, event, listener, *args, **kwargs):51    """52    Adds an event listener to `target`.53    :param target: an object that supports listening to the events of the given54                   type (the add*Listener methods must be inherited from a Java55                   class so that autodetection will work)56    :param eventInterface: the interface that the listener wrapper has to57                           implement (e.g. :class:`java.awt.MouseListener`)58    :param event: name(s) of the event(s) to listen for (e.g. "mouseClicked")59    :param listener: callable that is called with ``(event, *args, **kwargs)``60                     when the event is fired61    :type eventInterface: Java interface...

domstorage.py

Source:domstorage.py

...84        """85        event_name = "DOMStorage.domStorageItemAdded"86        if listener is None:87            future = self.client.loop.create_future()88            def _listener(event: Optional[Dict] = None) -> None:89                future.set_result(event)90            self.client.once(event_name, _listener)91            return future92        self.client.on(event_name, listener)93        return lambda: self.client.remove_listener(event_name, listener)94    def domStorageItemRemoved(95        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None96    ) -> Any:97        """98        See `https://chromedevtools.github.io/devtools-protocol/tot/DOMStorage#event-domStorageItemRemoved`99        :param listener: Optional listener function100        :return: If a listener was supplied the return value is a callable that101        will remove the supplied listener otherwise a future that resolves102        with the value of the event103        """104        event_name = "DOMStorage.domStorageItemRemoved"105        if listener is None:106            future = self.client.loop.create_future()107            def _listener(event: Optional[Dict] = None) -> None:108                future.set_result(event)109            self.client.once(event_name, _listener)110            return future111        self.client.on(event_name, listener)112        return lambda: self.client.remove_listener(event_name, listener)113    def domStorageItemUpdated(114        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None115    ) -> Any:116        """117        See `https://chromedevtools.github.io/devtools-protocol/tot/DOMStorage#event-domStorageItemUpdated`118        :param listener: Optional listener function119        :return: If a listener was supplied the return value is a callable that120        will remove the supplied listener otherwise a future that resolves121        with the value of the event122        """123        event_name = "DOMStorage.domStorageItemUpdated"124        if listener is None:125            future = self.client.loop.create_future()126            def _listener(event: Optional[Dict] = None) -> None:127                future.set_result(event)128            self.client.once(event_name, _listener)129            return future130        self.client.on(event_name, listener)131        return lambda: self.client.remove_listener(event_name, listener)132    def domStorageItemsCleared(133        self, listener: Optional[Callable[[Dict[str, Any]], Any]] = None134    ) -> Any:135        """136        See `https://chromedevtools.github.io/devtools-protocol/tot/DOMStorage#event-domStorageItemsCleared`137        :param listener: Optional listener function138        :return: If a listener was supplied the return value is a callable that139        will remove the supplied listener otherwise a future that resolves140        with the value of the event141        """142        event_name = "DOMStorage.domStorageItemsCleared"143        if listener is None:144            future = self.client.loop.create_future()145            def _listener(event: Optional[Dict] = None) -> None:146                future.set_result(event)147            self.client.once(event_name, _listener)148            return future149        self.client.on(event_name, listener)...

py4j_signals_test.py

Source:py4j_signals_test.py

1# -*- coding: UTF-8 -*-2from __future__ import unicode_literals, absolute_import3from collections import defaultdict4import unittest5from py4j.java_gateway import (6    server_connection_started, server_connection_stopped,7    server_started, server_stopped, pre_server_shutdown, post_server_shutdown,8    JavaGateway, GatewayParameters, CallbackServerParameters)9from py4j.clientserver import (10    ClientServer, JavaParameters, PythonParameters)11from py4j.tests.client_server_test import (12    clientserver_example_app_process)13from py4j.tests.java_callback_test import (14    IHelloImpl, gateway_example_app_process)15from py4j.tests.py4j_callback_recursive_example import (16    HelloState)17class MockListener(object):18    def __init__(self, test_case):19        self.test_case = test_case20        self.received = defaultdict(int)21    def started(self, sender, **kwargs):22        self.test_case.assertTrue(kwargs["server"] is not None)23        self.received["started"] += 124    def connection_started(self, sender, **kwargs):25        self.test_case.assertTrue(kwargs["connection"] is not None)26        self.received["connection_started"] += 127    def connection_stopped(self, sender, **kwargs):28        self.test_case.assertTrue(kwargs["connection"] is not None)29        self.received["connection_stopped"] += 130    def stopped(self, sender, **kwargs):31        self.test_case.assertTrue(kwargs["server"] is not None)32        self.received["stopped"] += 133    def pre_shutdown(self, sender, **kwargs):34        self.test_case.assertTrue(kwargs["server"] is not None)35        self.received["pre_shutdown"] += 136    def post_shutdown(self, sender, **kwargs):37        self.test_case.assertTrue(kwargs["server"] is not None)38        self.received["post_shutdown"] += 139class JavaGatewayTest(unittest.TestCase):40    def test_all_regular_signals_auto_start(self):41        listener = MockListener(self)42        with gateway_example_app_process(None):43            server_started.connect(listener.started)44            gateway = JavaGateway(45                gateway_parameters=GatewayParameters(),46                callback_server_parameters=CallbackServerParameters())47            server_stopped.connect(48                listener.stopped, sender=gateway.get_callback_server())49            server_connection_started.connect(50                listener.connection_started,51                sender=gateway.get_callback_server())52            server_connection_stopped.connect(53                listener.connection_stopped,54                sender=gateway.get_callback_server())55            pre_server_shutdown.connect(56                listener.pre_shutdown, sender=gateway.get_callback_server())57            post_server_shutdown.connect(58                listener.post_shutdown, sender=gateway.get_callback_server())59            example = gateway.entry_point.getNewExample()60            impl = IHelloImpl()61            self.assertEqual("This is Hello!", example.callHello(impl))62            gateway.shutdown()63        self.assertEqual(1, listener.received["started"])64        self.assertEqual(1, listener.received["stopped"])65        self.assertEqual(1, listener.received["pre_shutdown"])66        self.assertEqual(1, listener.received["post_shutdown"])67        self.assertEqual(1, listener.received["connection_started"])68        self.assertEqual(1, listener.received["connection_stopped"])69class ClientServerTest(unittest.TestCase):70    def test_all_regular_signals(self):71        listener = MockListener(self)72        server_started.connect(listener.started)73        hello_state = HelloState()74        client_server = ClientServer(75            JavaParameters(), PythonParameters(), hello_state)76        server_stopped.connect(77            listener.stopped, sender=client_server.get_callback_server())78        server_connection_started.connect(79            listener.connection_started,80            sender=client_server.get_callback_server())81        server_connection_stopped.connect(82            listener.connection_stopped,83            sender=client_server.get_callback_server())84        pre_server_shutdown.connect(85            listener.pre_shutdown, sender=client_server.get_callback_server())86        post_server_shutdown.connect(87            listener.post_shutdown, sender=client_server.get_callback_server())88        with clientserver_example_app_process(True):89            client_server.shutdown()90        self.assertEqual(1, listener.received["started"])91        self.assertEqual(1, listener.received["stopped"])92        self.assertEqual(1, listener.received["pre_shutdown"])93        self.assertEqual(1, listener.received["post_shutdown"])94        self.assertEqual(1, listener.received["connection_started"])95        self.assertEqual(1, listener.received["connection_stopped"])96    def test_signals_started_from_python(self):97        listener = MockListener(self)98        with clientserver_example_app_process():99            server_started.connect(listener.started)100            client_server = ClientServer(101                JavaParameters(), PythonParameters())102            example = client_server.entry_point.getNewExample()103            impl = IHelloImpl()104            self.assertEqual("This is Hello!", example.callHello(impl))105            server_stopped.connect(106                listener.stopped, sender=client_server.get_callback_server())107            server_connection_started.connect(108                listener.connection_started,109                sender=client_server.get_callback_server())110            server_connection_stopped.connect(111                listener.connection_stopped,112                sender=client_server.get_callback_server())113            pre_server_shutdown.connect(114                listener.pre_shutdown,115                sender=client_server.get_callback_server())116            post_server_shutdown.connect(117                listener.post_shutdown,118                sender=client_server.get_callback_server())119            client_server.shutdown()120        self.assertEqual(1, listener.received["started"])121        self.assertEqual(1, listener.received["stopped"])122        self.assertEqual(1, listener.received["pre_shutdown"])123        self.assertEqual(1, listener.received["post_shutdown"])124        # Connection initiated from JavaClient, so no signal sent.125        self.assertEqual(0, listener.received["connection_started"])...

async_listener_unittest.py

Source:async_listener_unittest.py

1#!/usr/bin/python2# Copyright 2017 The Chromium OS Authors. All rights reserved.3# Use of this source code is governed by a BSD-style license that can be4# found in the LICENSE file.5import os6import shutil7import socket8import tempfile9import unittest10from multiprocessing import connection11import common12from autotest_lib.client.common_lib import error13from autotest_lib.site_utils.lxc import unittest_setup14from autotest_lib.site_utils.lxc.container_pool import async_listener15from autotest_lib.site_utils.lxc.container_pool import unittest_client16# Namespace object for parsing cmd line options.17options = None18class AsyncListenerTests(unittest.TestCase):19    """Unit tests for the AsyncListener class."""20    def setUp(self):21        self.tmpdir = tempfile.mkdtemp()22        self.address = os.path.join(self.tmpdir, 'socket')23        self.authkey = 'foo'24        self.listener = async_listener.AsyncListener(self.address)25    def tearDown(self):26        self.listener.close()27        shutil.rmtree(self.tmpdir)28    def testInvalidAddresses(self):29        """Verifies that invalid socket paths raise errors."""30        with self.assertRaises(socket.error):31            invalid_address = os.path.join(self.tmpdir, 'foo', 'socket')32            async_listener.AsyncListener(invalid_address)33    def testAddressConflict(self):34        """Verifies that address conflicts raise errors."""35        with self.assertRaises(socket.error):36            async_listener.AsyncListener(self.address)37    def testInetAddresses(self):38        """Verifies that inet addresses raise errors."""39        with self.assertRaises(TypeError):40            async_listener.AsyncListener(('127.0.0.1', 0))41    def testStartStop(self):42        """Tests starting and stopping the async listener."""43        try:44            self.assertFalse(self.listener.is_running())45            self.listener.start()46            self.assertTrue(self.listener.is_running())47            # Establish a connection to verify that the listener is actually48            # alive.49            host, client = self._make_connection()50            client.close()51            host.close()52            self.assertTrue(self.listener.stop())53            self.assertFalse(self.listener.is_running())54        except:55            self.fail(error.format_error())56    def testStop_notRunning(self):57        """Tests stopping the async listener when it's not running."""58        try:59            self.assertFalse(self.listener.is_running())60            # Verify that calling stop just returns False when the listener61            # isn't running.62            self.assertFalse(self.listener.stop())63        except:64            self.fail(error.format_error())65    def testConnection(self):66        """Tests starting a connection."""67        self.listener.start()68        # Verify no pending connections.69        self.assertIsNone(self.listener.get_connection(timeout=1))70        # Start a client connection, verify that the listener now returns a71        # pending connection.  Keep a reference to the client, so it doesn't get72        # GC'd while the test is in progress.73        _unused = connection.Client(self.address)74        self.assertIsNotNone(self.listener.get_connection(timeout=1))75    def testHostToClientComms(self):76        """Tests that client/host connections can exchange data."""77        self.listener.start()78        host, client = self._make_connection()79        # Send a message on the host side socket, check that the client receives80        # it.81        msg = 'quickly moving jedi zap a brown sith fox'82        host.send(msg)83        received_msg = client.recv()84        self.assertEqual(msg, received_msg)85    def testMultipleConnections(self):86        """Tests for correct pairing of multiple host/client connections."""87        self.listener.start()88        # Create two sets of host/client connections.  Verify that messages sent89        # on each client are received by the correct respective host.90        host0, client0 = self._make_connection()91        host1, client1 = self._make_connection()92        msg0 = 'a large fawn jumped quickly over white zinc boxes'93        msg1 = 'six big devils from japan quickly forgot how to waltz'94        client0.send(msg0)95        client1.send(msg1)96        received_msg0 = host0.recv()97        received_msg1 = host1.recv()98        self.assertEquals(msg0, received_msg0)99        self.assertEquals(msg1, received_msg1)100    def _make_connection(self):101        """Creates pairs of host/client connections.102        @return: A host, client connection pair.103        """104        client = unittest_client.connect(self.address)105        host = self.listener.get_connection(timeout=1)106        return host, client107if __name__ == '__main__':108    unittest_setup.setup()...

test_listener.py

Source:test_listener.py

...91                (8080, 80, 'HTTP', 'HTTP'),92            ]93        )94class TestListenerGetItem(unittest.TestCase):95    def test_getitem_for_http_listener(self):96        listener = Listener(load_balancer_port=80,97                            instance_port=80,98                            protocol='HTTP',99                            instance_protocol='HTTP')100        self.assertEqual(listener[0], 80)101        self.assertEqual(listener[1], 80)102        self.assertEqual(listener[2], 'HTTP')103        self.assertEqual(listener[3], 'HTTP')104    def test_getitem_for_https_listener(self):105        listener = Listener(load_balancer_port=443,106                            instance_port=80,107                            protocol='HTTPS',108                            instance_protocol='HTTP',109                            ssl_certificate_id='look_at_me_im_an_arn')110        self.assertEqual(listener[0], 443)111        self.assertEqual(listener[1], 80)112        self.assertEqual(listener[2], 'HTTPS')113        self.assertEqual(listener[3], 'HTTP')114        self.assertEqual(listener[4], 'look_at_me_im_an_arn')115if __name__ == '__main__':...

inspector.py

Source:inspector.py

...42        """43        event_name = "Inspector.detached"44        if listener is None:45            future = self.client.loop.create_future()46            def _listener(event: Optional[Dict] = None) -> None:47                future.set_result(event)48            self.client.once(event_name, _listener)49            return future50        self.client.on(event_name, listener)51        return lambda: self.client.remove_listener(event_name, listener)52    def targetCrashed(self, listener: Optional[Callable[[Any], Any]] = None) -> Any:53        """54        Fired when debugging target has crashed55        See `https://chromedevtools.github.io/devtools-protocol/tot/Inspector#event-targetCrashed`56        :param listener: Optional listener function57        :return: If a listener was supplied the return value is a callable that58        will remove the supplied listener otherwise a future that resolves59        with the value of the event60        """61        event_name = "Inspector.targetCrashed"62        if listener is None:63            future = self.client.loop.create_future()64            def _listener(event: Optional[Dict] = None) -> None:65                future.set_result(event)66            self.client.once(event_name, _listener)67            return future68        self.client.on(event_name, listener)69        return lambda: self.client.remove_listener(event_name, listener)70    def targetReloadedAfterCrash(71        self, listener: Optional[Callable[[Any], Any]] = None72    ) -> Any:73        """74        Fired when debugging target has reloaded after crash75        See `https://chromedevtools.github.io/devtools-protocol/tot/Inspector#event-targetReloadedAfterCrash`76        :param listener: Optional listener function77        :return: If a listener was supplied the return value is a callable that78        will remove the supplied listener otherwise a future that resolves79        with the value of the event80        """81        event_name = "Inspector.targetReloadedAfterCrash"82        if listener is None:83            future = self.client.loop.create_future()84            def _listener(event: Optional[Dict] = None) -> None:85                future.set_result(event)86            self.client.once(event_name, _listener)87            return future88        self.client.on(event_name, listener)...

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.