| 908 | @support.no_tracing |
| 909 | @threading_helper.requires_working_threading() |
| 910 | def test_threads_write(self): |
| 911 | # Issue6750: concurrent writes could duplicate data |
| 912 | event = threading.Event() |
| 913 | with self.open(os_helper.TESTFN, "w", encoding="utf-8", buffering=1) as f: |
| 914 | def run(n): |
| 915 | text = "Thread%03d\n" % n |
| 916 | event.wait() |
| 917 | f.write(text) |
| 918 | threads = [threading.Thread(target=run, args=(x,)) |
| 919 | for x in range(20)] |
| 920 | with threading_helper.start_threads(threads, event.set): |
| 921 | time.sleep(0.02) |
| 922 | with self.open(os_helper.TESTFN, encoding="utf-8") as f: |
| 923 | content = f.read() |
| 924 | for n in range(20): |
| 925 | self.assertEqual(content.count("Thread%03d\n" % n), 1) |
| 926 | |
| 927 | def test_flush_error_on_close(self): |
| 928 | # Test that text file is closed despite failed flush |