test_functional.py 文件源码

def test_abor_during_transfer(self):
        # Case 4: ABOR while a data transfer on DTP channel is in
        # progress: close data channel, respond with 426, respond
        # with 226.
        data = b'abcde12345' * 1000000
        with open(TESTFN, 'w+b') as f:
            f.write(data)
        try:
            self.client.voidcmd('TYPE I')
            with contextlib.closing(
                    self.client.transfercmd('retr ' + TESTFN)) as conn:
                bytes_recv = 0
                while bytes_recv < 65536:
                    chunk = conn.recv(BUFSIZE)
                    bytes_recv += len(chunk)

                # stop transfer while it isn't finished yet
                self.client.putcmd('ABOR')

                # transfer isn't finished yet so ftpd should respond with 426
                self.assertEqual(self.client.getline()[:3], "426")

                # transfer successfully aborted, so should now respond
                # with a 226
                self.assertEqual('226', self.client.voidresp()[:3])
        finally:
            # We do not use os.remove() because file could still be
            # locked by ftpd thread.  If DELE through FTP fails try
            # os.remove() as last resort.
            try:
                self.client.delete(TESTFN)
            except (ftplib.Error, EOFError, socket.error):
                safe_remove(TESTFN)