How to use set_recv_cap_data_info method in fMBT

Best Python code snippet using fMBT_python

server.py

Source:server.py

...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_counter...

messages.py

Source:messages.py

...54        # the recv_caps attribute will not be present.55        if not hasattr(self, "recv_caps"):56            self.recv_caps = 057        return self.recv_caps58    def set_recv_cap_data_info(self, value):59        current_caps = getattr(self, "recv_caps", 0)60        if value: # set to true61            self.recv_caps = current_caps | RECV_CAP_DATA_INFO62        elif self.recv_cap_data_info(): # set to false63            self.recv_caps = current_caps - RECV_CAP_DATA_INFO64    def __str__(self):65        self.recv_cap_data_info() # make sure recv_caps exist66        return ('Exec(namespace=%r, code=%r, expr=%r, '67                'lock=%r, async=%r, recv_caps=%r)' % (68                    self.namespace, self.code, self.expr,69                    self.lock, self.async, self.recv_caps))70class Exec_rv(object):71    def __init__(self, code_exc, expr_exc, expr_rv):72        self.code_exc = code_exc...

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.