Method test_flush

Lib/test/test_free_threading/test_mmap.py:256–276

(self)

    @unittest.skipUnless(os.name == "posix", "requires Posix")
    @unittest.skipUnless(hasattr(mmap.mmap, "resize"), "requires mmap.resize")
    def test_flush(self):
        mmap_filename = "test_mmap_file"
        resize_to = 1024

        def resize_and_flush(mm_obj):
            mm_obj.resize(resize_to)
            mm_obj.flush()

        with tempfile.TemporaryDirectory() as tmpdirname:
            file_path = f"{tmpdirname}/{mmap_filename}"
            with open(file_path, "wb+") as file:
                file.write(b"CPython")
                file.flush()
                with mmap.mmap(file.fileno(), 1) as mm_obj:
                    run_concurrently(
                        worker_func=resize_and_flush,
                        args=(mm_obj,),
                        nthreads=NTHREADS,
                    )

            self.assertEqual(os.path.getsize(file_path), resize_to)

    def test_mmap_export_as_memoryview(self):
        """