KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use onConnect method in Playwright Python

Best Python code snippet using playwright-python

Server.py

Source:Server.py

1import socket2from datetime import datetime3import socket, time, threading, time, sys, json4'''5TCP Server class6'''7class Server:8    MAX_CLIENTS = 59    PORT_IN_USE_TIMEOUT = 310    def __init__(self, address, port, messageDataType="json", byteOrder="little", sendMostRecent=True):11        '''12        Create a TCP Server object. Call start() to run it.13        @param string address address to run on. eg: '192.168.1.23'14        @param int port port to host on. eg: 300015        @param string messageDataType 'json' or 'string' to auto parse messages. Otherwise will be binary16        @param string byteOrder 'little' or 'big' endian. Other ReliableCommunication scripts use 'little'. But if you are connecting to a different server, they may use big endian numbers for their headers.17        @param bool sendMostRecent (unused) whether to drop messages queued for sending18        '''19        self.port = port20        self.address = address21        self.byteOrder = byteOrder22        self.conn = None23        self.clients = []24        self.sock = None25        self.STOP = False26        self.dataToSend = None27        self.sendMostRecent = sendMostRecent28        self.lock = threading.Lock()29        self.messageDataType = messageDataType30        self.__onmessage_callbacks__ = []31        self.__onconnect_callbacks__ = []32        self.__onclose_callbacks__ = []33        self.thread = threading.Thread(target=self.__accept_clients_loop__, name="Server {} newclient_accept".format(self.port))34        print("[Server "+str(self.port)+"] Initialized.")35    def start(self):36        '''37        Starts the server - begins accepting clients38        will create threads for each client that connects.39        Allows for Server.MAX_CLIENTS number of clients to connect40        '''41        self.thread.start()42    def __accept_clients_loop__(self):43        ''' Constantly listen and accept clients '''44        print("[Server {}] Open for new connections".format(self.port))45        # Constantly look for a connection46        while not self.STOP:47            self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)48            try:49                self.sock.bind((self.address, self.port))50            except:51                print("[Server "+str(self.port)+"] Port already in use")52                self.sock.close()53                self.sock = None54                time.sleep(Server.PORT_IN_USE_TIMEOUT)55                continue56            self.sock.listen(Server.MAX_CLIENTS)57            while not self.STOP:58                # Accept incoming connections59                self.sock.settimeout(3)60                try:61                    conn, client = self.sock.accept()62                    # Create Client object63                    clientObject = Client(client[0], client[1], True, conn,  self.messageDataType, self.byteOrder, self.sendMostRecent)64                    # subscribe to client events65                    clientObject.add_onmessage_callback(self.__onmessage_caller__)66                    clientObject.add_onclose_callback(self.__remove_client__)67                    clientObject.add_onclose_callback(self.__onclose_caller__)68                    self.clients.append(clientObject)69                    # Start listener loop70                    clientObject.listener.start()71                    # Call onConnect subscribers72                    threading.Thread(target=self.__onconnect_caller__, args=(clientObject,), name="Server {} onconnect callbacks".format(self.port)).start()73                except socket.timeout:74                    continue75                except Exception as e:76                    self.stop()77                    raise e78        if (self.sock):79            self.sock.close()80            print("[Server {}] Socket Closed".format(self.port))81    '''82    CALLBACK METHODS83    '''84    def __onmessage_caller__(self, message):85        ''' Calls all of the subscribed listeners whenever a client gets a message '''86        for callback in self.__onmessage_callbacks__:87            callback(message)88    def __onclose_caller__(self, client):89        ''' Calls all of the subscribed onclose listeners whenever a client disconnects '''90        for callback in  self.__onclose_callbacks__:91            callback(client)92    def __onconnect_caller__(self, client):93        ''' Calls all of the subscribed onconnect listeners whenever a client connects '''94        for callback in self.__onconnect_callbacks__:95            callback(client)96    def add_onmessage_callback(self, func):97        '''98        Adds passed function to list of callback functions.99        All functions will be called when server receives a message from any of the clients100        function will be called in the order they are added101        @param func the function to add. eg: myServer.add_onmessage_callback(dosomething)102        '''103        self.__onmessage_callbacks__.append(func)104    def add_onclose_callback(self, func):105        '''106        Adds passed function to list of callback functions.107        All functions will be called when any client disconnects.108        functions will be called in the order they are added109        @param func the function to add. eg: myServer.add_onclose_callback(dosomething)110        '''111        self.__onclose_callbacks__.append(func)112    def add_onconnect_callback(self, func):113        '''114        Adds passed function to list of callback functions.115        All functions will be called when any client connects.116        functions will be called in the order they are added117        @param func the function to add. eg: myServer.add_onclose_callback(dosomething)118        '''119        self.__onconnect_callbacks__.append(func)120    def remove_onmessage_callback(self, func=None, index=0):121        '''122        Removes passed function OR index from list of callbacks123        @param func (optional) the function to add. If None, will use 'index'124        @param index the index of the function to remove. 'func' must be None.125        '''126        Server.__remove_func_from_list__(self.__onmessage_callbacks__, func, index)127    def remove_onclose_callback(self, func=None, index=0):128        '''129        Removes passed function OR index from list of callbacks130        @param func (optional) the function to add. If None, will use 'index'131        @param index the index of the function to remove. 'func' must be None.132        '''133        Server.__remove_func_from_list__(self.__onclose_callbacks__, func, index)134    def remove_onconnect_callback(self, func=None, index=0):135        '''136        Removes passed function OR index from list of callbacks137        @param func (optional) the function to add. If None, will use 'index'138        @param index the index of the function to remove. 'func' must be None.139        '''140        Server  .__remove_func_from_list__(self.__onconnect_callbacks__, func, index)141    def __remove_client__(self, client):142        ''' removes client from server's list of clients '''143        self.clients.remove(client)144    @staticmethod145    def __remove_func_from_list__(listToModify, func=None, index=0):146        ''' logic to remove either a function or index from a list '''147        if func is not None:148            if func in listToModify:149                listToModify.remove(func)150                return True151            else:152                return False153        if 0 < index < len(listToModify):154            listToModify.pop(index)155            return True156        else:157            return False158    '''159    SENDING METHODS160    '''161    def broadcast(self, data):162        '''163        Send a message to all clients connected to the server164        @param data the message to send - either json, string, or binary (can be different from what the server parses)165        '''166        for client in self.clients:167            threading.Thread(target=client.send, args=(data,), name="Client {}:{} send".format(client.address, client.port)).start()168    def sendTo(self, data, server_client=0):169        '''170        Send a message to a particular client171        @param data to message to send - either json, string, or binary172        @param server_client can be client index or the client object you wish to send to173        '''174        if type(server_client) is type(0):175            if server_client < len(self.clients):176                self.clients[server_client].send(data)177                return178            else:179                raise IndexError("Passed index {} but only {} clients exist".format(server_client, len(self.clients)))180        if type(server_client) is type(Client):181            server_client.send(data)182    def __del__(self):183        self.stop()184    def stop(self):185        '''186        Stops the server. Disconnects clients. Ends all threads.187        Use this to cleanly close everything.188        '''189        if not self.STOP:190            self.STOP = True191            for client in self.clients:192                client.conn.shutdown(1)193                client.close()194            print("[Server {}] Stopping... ({} second timeout)".format(self.port, Server.PORT_IN_USE_TIMEOUT))195'''196TCP Client class197Instantiating and calling connect() starts a TCP client connection to the passed address and port198Also used by Server199'''200class Client:201    def __init__(self, address, port, controlledByServer=False, connection=None, messageDataType="json", byteOrder="little", sendMostRecent=False, autoReconnect=False):202        '''203        Creates an object for threaded management of a TCP connection with a server. (can also be used by a server to manage clients)204        call myClient.connect() to establish connection with server and begin receiving messages205        @param string address the device address to connect to. eg: "192.168.1.55"206        @param int port the server port to connect to. eg: 6000207        @param bool controlledByServer wether the instance is being managed by a server. False by default208        @param Socket connection if controlled by a server, this is the socket connection object to a client. None by default209        @param string messageDataType 'json' or 'string' to automatically parse incoming messages as either of these. Otherwise will use binary210        @param string byteOrder 'little' or 'big' endian depending on the headers being used.211        @param bool sendMostRecent whether to drop accumulated packets and only send the most recent messages212        @param bool autoReconnect automatically reconnect to the server if connection is lost. Forced to False if controlled by server213        '''214        # connection and message passing type215        self.address = address216        self.port = port217        self.conn = connection218        self.messageType = messageDataType219        self.byteOrder = byteOrder220        # state management221        self.STOP = False222        self.listener = None223        # listeners224        self.onMessage = []225        self.onClose = []226        self.onConnect = []227        # options228        self.autoReconnect = False229        self.__can_connect__ = False230        self.sendMostRecent = sendMostRecent231        if self.conn is None or controlledByServer is False:232            self.__can_connect__ = True233            self.autoReconnect = autoReconnect234        else:235            self.listener = threading.Thread(target=self.__listen__, name="Client of {}:{} listener".format(self.address, self.port))236    '''237    CONTROL METHODS238    '''239    def connect(self):240        '''241        Starts connection with server.242        '''243        if self.__can_connect__:244            self.STOP = False245            self.conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)246            self.conn.connect((self.address, self.port))247            self.listener = threading.Thread(target=self.__listen__, name="Client of {}:{} listener".format(self.address, self.port))248            self.listener.start()249            threading.Thread(target=self.__onconnect_caller__, name="Client {}:{} onconnect callbacks".format(self.address, self.port)).start()250        else:251            raise Exception("Cannot establish client connection inside a server")252    def __listen__(self):253        ''' Constantly listens for messages, automatically parses as json or string, and starts callback threads '''254        while not self.STOP:255            if (self.conn):256                try:257                    # Get Message Header258                    self.conn.settimeout(3)259                    datalen = int.from_bytes(self.conn.recv(4), self.byteOrder)260                    data = self.conn.recv(datalen)261                    # Parse Data into a message based self.messageType262                    msg = data263                    if self.messageType == "json":264                        msg = Client.parseJson(data)265                    elif self.messageType == "string":266                        msg = msg.decode("utf-8")267                    # Callback268                    threading.Thread(target=self.__onmessage_caller__, args=(msg,), name="Client {}:{} onmessage_callbacks".format(self.address, self.port)).start()269                except socket.timeout:270                    continue271                except ConnectionResetError:272                    self.close()273                    continue274                except ConnectionAbortedError:275                    self.close()276                    continue277                except:278                    print("[Client {}:{}] Exception in read loop \n\t{}".format(self.address, self.port, sys.exc_info()))279                    self.close()280                    continue281            else:282                self.close()283        # Close out284        self.conn.close()285    def send(self, message):286        ''' Sends a message '''287        # TODO: make this into a queue288        as_bytes = None289        as_string = None290        # Convert to bytes291        if type(message) is type({}):292            as_string = json.dumps(message)293        if type(message) is type(""):294            as_string = message295        if type(message) is type(b''):296            as_bytes = message297        if as_string is not None:298            as_bytes = as_string.encode("utf-8")299        # Add Header300        if (self.conn is not None and not self.STOP):301            # Get Message Length302            messageLength = (len(as_bytes)).to_bytes(4, byteorder=self.byteOrder, signed=False)303            # SEND304            try:305                self.conn.send(bytes(messageLength)) # 4 bytes with the size of the image306                self.conn.send(bytes(as_bytes)) # If throwing error, check if numpy array is converting to byte array. May need to call bytes(data.tobytes()) ERROR: only integer arrays with one element can be...307            except TypeError:308                tb = sys.exc_info()[2]309                print("[Client {}:{}] Exception sending data {}\n\t{} {}".format(self.address, self.port, sys.exc_info()[1], tb.tb_frame.f_code.co_filename, tb.tb_lineno))310            # except ConnectionAbortedError:311            #     self.close()312            # except ConnectionResetError:313            #     self.close()314            # except BrokenPipeError:315            #     self.close()316            # except OSError:317            #     self.close()318            except:319                self.close()320    def close(self):321        if not self.STOP:322            self.STOP = True323            # Call callbacks324            threading.Thread(target=self.__onclose_caller__, name="Client {}:{} close callbacks".format(self.address, self.port)).start()325            # Autoreconnect326            if (self.autoReconnect):327                time.sleep(1)328                self.connect()329    '''330    CALLBACK METHODS331    '''332    def __onmessage_caller__(self, message):333        ''' Calls all of the subscribed listeners whenever a client gets a message '''334        for callback in self.onMessage:335            callback(message)336    def __onclose_caller__(self):337        ''' Calls all of the subscribed listeners whenever disconnected from server '''338        for callback in self.onClose:339            callback(self)340    def __onconnect_caller__(self):341        ''' Calls all subscribers when (re)connected to server '''342        for callback in self.onConnect:343            callback(self)344    def add_onmessage_callback(self, func):345        '''346        Adds passed function to list of callback functions.347        All functions will be called when client receives a message from the server348        function will be called in the order they are added349        @param func the function to add. eg: myClient.add_onmessage_callback(dosomething)350        '''351        self.onMessage.append(func)352    def add_onclose_callback(self, func):353        '''354        Adds passed function to list of callback functions.355        All functions will be called when disconnected from server.356        functions will be called in the order they are added357        @param func the function to add. eg: myClient.add_onclose_callback(dosomething)358        '''359        self.onClose.append(func)360    def add_onconnect_callback(self, func):361        '''362        Adds passed function to list of callback functions.363        All functions will be called when connection with server is established or re-established.364        functions will be called in the order they are added365        @param func the function to add. eg: myClient.add_onclose_callback(dosomething)366        '''367        self.onConnect.append(func)368    def remove_onmessage_callback(self, func=None, index=0):369        '''370        Removes passed function OR index from list of callbacks371        @param func (optional) the function to add. If None, will use 'index'372        @param index the index of the function to remove. 'func' must be None.373        '''374        Client.__remove_func_from_list__(self.onMessage, func, index)375    def remove_onclose_callback(self, func=None, index=0):376        '''377        Removes passed function OR index from list of callbacks378        @param func (optional) the function to add. If None, will use 'index'379        @param index the index of the function to remove. 'func' must be None.380        '''381        Client.__remove_func_from_list__(self.onClose, func, index)382    def remove_onconnect_callback(self, func=None, index=0):383        '''384        Removes passed function OR index from list of callbacks385        @param func (optional) the function to add. If None, will use 'index'386        @param index the index of the function to remove. 'func' must be None.387        '''388        Client.__remove_func_from_list__(self.onConnect, func, index)389    '''390    HELPER391    '''392    @staticmethod393    def __remove_func_from_list__(listToModify, func=None, index=0):394        ''' logic to remove either a function or index from a list '''395        if func is not None:396            if func in listToModify:397                listToModify.remove(func)398                return True399            else:400                return False401        if 0 < index < len(listToModify):402            listToModify.pop(index)403            return True404        else:405            return False406    @staticmethod407    def parseJson(data):408        data = data.decode("utf-8")409        msg = json.loads(data)...

test_ptcp.py

Source:test_ptcp.py

1# -*- test-case-name: vertex.test.test_ptcp -*-2import random, os3from twisted.internet import reactor, protocol, defer, error4from twisted.trial import unittest5from vertex import ptcp6def reallyLossy(method):7    r = random.Random()8    r.seed(42)9    def worseMethod(*a, **kw):10        if r.choice([True, True, False]):11            method(*a, **kw)12    return worseMethod13def insufficientTransmitter(method,  mtu):14    def worseMethod(bytes, addr):15        method(bytes[:mtu], addr)16    return worseMethod17class TestProtocol(protocol.Protocol):18    buffer = None19    def __init__(self):20        self.onConnect = defer.Deferred()21        self.onDisconn = defer.Deferred()22        self._waiting = None23        self.buffer = []24    def connectionMade(self):25        self.onConnect.callback(None)26    def connectionLost(self, reason):27        self.onDisconn.callback(None)28    def gotBytes(self, bytes):29        assert self._waiting is None30        if ''.join(self.buffer) == bytes:31            return defer.succeed(None)32        self._waiting = (defer.Deferred(), bytes)33        return self._waiting[0]34    def dataReceived(self, bytes):35        self.buffer.append(bytes)36        if self._waiting is not None:37            bytes = ''.join(self.buffer)38            if not self._waiting[1].startswith(bytes):39                x = len(os.path.commonprefix([bytes, self._waiting[1]]))40                print x41                print 'it goes wrong starting with', repr(bytes[x:x+100]), repr(self._waiting[1][x:x+100])42            if bytes == self._waiting[1]:43                self._waiting[0].callback(None)44                self._waiting = None45class Django(protocol.ClientFactory):46    def __init__(self):47        self.onConnect = defer.Deferred()48    def buildProtocol(self, addr):49        p = protocol.ClientFactory.buildProtocol(self, addr)50        self.onConnect.callback(p)51        return p52    def clientConnectionFailed(self, conn, err):53        self.onConnect.errback(err)54class ConnectedPTCPMixin:55    serverPort = None56    def setUpForATest(self,57                      ServerProtocol=TestProtocol, ClientProtocol=TestProtocol):58        serverProto = ServerProtocol()59        clientProto = ClientProtocol()60        self.serverProto = serverProto61        self.clientProto = clientProto62        sf = protocol.ServerFactory()63        sf.protocol = lambda: serverProto64        cf = Django()65        cf.protocol = lambda: clientProto66        serverTransport = ptcp.PTCP(sf)67        clientTransport = ptcp.PTCP(None)68        self.serverTransport = serverTransport69        self.clientTransport = clientTransport70        serverPort = reactor.listenUDP(0, serverTransport)71        clientPort = reactor.listenUDP(0, clientTransport)72        self.clientPort = clientPort73        self.serverPort = serverPort74        return (75            serverProto, clientProto,76            sf, cf,77            serverTransport, clientTransport,78            serverPort, clientPort79            )80    def tearDown(self):81        td = []82        for ptcp in (self.serverTransport, self.clientTransport):83            td.append(ptcp.waitForAllConnectionsToClose())84        d = defer.DeferredList(td)85        return d86class TestProducerProtocol(protocol.Protocol):87    NUM_WRITES = 3288    WRITE_SIZE = 3289    def __init__(self):90        self.onConnect = defer.Deferred()91        self.onPaused = defer.Deferred()92    def connectionMade(self):93        self.onConnect.callback(None)94        self.count = -195        self.transport.registerProducer(self, False)96    def pauseProducing(self):97        if self.onPaused is not None:98            self.onPaused.callback(None)99            self.onPaused = None100    def resumeProducing(self):101        self.count += 1102        if self.count < self.NUM_WRITES:103            bytes = chr(self.count) * self.WRITE_SIZE104            # print 'Issuing a write', len(bytes)105            self.transport.write(bytes)106            if self.count == self.NUM_WRITES - 1:107                # Last time through, intentionally drop the connection before108                # the buffer is empty to ensure we handle this case properly.109                # print 'Disconnecting'110                self.transport.loseConnection()111        else:112            # print 'Unregistering'113            self.transport.unregisterProducer()114class PTCPTransportTestCase(ConnectedPTCPMixin, unittest.TestCase):115    def setUp(self):116        """117        I have no idea why one of these values is divided by 10 and the118        other is multiplied by 10.  -exarkun119        """120        self.patch(121            ptcp.PTCPConnection, '_retransmitTimeout',122            ptcp.PTCPConnection._retransmitTimeout / 10)123        self.patch(124            ptcp.PTCPPacket, 'retransmitCount',125            ptcp.PTCPPacket.retransmitCount * 10)126    def xtestWhoAmI(self):127        (serverProto, clientProto,128         sf, cf,129         serverTransport, clientTransport,130         serverPort, clientPort) = self.setUpForATest()131        def gotAddress(results):132            (serverSuccess, serverAddress), (clientSuccess, clientAddress) = results133            self.failUnless(serverSuccess)134            self.failUnless(clientSuccess)135            self.assertEquals(serverAddress[1], serverPort.getHost().port)136            self.assertEquals(clientAddress[1], clientPort.getHost().port)137        def connectionsMade(ignored):138            return defer.DeferredList([serverProto.transport.whoami(), clientProto.transport.whoami()]).addCallback(gotAddress)139        clientConnID = clientTransport.connect(cf, '127.0.0.1', serverPort.getHost().port)140        return defer.DeferredList([serverProto.onConnect, clientProto.onConnect]).addCallback(connectionsMade)141    #testWhoAmI.skip = 'arglebargle'142    def testVerySimpleConnection(self):143        (serverProto, clientProto,144         sf, cf,145         serverTransport, clientTransport,146         serverPort, clientPort) = self.setUpForATest()147        clientConnID = clientTransport.connect(cf, '127.0.0.1', serverPort.getHost().port)148        def sendSomeBytes(ignored, n=10, server=False):149            if n:150                bytes = 'not a lot of bytes' * 1000151                if server:152                    serverProto.transport.write(bytes)153                else:154                    clientProto.transport.write(bytes)155                if server:156                    clientProto.buffer = []157                    d = clientProto.gotBytes(bytes)158                else:159                    serverProto.buffer = []160                    d = serverProto.gotBytes(bytes)161                return d.addCallback(sendSomeBytes, n - 1, not server)162        def loseConnections(ignored):163            serverProto.transport.loseConnection()164            clientProto.transport.loseConnection()165            return defer.DeferredList([166                    serverProto.onDisconn,167                    clientProto.onDisconn168                    ])169        dl = defer.DeferredList([serverProto.onConnect, clientProto.onConnect])170        dl.addCallback(sendSomeBytes)171        dl.addCallback(loseConnections)172        return dl173    def testProducerConsumer(self):174        (serverProto, clientProto,175         sf, cf,176         serverTransport, clientTransport,177         serverPort, clientPort) = self.setUpForATest(178            ServerProtocol=TestProducerProtocol)179        def disconnected(ignored):180            self.assertEquals(181                ''.join(clientProto.buffer),182                ''.join([chr(n) * serverProto.WRITE_SIZE183                         for n in range(serverProto.NUM_WRITES)]))184        clientConnID = clientTransport.connect(cf, '127.0.0.1', serverPort.getHost().port)185        return clientProto.onDisconn.addCallback(disconnected)186    def testTransportProducer(self):187        (serverProto, clientProto,188         sf, cf,189         serverTransport, clientTransport,190         serverPort, clientPort) = self.setUpForATest()191        resumed = []192        def resumeProducing():193            resumed.append(True)194            clientProto.transport.resumeProducing()195        def cbBytes(ignored):196            self.failUnless(resumed)197            clientProto.transport.loseConnection()198        def cbConnect(ignored):199            BYTES = 'Here are bytes'200            clientProto.transport.pauseProducing()201            serverProto.transport.write(BYTES)202            reactor.callLater(2, resumeProducing)203            return clientProto.gotBytes(BYTES).addCallback(cbBytes)204        clientConnID = clientTransport.connect(cf, '127.0.0.1', serverPort.getHost().port)205        connD = defer.DeferredList([clientProto.onConnect, serverProto.onConnect])206        connD.addCallback(cbConnect)207        return connD208    def testTransportProducerProtocolProducer(self):209        (serverProto, clientProto,210         sf, cf,211         serverTransport, clientTransport,212         serverPort, clientPort) = self.setUpForATest(213            ServerProtocol=TestProducerProtocol)214        paused = []215        def cbPaused(ignored):216            # print 'Paused'217            paused.append(True)218            # print 'RESUMING', clientProto, clientTransport, clientPort219            clientProto.transport.resumeProducing()220        serverProto.onPaused.addCallback(cbPaused)221        def cbBytes(ignored):222            # print 'Disconnected'223            self.assertEquals(224                ''.join(clientProto.buffer),225                ''.join([chr(n) * serverProto.WRITE_SIZE226                         for n in range(serverProto.NUM_WRITES)]))227        def cbConnect(ignored):228            # The server must write enough to completely fill the outgoing buffer,229            # since our peer isn't ACKing /anything/ and our server waits for230            # writes to be acked before proceeding.231            serverProto.WRITE_SIZE = serverProto.transport.sendWindow * 5232            # print 'Connected'233            # print 'PAUSING CLIENT PROTO', clientProto, clientTransport, clientPort234            clientProto.transport.pauseProducing()235            return clientProto.onDisconn.addCallback(cbBytes)236        clientConnID = clientTransport.connect(cf, '127.0.0.1', serverPort.getHost().port)237        connD = defer.DeferredList([clientProto.onConnect, serverProto.onConnect])238        connD.addCallback(cbConnect)239        return connD240class LossyTransportTestCase(PTCPTransportTestCase):241    def setUpForATest(self, *a, **kw):242        results = PTCPTransportTestCase.setUpForATest(self, *a, **kw)243        results[-2].write = reallyLossy(results[-2].write)244        results[-1].write = reallyLossy(results[-1].write)245        return results246class SmallMTUTransportTestCase(PTCPTransportTestCase):247    def setUpForATest(self, *a, **kw):248        results = PTCPTransportTestCase.setUpForATest(self, *a, **kw)249        results[-2].write = insufficientTransmitter(results[-2].write, 128)250        results[-1].write = insufficientTransmitter(results[-1].write, 128)251        return results252class TimeoutTestCase(ConnectedPTCPMixin, unittest.TestCase):253    def setUp(self):254        """255        Shorten the retransmit timeout so that tests finish more quickly.256        """257        self.patch(258            ptcp.PTCPConnection, '_retransmitTimeout',259            ptcp.PTCPConnection._retransmitTimeout / 10)260    def testConnectTimeout(self):261        (serverProto, clientProto,262         sf, cf,263         serverTransport, clientTransport,264         serverPort, clientPort) = self.setUpForATest()265        clientTransport.sendPacket = lambda *a, **kw: None266        clientConnID = clientTransport.connect(cf, '127.0.0.1', serverPort.getHost().port)267        return cf.onConnect.addBoth(lambda result: result.trap(error.TimeoutError) and None)268    def testDataTimeout(self):269        (serverProto, clientProto,270         sf, cf,271         serverTransport, clientTransport,272         serverPort, clientPort) = self.setUpForATest()273        def cbConnected(ignored):274            serverProto.transport.ptcp.sendPacket = lambda *a, **kw: None275            clientProto.transport.write('Receive this data.')276            serverProto.transport.write('Send this data.') # have to send data277                                                           # or the server will278                                                           # never time out:279                                                           # need a280                                                           # SO_KEEPALIVE281                                                           # option somewhere282            return clientProto.onDisconn283        clientConnID = clientTransport.connect(cf, '127.0.0.1', serverPort.getHost().port)284        d = defer.DeferredList([serverProto.onConnect, clientProto.onConnect])285        d.addCallback(cbConnected)...

gatt_core.py

Source:gatt_core.py

1import sys2import errno3import select as native_select4import functools5import threading6import gevent7from gevent.select import select8# this is hack because the above does not work9from gevent import monkey10monkey.patch_select()11from PyBT.gap import GAP12from PyBT.stack import BTEvent13def needs_connection(func):14    @functools.wraps(func)15    def inner(self, *args):16        if not self.connected:17            raise ConnectionError("This command requires a connection")18        return func(self, *args)19    return inner20class ConnectionError(Exception):21    pass22class SocketHandler(object):23    def __init__(self, conn):24        self.conn = conn25    def dump_gap(self, data):26        if len(data) > 0:27            try:28                gap = GAP()29                gap.decode(data)30                print "GAP: %s" % gap31            except Exception as e:32                print e33                pass34    # Make this look a bit like a thread.35    def run(self):36        # FIXME(richo) Mutex around shared mutable state37        seen = self.conn.seen38        while True:39            try:40                select([self.conn.central.stack], [], [])41            except native_select.error as ex:42                if ex[0] == errno.EINTR:43                    continue44                raise45            event = self.conn.central.stack.handle_data()46            if event.type == BTEvent.SCAN_DATA:47                addr, type, data = event.data48                print ("Saw %s (%s)" %49                       (addr, "public" if type == 0 else "random"))50                if addr in seen:51                    if len(data) > len(seen[addr][1]):52                        seen[addr] = (type, data)53                        self.dump_gap(data)54                else:55                    seen[addr] = (type, data)56                    self.dump_gap(data)57            elif event.type == BTEvent.CONNECTED:58                # FIXME(richo) Mutex59                self.conn.connected = True60                print "Connected!"61                if len(self.conn.onconnect) > 0:62                    print "Running onconnect comands"63                    while self.conn.onconnect():64                        cmd = self.conn.onconnect(0)65                        cmd()66            elif event.type == BTEvent.DISCONNECTED:67                self.conn.connected = False68                print "Disconnected"69            elif event.type == BTEvent.ATT_DATA:70                pkt = event.data71                # ack handle value notification72                if pkt.opcode == 0x1d:73                    self.central.stack.raw_att("\x1e")74                print event75            elif event.type != BTEvent.NONE:76                print event77class Connection(object):78    def __init__(self, central):79        self.connected = False80        self.central = central81        self.seen = {}82        self.onconnect = []83    def start(self):84        self._dispatchSocketHandler()85    def _dispatchSocketHandler(self):86        handler = SocketHandler(self)87        gevent.spawn(handler.run)88    # Public command functions89    def scan(self, arg):90        if arg == 'on':91            self.central.stack.scan()92        else:93            self.central.stack.scan_stop()94    def connect(self, addr, kind=None):95        if kind is None:96            # We may have inferred it's kind from seeing it advertising97            kind = self.seen.get(addr, (None,))[0]98        if kind is None:99            print "Error: please give address type"100        else:101            print "Connecting.."102            self.central.stack.connect(addr, kind)103    def quit(self):104        # FIXME(richo) Actually do some cleanup, try to put the hci device back105        # together106        sys.exit(0)107    @needs_connection108    def write_req(self, handle, value):109        self.central.att.write_req(handle=handle, value=value)110    @needs_connection111    def write_cmd(self, handle, value):112        self.central.att.write_cmd(handle=handle, value=value)113    @needs_connection114    def read(self, handle):115        self.central.att.read(handle)116    def set_interval(self, int_min, int_max):117        self.central.stack.interval_min = int_min118        self.central.stack.interval_max = int_max119    def on_connect(self, thunk):120        self.onconnect.append(thunk)121    def raw(self, cmd):...

test_obniz.py

Source:test_obniz.py

1from .utils import assert_finished, assert_send, receive_json2class TestObniz:3    def test_message(self, obniz):4        targets = ["1234-1231", "1234-1230"]5        obniz.message(targets, "pressed")6        assert_send(7            obniz, [{"message": {"data": "pressed", "to": ["1234-1231", "1234-1230"]}}]8        )9        assert_finished(obniz)10    def test_message_receive(self, obniz, mocker):11        obniz.onmessage = mocker.stub()12        receive_json(13            obniz, [{"message": {"data": "button pressed", "from": "1234-5678"}}]14        )15        assert obniz.onmessage.call_count == 116        assert len(obniz.onmessage.call_args[0]) == 217        assert obniz.onmessage.call_args[0][0] == "button pressed"18        assert obniz.onmessage.call_args[0][1] == "1234-5678"19    def test_message_receive2(self, obniz, mocker):20        obniz.onmessage = mocker.stub()21        receive_json(obniz, [{"message": {"data": [1, 2, 3, 4, 5, 10], "from": None}}])22        assert obniz.onmessage.call_count == 123        assert len(obniz.onmessage.call_args[0]) == 224        assert obniz.onmessage.call_args[0][0] == [1, 2, 3, 4, 5, 10]25        assert obniz.onmessage.call_args[0][1] is None26    def test_reset_on_disconnect(self, obniz):27        obniz.reset_on_disconnect(False)28        assert_send(obniz, [{"ws": {"reset_obniz_on_ws_disconnection": False}}])29        assert_finished(obniz)30    def test_ready(self, obniz, mocker):31        onconnect_stub = mocker.stub()32        def onconnect(x):33            onconnect_stub(x)34        obniz.onconnect = onconnect35        receive_json(obniz, [{'ws': {'ready': True, 'obniz': {'hw': 'obnizb1', 'firmware': '2.0.2'}}}])36        assert onconnect_stub.call_count == 137        assert len(onconnect_stub.call_args[0]) == 138        assert onconnect_stub.call_args[0][0] == obniz39        assert_send(obniz, [{"ws": {"reset_obniz_on_ws_disconnection": True}}])40        assert_finished(obniz)41    def test_warning(self, obniz, mocker):42        obniz.warning = mocker.stub()43        receive_json(obniz, [{"debug": {"warning": {"message": "unknown command"}}}])44        assert obniz.warning.call_count == 145        assert len(obniz.warning.call_args[0]) == 146        assert obniz.warning.call_args[0][0] == {47            "alert": "warning",48            "message": "Warning: unknown command",49        }50        assert_finished(obniz)51    def test_error(self, obniz, mocker):52        error = obniz.error53        obniz.error = mocker.stub()54        receive_json(obniz, [{"debug": {"error": {"message": "voltage down"}}}])55        assert obniz.error.call_count == 156        assert len(obniz.error.call_args[0]) == 157        assert obniz.error.call_args[0][0] == {58            "alert": "error",59            "message": "Error: voltage down",60        }61        obniz.error = error...

Playwright tutorial

LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.

Chapters:

What is Playwright : Playwright is comparatively new but has gained good popularity. Get to know some history of the Playwright with some interesting facts connected with it.
How To Install Playwright : Learn in detail about what basic configuration and dependencies are required for installing Playwright and run a test. Get a step-by-step direction for installing the Playwright automation framework.
Playwright Futuristic Features: Launched in 2020, Playwright gained huge popularity quickly because of some obliging features such as Playwright Test Generator and Inspector, Playwright Reporter, Playwright auto-waiting mechanism and etc. Read up on those features to master Playwright testing.
What is Component Testing: Component testing in Playwright is a unique feature that allows a tester to test a single component of a web application without integrating them with other elements. Learn how to perform Component testing on the Playwright automation framework.
Inputs And Buttons In Playwright: Every website has Input boxes and buttons; learn about testing inputs and buttons with different scenarios and examples.
Functions and Selectors in Playwright: Learn how to launch the Chromium browser with Playwright. Also, gain a better understanding of some important functions like “BrowserContext,” which allows you to run multiple browser sessions, and “newPage” which interacts with a page.
Handling Alerts and Dropdowns in Playwright : Playwright interact with different types of alerts and pop-ups, such as simple, confirmation, and prompt, and different types of dropdowns, such as single selector and multi-selector get your hands-on with handling alerts and dropdown in Playright testing.
Playwright vs Puppeteer: Get to know about the difference between two testing frameworks and how they are different than one another, which browsers they support, and what features they provide.
Run Playwright Tests on LambdaTest: Playwright testing with LambdaTest leverages test performance to the utmost. You can run multiple Playwright tests in Parallel with the LammbdaTest test cloud. Get a step-by-step guide to run your Playwright test on the LambdaTest platform.
Playwright Python Tutorial: Playwright automation framework support all major languages such as Python, JavaScript, TypeScript, .NET and etc. However, there are various advantages to Python end-to-end testing with Playwright because of its versatile utility. Get the hang of Playwright python testing with this chapter.
Playwright End To End Testing Tutorial: Get your hands on with Playwright end-to-end testing and learn to use some exciting features such as TraceViewer, Debugging, Networking, Component testing, Visual testing, and many more.
Playwright Video Tutorial: Watch the video tutorials on Playwright testing from experts and get a consecutive in-depth explanation of Playwright automation testing.