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

How to use _send_file_chunks method in autotest

Best Python code snippet using autotest_python

rss_client.py

Source:rss_client.py

...147        str = self._receive(size, timeout)148        self.transferred += len(str) + 4149        self._report_stats("Received")150        return str151    def _send_file_chunks(self, filename, timeout=60):152        if self._log_func:153            self._log_func("Sending file %s" % filename)154        f = open(filename, "rb")155        try:156            try:157                end_time = time.time() + timeout158                while True:159                    data = f.read(CHUNKSIZE)160                    self._send_packet(data, end_time - time.time())161                    if len(data) < CHUNKSIZE:162                        break163            except FileTransferError, e:164                e.filename = filename165                raise166        finally:167            f.close()168    def _receive_file_chunks(self, filename, timeout=60):169        if self._log_func:170            self._log_func("Receiving file %s" % filename)171        f = open(filename, "wb")172        try:173            try:174                end_time = time.time() + timeout175                while True:176                    data = self._receive_packet(end_time - time.time())177                    f.write(data)178                    if len(data) < CHUNKSIZE:179                        break180            except FileTransferError, e:181                e.filename = filename182                raise183        finally:184            f.close()185    def _send_msg(self, msg, timeout=60):186        self._send(struct.pack("=I", msg))187    def _receive_msg(self, timeout=60):188        s = self._receive(4, timeout)189        return struct.unpack("=I", s)[0]190    def _handle_transfer_error(self):191        # Save original exception192        e = sys.exc_info()193        try:194            # See if we can get an error message195            msg = self._receive_msg()196        except FileTransferError:197            # No error message -- re-raise original exception198            raise e[0], e[1], e[2]199        if msg == RSS_ERROR:200            errmsg = self._receive_packet()201            raise FileTransferServerError(errmsg)202        raise e[0], e[1], e[2]203class FileUploadClient(FileTransferClient):204    """205    Connect to a RSS (remote shell server) and upload files or directory trees.206    """207    def __init__(self, address, port, log_func=None, timeout=20):208        """209        Connect to a server.210        @param address: The server's address211        @param port: The server's port212        @param log_func: If provided, transfer stats will be passed to this213                function during the transfer214        @param timeout: Time duration to wait for connection to succeed215        @raise FileTransferConnectError: Raised if the connection fails216        @raise FileTransferProtocolError: Raised if an incorrect magic number217                is received218        @raise FileTransferSocketError: Raised if the RSS_UPLOAD message cannot219                be sent to the server220        """221        super(FileUploadClient, self).__init__(address, port, log_func, timeout)222        self._send_msg(RSS_UPLOAD)223    def _upload_file(self, path, end_time):224        if os.path.isfile(path):225            self._send_msg(RSS_CREATE_FILE)226            self._send_packet(os.path.basename(path))227            self._send_file_chunks(path, end_time - time.time())228        elif os.path.isdir(path):229            self._send_msg(RSS_CREATE_DIR)230            self._send_packet(os.path.basename(path))231            for filename in os.listdir(path):232                self._upload_file(os.path.join(path, filename), end_time)233            self._send_msg(RSS_LEAVE_DIR)234    def upload(self, src_pattern, dst_path, timeout=600):235        """236        Send files or directory trees to the server.237        The semantics of src_pattern and dst_path are similar to those of scp.238        For example, the following are OK:239            src_pattern='/tmp/foo.txt', dst_path='C:\\'240                (uploads a single file)241            src_pattern='/usr/', dst_path='C:\\Windows\\'...

rss_file_transfer.py

Source:rss_file_transfer.py

...96        self._send(str)97    def _receive_packet(self, timeout=10):98        size = struct.unpack("=I", self._receive(4))[0]99        return self._receive(size, timeout)100    def _send_file_chunks(self, filename, timeout=30):101        f = open(filename, "rb")102        try:103            end_time = time.time() + timeout104            while time.time() < end_time:105                data = f.read(CHUNKSIZE)106                self._send_packet(data)107                if len(data) < CHUNKSIZE:108                    break109            else:110                raise FileTransferTimeoutError("Timeout expired while sending "111                                               "file %s" % filename)112        finally:113            f.close()114    def _receive_file_chunks(self, filename, timeout=30):115        f = open(filename, "wb")116        try:117            end_time = time.time() + timeout118            while True:119                try:120                    data = self._receive_packet(end_time - time.time())121                except FileTransferTimeoutError:122                    raise FileTransferTimeoutError("Timeout expired while "123                                                   "receiving file %s" %124                                                   filename)125                except FileTransferProtocolError:126                    raise FileTransferProtocolError("Error receiving file %s" %127                                                    filename)128                f.write(data)129                if len(data) < CHUNKSIZE:130                    break131        finally:132            f.close()133    def _send_msg(self, msg, timeout=10):134        self._send(struct.pack("=I", msg))135    def _receive_msg(self, timeout=10):136        s = self._receive(4, timeout)137        return struct.unpack("=I", s)[0]138    def _handle_transfer_error(self):139        # Save original exception140        e = sys.exc_info()141        try:142            # See if we can get an error message143            msg = self._receive_msg()144        except FileTransferError:145            # No error message -- re-raise original exception146            raise e[0], e[1], e[2]147        if msg == RSS_ERROR:148            errmsg = self._receive_packet()149            raise FileTransferServerError("Server said: %s" % errmsg)150        raise e[0], e[1], e[2]151class FileUploadClient(FileTransferClient):152    """153    Connect to a RSS (remote shell server) and upload files or directory trees.154    """155    def __init__(self, address, port, timeout=10):156        """157        Connect to a server.158        @param address: The server's address159        @param port: The server's port160        @param timeout: Time duration to wait for connection to succeed161        @raise FileTransferConnectError: Raised if the connection fails162        @raise FileTransferProtocolError: Raised if an incorrect magic number163                is received164        @raise FileTransferSendError: Raised if the RSS_UPLOAD message cannot165                be sent to the server166        """167        super(FileUploadClient, self).__init__(address, port, timeout)168        self._send_msg(RSS_UPLOAD)169    def _upload_file(self, path, end_time):170        if os.path.isfile(path):171            self._send_msg(RSS_CREATE_FILE)172            self._send_packet(os.path.basename(path))173            self._send_file_chunks(path, max(0, end_time - time.time()))174        elif os.path.isdir(path):175            self._send_msg(RSS_CREATE_DIR)176            self._send_packet(os.path.basename(path))177            for filename in os.listdir(path):178                self._upload_file(os.path.join(path, filename), end_time)179            self._send_msg(RSS_LEAVE_DIR)180    def upload(self, src_pattern, dst_path, timeout=600):181        """182        Send files or directory trees to the server.183        The semantics of src_pattern and dst_path are similar to those of scp.184        For example, the following are OK:185            src_pattern='/tmp/foo.txt', dst_path='C:\\'186                (uploads a single file)187            src_pattern='/usr/', dst_path='C:\\Windows\\'...

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.