Add tests

Author: cmaloney

Lib/test/_test_eintr.py

@@ -152,6 +152,37 @@ def test_read(self):
                 self.assertEqual(data, os.read(rd, len(data)))
             self.assertEqual(proc.wait(), 0)
 
+    def test_readinto(self):
+        rd, wr = os.pipe()
+        self.addCleanup(os.close, rd)
+        # wr closed explicitly by parent
+
+        # the payload below are smaller than PIPE_BUF, hence the writes will be
+        # atomic
+        datas = [b"hello", b"world", b"spam"]
+        bufs = [bytearray(5), bytearray(5), bytearray(4)]
+
+        code = '\n'.join((
+            'import os, sys, time',
+            '',
+            'wr = int(sys.argv[1])',
+            'datas = %r' % datas,
+            'sleep_time = %r' % self.sleep_time,
+            '',
+            'for data in datas:',
+            '    # let the parent block on read()',
+            '    time.sleep(sleep_time)',
+            '    os.write(wr, data)',
+        ))
+
+        proc = self.subprocess(code, str(wr), pass_fds=[wr])
+        with kill_on_error(proc):
+            os.close(wr)
+            for data, buffer in zip(datas, bufs):
+                os.readinto(rd, buffer)
+                self.assertEqual(data, buffer)
+            self.assertEqual(proc.wait(), 0)
+
     def test_write(self):
         rd, wr = os.pipe()
         self.addCleanup(os.close, wr)

Lib/test/test_os.py

@@ -230,6 +230,47 @@ def test_read(self):
             self.assertEqual(type(s), bytes)
             self.assertEqual(s, b"spam")
 
+    def test_readinto(self):
+        with open(os_helper.TESTFN, "w+b") as fobj:
+            fobj.write(b"spam")
+            fobj.flush()
+            fd = fobj.fileno()
+            os.lseek(fd, 0, 0)
+            buffer = bytearray(8)
+            s = os.readinto(fd, buffer)
+            self.assertEqual(type(s), int)
+            self.assertEqual(s, 4)
+            # Should overwrite the first 4 bytes of the buffer.
+            self.assertEqual(bytes(buffer), b"spam\0\0\0\0")
+
+            # Readinto at EOF shold return 0 and not touch buffer
+            buffer[:] = b"notspam\0"
+            s = os.readinto(fd, buffer)
+            self.assertEqual(type(s), int)
+            self.assertEqual(s, 0)
+            self.assertEqual(bytes(buffer), b"notspam\0")
+            s = os.readinto(fd, buffer)
+            self.assertEqual(s, 0)
+            self.assertEqual(bytes(buffer), b"notspam\0")
+
+    def test_readinto_badbuffer(self):
+        with open(os_helper.TESTFN, "w+b") as fobj:
+            fobj.write(b"spam")
+            fobj.flush()
+            fd = fobj.fileno()
+            os.lseek(fd, 0, 0)
+
+            for bad_arg in ("test", bytes(), 14):
+                with self.subTest(f"{type(bad_arg)}"):
+                    with self.assertRaises(TypeError):
+                        os.readinto(fd, bad_arg)
+
+            # No data should have been read with the bad arguments.
+            buffer = bytearray(8)
+            s = os.readinto(fd, buffer)
+            self.assertEqual(s, 4)
+            self.assertEqual(bytes(buffer), b"spam\0\0\0\0")
+
     @support.cpython_only
     # Skip the test on 32-bit platforms: the number of bytes must fit in a
     # Py_ssize_t type
@@ -249,6 +290,29 @@ def test_large_read(self, size):
         # operating system is free to return less bytes than requested.
         self.assertEqual(data, b'test')
 
+
+    @support.cpython_only
+    # Skip the test on 32-bit platforms: the number of bytes must fit in a
+    # Py_ssize_t type
+    @unittest.skipUnless(INT_MAX < PY_SSIZE_T_MAX,
+                         "needs INT_MAX < PY_SSIZE_T_MAX")
+    @support.bigmemtest(size=INT_MAX + 10, memuse=1, dry_run=False)
+    def test_large_readinto(self, size):
+        self.addCleanup(os_helper.unlink, os_helper.TESTFN)
+        create_file(os_helper.TESTFN, b'test')
+
+        # Issue #21932: For readinto the buffer contains the length rather than
+        # a length being passed explicitly to read, shold still get capped to a
+        # valid size / not raise an OverflowError for sizes larger than INT_MAX.
+        buffer = bytearray(INT_MAX+10)
+        with open(os_helper.TESTFN, "rb") as fp:
+            length = os.readinto(fp.fileno(), buffer)
+
+        # The test does not try to read more than 2 GiB at once because the
+        # operating system is free to return less bytes than requested.
+        self.assertEqual(length, 4)
+        self.assertEqual(buffer[:4], b'test')
+
     def test_write(self):
         # os.write() accepts bytes- and buffer-like objects but not strings
         fd = os.open(os_helper.TESTFN, os.O_CREAT | os.O_WRONLY)
@@ -2467,6 +2531,10 @@ def test_lseek(self):
     def test_read(self):
         self.check(os.read, 1)
 
+    @unittest.skipUnless(hasattr(os, 'readinto'), 'test needs os.readinto()')
+    def test_readinto(self):
+        self.check(os.readinto, bytearray(5))
+
     @unittest.skipUnless(hasattr(os, 'readv'), 'test needs os.readv()')
     def test_readv(self):
         buf = bytearray(10)