Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _remote_execute_and_forward method in fMBT

Best Python code snippet using fMBT_python

server.py

Source:server.py

...275    pythonshare._send(exec_msg, rns.to_remote)276    # _recv raises EOFError() if disconnected,277    # let it raise through.278    return pythonshare._recv(rns.from_remote)279def _remote_execute_and_forward(ns, exec_msg, to_client, peername=None):280    """returns (forward_status, info)281    forward_status values:282       True:  everything successfully forwarded,283              info contains pair (forwarded byte count, full length).284       False: not everything forwarded,285              info contains pair (forwarded byte count, full length).286              to_client file/socket is not functional.287       None:  no forwarding,288              info contains Exec_rv that should be sent normally.289    Raises EOFError if connection to remote namespace is not functional.290    The peername parameter is used for logging only.291    """292    client_supports_rv_info = exec_msg.recv_cap_data_info()293    exec_msg.set_recv_cap_data_info(True)294    rns = _g_remote_namespaces[ns]295    pythonshare._send(exec_msg, rns.to_remote)296    from_remote = rns.from_remote297    # Must keep simultaneously two locks:298    # - send lock on to_client299    # - recv lock on from_remote300    pythonshare._acquire_recv_lock(from_remote)301    try:302        response = pythonshare._recv(from_remote, acquire_recv_lock=False)303        if not isinstance(response, messages.Data_info):304            # Got direct response without forward mode305            return (None, response)306        pythonshare._acquire_send_lock(to_client)307        if client_supports_rv_info:308            # send data_info to client309            pythonshare._send(response, to_client, acquire_send_lock=False)310        try:311            if opt_debug and peername:312                daemon_log("%s:%s <= Exec_rv([forwarding %s B])" % (peername + (response.data_length,)))313            forwarded_bytes = pythonshare._forward(314                from_remote, to_client, response.data_length,315                acquire_recv_lock=False,316                acquire_send_lock=False)317            if forwarded_bytes == response.data_length:318                return (True, (forwarded_bytes, response.data_length))319            else:320                return (False, (forwarded_bytes, response.data_length))321        finally:322            pythonshare._release_send_lock(to_client)323    finally:324        exec_msg.set_recv_cap_data_info(client_supports_rv_info)325        pythonshare._release_recv_lock(from_remote)326def _connection_lost(conn_id, *closables):327    if closables:328        pythonshare._close(*closables)329    try:330        for ns in _g_namespace_users[conn_id]:331            try:332                _g_local_namespaces[ns]["pythonshare_ns"].call_on_disconnect(conn_id)333            except KeyError:334                pass335    except KeyError:336        pass337def _serve_connection(conn, conn_opts):338    global _g_async_rv_counter339    global _g_server_shutdown340    if isinstance(conn, client.Connection):341        to_client = conn._to_server342        from_client = conn._from_server343    else: # conn is a connected socket344        to_client = conn.makefile("w")345        from_client = conn.makefile("r")346    try:347        peername = conn.getpeername()348    except socket.error:349        peername = ("unknown", "?")350    if opt_debug:351        daemon_log("connected %s:%s" % peername)352    conn_id = "%s-%s" % (timestamp(), id(conn))353    auth_ok = False354    passwords = [k for k in conn_opts.keys() if k.startswith("password.")]355    kill_server_on_close = conn_opts.get("kill-server-on-close", False)356    if passwords:357        # password authentication is required for this connection358        received_password = pythonshare._recv(from_client)359        for password_type in passwords:360            algorithm = password_type.split(".")[1]361            if type(received_password) == str:362                if (algorithm == "plaintext" and363                    received_password == conn_opts[password_type]):364                    auth_ok = True365                elif (hasattr(hashlib, algorithm) and366                      getattr(hashlib, algorithm)(received_password).hexdigest() ==367                      conn_opts[password_type]):368                    auth_ok = True369        try:370            if auth_ok:371                pythonshare._send(messages.Auth_rv(True), to_client)372                if opt_debug:373                    daemon_log("%s:%s authentication ok" % peername)374            else:375                pythonshare._send(messages.Auth_rv(False), to_client)376                if opt_debug:377                    daemon_log("%s:%s authentication failed" % peername)378        except socket.error:379            daemon_log("authentication failed due to socket error")380            auth_ok = False381    else:382        auth_ok = True # no password required383    whitelist_local = conn_opts.get("whitelist_local", None)384    while auth_ok:385        try:386            obj = pythonshare._recv(from_client)387            if opt_debug:388                daemon_log("%s:%s => %s" % (peername + (obj,)))389        except (EOFError, pythonshare.socket.error):390            break391        if isinstance(obj, messages.Register_ns):392            try:393                _init_remote_namespace(obj.ns, conn, to_client, from_client)394                pythonshare._send(messages.Ns_rv(True), to_client)395                # from this point on, this connection is reserved for396                # sending remote namespace traffic. The connection will be397                # used by other threads, this thread stops here.398                return399            except Exception, e:400                pythonshare._send(messages.Ns_rv(False, exception2string(sys.exc_info())), to_client)401        elif isinstance(obj, messages.Drop_ns):402            try:403                if obj.ns in _g_local_namespaces:404                    _drop_local_namespace(obj.ns)405                elif obj.ns in _g_remote_namespaces:406                    _drop_remote_namespace(obj.ns)407                else:408                    raise ValueError('Unknown namespace "%s"' % (obj.ns,))409                pythonshare._send(messages.Ns_rv(True), to_client)410            except Exception, e:411                if opt_debug:412                    daemon_log("namespace drop error: %s" % (e,))413                pythonshare._send(messages.Ns_rv(False, exception2string(sys.exc_info())), to_client)414        elif isinstance(obj, messages.Request_ns):415            ns = obj.ns416            if (ns in _g_remote_namespaces or417                ns in _g_local_namespaces):418                _register_exported_namespace(ns, conn)419                pythonshare._send(messages.Ns_rv(True), to_client)420                # from this point on, this connection is reserved for421                # receiving executions on requested namespace. This422                # thread starts serving the connection.423        elif isinstance(obj, messages.Exec):424            ns = obj.namespace425            if ns in _g_remote_namespaces: # execute in remote namespace426                try:427                    _fwd_status, _fwd_info = _remote_execute_and_forward(428                        ns, obj, to_client, peername)429                    if _fwd_status == True:430                        # successfully forwarded431                        if opt_debug:432                            daemon_log("%s:%s forwarded %s B" % (peername + (_fwd_info[0],)))433                        exec_rv = None # return value fully forwarded434                    elif _fwd_status == False:435                        # connection to client is broken436                        if opt_debug:437                            daemon_log("%s:%s error after forwarding %s/%s B" % (peername + _fwd_info))438                        break439                    elif _fwd_status is None:440                        # nothing forwarded, send return value by normal means441                        exec_rv = _fwd_info...

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.