test largefile support on system where this makes sense size of file to create 2 gib 2 gib 2 147 483 648 bytes test that each file function works as expected for large i e 2 gib files pyio fileio readall uses a temporary bytearray then casted to bytes so memuse2 is needed bpo24658 test that a read greater than 2gb does not fail the a that was written at the end of file above the a that was written at the end of file above else we ve lost track of the true size cut it back via seek truncate with no argument ensure that truncatesmaller than true size shrinks the file xxx truncatelarger than true size is illdefined across platform cut it waaaaay back issue 5016 seekable can return false when the current position is negative when truncated to an int exact required disk space would be size 2 but let s give it a bit more tolerance internally shutil copyfile can use fast copy methods like os sendfile exact required disk space would be size 2 but let s give it a bit more tolerance the default handler for sigxfsz is to abort the process by ignoring it system calls exceeding the file size resource limit will raise oserror instead of crashing the interpreter on windows and mac osx this test consumes large resources it takes a long time to build the 2 gib file and takes 2 gib of disk space therefore the resource must be enabled to run this test if not nothing after this line stanza will be executed only run if the current filesystem supports large files skip this test on windows since we now always support large files 231 2147483648 seeking is not enough of a test you must write and flush too c implementation of io python implementation of io size of file to create 2 gib 2 gib 2 147 483 648 bytes test that each file function works as expected for large i e 2 gib files _pyio fileio readall uses a temporary bytearray then casted to bytes so memuse 2 is needed bpo 24658 test that a read greater than 2gb does not fail seek from the end the a that was written at the end of file above the a that was written at the end of file above else we ve lost track of the true size cut it back via seek truncate with no argument else pointer moved else wasn t truncated ensure that truncate smaller than true size shrinks the file xxx truncate larger than true size is ill defined across platform cut it waaaaay back else pointer moved else wasn t truncated issue 5016 seekable can return false when the current position is negative when truncated to an int exact required disk space would be size 2 but let s give it a bit more tolerance internally shutil copyfile can use fast copy methods like os sendfile exact required disk space would be size 2 but let s give it a bit more tolerance the default handler for sigxfsz is to abort the process by ignoring it system calls exceeding the file size resource limit will raise oserror instead of crashing the interpreter on windows and mac osx this test consumes large resources it takes a long time to build the 2 gib file and takes 2 gib of disk space therefore the resource must be enabled to run this test if not nothing after this line stanza will be executed only run if the current filesystem supports large files skip this test on windows since we now always support large files 2 31 2147483648 seeking is not enough of a test you must write and flush too
import os
import stat
import sys
import unittest
import socket
import shutil
import threading
from test.support import requires, bigmemtest, requires_resource
from test.support import SHORT_TIMEOUT
from test.support import socket_helper
from test.support.os_helper import TESTFN, unlink
import io  
import _pyio as pyio 


size = 2_500_000_000
TESTFN2 = TESTFN + '2'


class LargeFileTest:

    def setUp(self):
        if os.path.exists(TESTFN):
            mode = 'r+b'
        else:
            mode = 'w+b'

        with self.open(TESTFN, mode) as f:
            current_size = os.fstat(f.fileno())[stat.ST_SIZE]
            if current_size == size+1:
                return

            if current_size == 0:
                f.write(b'z')

            f.seek(0)
            f.seek(size)
            f.write(b'a')
            f.flush()
            self.assertEqual(os.fstat(f.fileno())[stat.ST_SIZE], size+1)

    @classmethod
    def tearDownClass(cls):
        with cls.open(TESTFN, 'wb'):
            pass
        if not os.stat(TESTFN)[stat.ST_SIZE] == 0:
            raise cls.failureException('File was not truncated by opening '
                                       'with mode "wb"')
        unlink(TESTFN2)


class TestFileMethods(LargeFileTest):
    

    
    
    @bigmemtest(size=size, memuse=2, dry_run=False)
    def test_large_read(self, _size):
        
        with self.open(TESTFN, "rb") as f:
            self.assertEqual(len(f.read()), size + 1)
            self.assertEqual(f.tell(), size + 1)

    def test_osstat(self):
        self.assertEqual(os.stat(TESTFN)[stat.ST_SIZE], size+1)

    def test_seek_read(self):
        with self.open(TESTFN, 'rb') as f:
            self.assertEqual(f.tell(), 0)
            self.assertEqual(f.read(1), b'z')
            self.assertEqual(f.tell(), 1)
            f.seek(0)
            self.assertEqual(f.tell(), 0)
            f.seek(0, 0)
            self.assertEqual(f.tell(), 0)
            f.seek(42)
            self.assertEqual(f.tell(), 42)
            f.seek(42, 0)
            self.assertEqual(f.tell(), 42)
            f.seek(42, 1)
            self.assertEqual(f.tell(), 84)
            f.seek(0, 1)
            self.assertEqual(f.tell(), 84)
            f.seek(0, 2)  
            self.assertEqual(f.tell(), size + 1 + 0)
            f.seek(-10, 2)
            self.assertEqual(f.tell(), size + 1 - 10)
            f.seek(-size-1, 2)
            self.assertEqual(f.tell(), 0)
            f.seek(size)
            self.assertEqual(f.tell(), size)
            
            self.assertEqual(f.read(1), b'a')
            f.seek(-size-1, 1)
            self.assertEqual(f.read(1), b'z')
            self.assertEqual(f.tell(), 1)

    def test_lseek(self):
        with self.open(TESTFN, 'rb') as f:
            self.assertEqual(os.lseek(f.fileno(), 0, 0), 0)
            self.assertEqual(os.lseek(f.fileno(), 42, 0), 42)
            self.assertEqual(os.lseek(f.fileno(), 42, 1), 84)
            self.assertEqual(os.lseek(f.fileno(), 0, 1), 84)
            self.assertEqual(os.lseek(f.fileno(), 0, 2), size+1+0)
            self.assertEqual(os.lseek(f.fileno(), -10, 2), size+1-10)
            self.assertEqual(os.lseek(f.fileno(), -size-1, 2), 0)
            self.assertEqual(os.lseek(f.fileno(), size, 0), size)
            
            self.assertEqual(f.read(1), b'a')

    def test_truncate(self):
        with self.open(TESTFN, 'r+b') as f:
            if not hasattr(f, 'truncate'):
                raise unittest.SkipTest("open().truncate() not available "
                                        "on this system")
            f.seek(0, 2)
            
            self.assertEqual(f.tell(), size+1)
            
            newsize = size - 10
            f.seek(newsize)
            f.truncate()
            self.assertEqual(f.tell(), newsize)  
            f.seek(0, 2)
            self.assertEqual(f.tell(), newsize)  
            
            
            newsize -= 1
            f.seek(42)
            f.truncate(newsize)
            self.assertEqual(f.tell(), 42)
            f.seek(0, 2)
            self.assertEqual(f.tell(), newsize)
            
            
            f.seek(0)
            f.truncate(1)
            self.assertEqual(f.tell(), 0)       
            f.seek(0)
            self.assertEqual(len(f.read()), 1)  

    def test_seekable(self):
        
        
        for pos in (2**31-1, 2**31, 2**31+1):
            with self.open(TESTFN, 'rb') as f:
                f.seek(pos)
                self.assertTrue(f.seekable())


def skip_no_disk_space(path, required):
    def decorator(fun):
        def wrapper(*args, **kwargs):
            if not hasattr(shutil, "disk_usage"):
                raise unittest.SkipTest("requires shutil.disk_usage")
            if shutil.disk_usage(os.path.realpath(path)).free < required:
                hsize = int(required / 1024 / 1024)
                raise unittest.SkipTest(
                    f"required {hsize} MiB of free disk space")
            return fun(*args, **kwargs)
        return wrapper
    return decorator


class TestCopyfile(LargeFileTest, unittest.TestCase):
    open = staticmethod(io.open)

    
    
    @skip_no_disk_space(TESTFN, size * 2.5)
    @requires_resource('cpu')
    def test_it(self):
        
        
        size = os.path.getsize(TESTFN)
        shutil.copyfile(TESTFN, TESTFN2)
        self.assertEqual(os.path.getsize(TESTFN2), size)
        with open(TESTFN2, 'rb') as f:
            self.assertEqual(f.read(5), b'z\x00\x00\x00\x00')
            f.seek(size - 5)
            self.assertEqual(f.read(), b'\x00\x00\x00\x00a')


@unittest.skipIf(not hasattr(os, 'sendfile'), 'sendfile not supported')
class TestSocketSendfile(LargeFileTest, unittest.TestCase):
    open = staticmethod(io.open)
    timeout = SHORT_TIMEOUT

    def setUp(self):
        super().setUp()
        self.thread = None

    def tearDown(self):
        super().tearDown()
        if self.thread is not None:
            self.thread.join(self.timeout)
            self.thread = None

    def tcp_server(self, sock):
        def run(sock):
            with sock:
                conn, _ = sock.accept()
                conn.settimeout(self.timeout)
                with conn, open(TESTFN2, 'wb') as f:
                    event.wait(self.timeout)
                    while True:
                        chunk = conn.recv(65536)
                        if not chunk:
                            return
                        f.write(chunk)

        event = threading.Event()
        sock.settimeout(self.timeout)
        self.thread = threading.Thread(target=run, args=(sock, ))
        self.thread.start()
        event.set()

    
    
    @skip_no_disk_space(TESTFN, size * 2.5)
    @requires_resource('cpu')
    def test_it(self):
        port = socket_helper.find_unused_port()
        with socket.create_server(("", port)) as sock:
            self.tcp_server(sock)
            with socket.create_connection(("127.0.0.1", port)) as client:
                with open(TESTFN, 'rb') as f:
                    client.sendfile(f)
        self.tearDown()

        size = os.path.getsize(TESTFN)
        self.assertEqual(os.path.getsize(TESTFN2), size)
        with open(TESTFN2, 'rb') as f:
            self.assertEqual(f.read(5), b'z\x00\x00\x00\x00')
            f.seek(size - 5)
            self.assertEqual(f.read(), b'\x00\x00\x00\x00a')


def setUpModule():
    try:
        import signal
        
        
        
        signal.signal(signal.SIGXFSZ, signal.SIG_IGN)
    except (ImportError, AttributeError):
        pass

    
    
    
    
    if sys.platform[:3] == 'win' or sys.platform == 'darwin':
        requires('largefile',
                 'test requires %s bytes and a long time to run' % str(size))
    else:
        
        
        
        f = open(TESTFN, 'wb', buffering=0)
        try:
            
            f.seek(2147483649)
            
            f.write(b'x')
            f.flush()
        except (OSError, OverflowError):
            raise unittest.SkipTest("filesystem does not have "
                                    "largefile support")
        finally:
            f.close()
            unlink(TESTFN)


class CLargeFileTest(TestFileMethods, unittest.TestCase):
    open = staticmethod(io.open)


class PyLargeFileTest(TestFileMethods, unittest.TestCase):
    open = staticmethod(pyio.open)


def tearDownModule():
    unlink(TESTFN)
    unlink(TESTFN2)


if __name__ == '__main__':
    unittest.main()