← Back to branch summary

~pythonregexp2.7/python/issue2636-11

« back to all changes in this revision

Viewing changes to Lib/test/test_io.py

  • Committer: Jeffrey C. "The TimeHorse" Jacobs
  • Date: 2008-09-21 17:53:26 UTC
  • mfrom: (39025.1.14 Regexp-2.7)
  • Revision ID: darklord@timehorse.com-20080921175326-92vaej2hc3yuecxb
Merged in changes from the core Regexp branch.

Show diffs side-by-side

added added

removed removed

Lines of Context:
Lib/test/test_io.py
6
6
import sys
7
7
import time
8
8
import array
 
9
import threading
 
10
import random
9
11
import unittest
10
 
from itertools import chain
 
12
from itertools import chain, cycle
11
13
from test import test_support
12
14
 
13
15
import codecs
390
392
        # this test. Else, write it.
391
393
        pass
392
394
 
 
395
    def testThreads(self):
 
396
        try:
 
397
            # Write out many bytes with exactly the same number of 0's,
 
398
            # 1's... 255's. This will help us check that concurrent reading
 
399
            # doesn't duplicate or forget contents.
 
400
            N = 1000
 
401
            l = range(256) * N
 
402
            random.shuffle(l)
 
403
            s = bytes(bytearray(l))
 
404
            with io.open(test_support.TESTFN, "wb") as f:
 
405
                f.write(s)
 
406
            with io.open(test_support.TESTFN, "rb", buffering=0) as raw:
 
407
                bufio = io.BufferedReader(raw, 8)
 
408
                errors = []
 
409
                results = []
 
410
                def f():
 
411
                    try:
 
412
                        # Intra-buffer read then buffer-flushing read
 
413
                        for n in cycle([1, 19]):
 
414
                            s = bufio.read(n)
 
415
                            if not s:
 
416
                                break
 
417
                            # list.append() is atomic
 
418
                            results.append(s)
 
419
                    except Exception as e:
 
420
                        errors.append(e)
 
421
                        raise
 
422
                threads = [threading.Thread(target=f) for x in range(20)]
 
423
                for t in threads:
 
424
                    t.start()
 
425
                time.sleep(0.02) # yield
 
426
                for t in threads:
 
427
                    t.join()
 
428
                self.assertFalse(errors,
 
429
                    "the following exceptions were caught: %r" % errors)
 
430
                s = b''.join(results)
 
431
                for i in range(256):
 
432
                    c = bytes(bytearray([i]))
 
433
                    self.assertEqual(s.count(c), N)
 
434
        finally:
 
435
            test_support.unlink(test_support.TESTFN)
 
436
 
 
437
 
393
438
 
394
439
class BufferedWriterTest(unittest.TestCase):
395
440
 
446
491
 
447
492
        self.assertEquals(b"abc", writer._write_stack[0])
448
493
 
 
494
    def testThreads(self):
 
495
        # BufferedWriter should not raise exceptions or crash
 
496
        # when called from multiple threads.
 
497
        try:
 
498
            # We use a real file object because it allows us to
 
499
            # exercise situations where the GIL is released before
 
500
            # writing the buffer to the raw streams. This is in addition
 
501
            # to concurrency issues due to switching threads in the middle
 
502
            # of Python code.
 
503
            with io.open(test_support.TESTFN, "wb", buffering=0) as raw:
 
504
                bufio = io.BufferedWriter(raw, 8)
 
505
                errors = []
 
506
                def f():
 
507
                    try:
 
508
                        # Write enough bytes to flush the buffer
 
509
                        s = b"a" * 19
 
510
                        for i in range(50):
 
511
                            bufio.write(s)
 
512
                    except Exception as e:
 
513
                        errors.append(e)
 
514
                        raise
 
515
                threads = [threading.Thread(target=f) for x in range(20)]
 
516
                for t in threads:
 
517
                    t.start()
 
518
                time.sleep(0.02) # yield
 
519
                for t in threads:
 
520
                    t.join()
 
521
                self.assertFalse(errors,
 
522
                    "the following exceptions were caught: %r" % errors)
 
523
        finally:
 
524
            test_support.unlink(test_support.TESTFN)
 
525
 
449
526
 
450
527
class BufferedRWPairTest(unittest.TestCase):
451
528