@skip('win32')
def test_cp16623():
 global FINISHED_COUNTER
 FINISHED_COUNTER = 0
 import time
 expected_lines = ['a', ('bbb' * 100), 'cc']
 total_threads = 50
 file_name = path_combine(testpath.temporary_dir, 'cp16623.txt')
 f = open(file_name, 'w')
 def write_stuff():
  global FINISHED_COUNTER
  global CP16623_LOCK
  for j in xrange(100):
   for i in xrange(50):
    print >>f, 'a'
   print >>f, ('bbb' * 1000)
   for i in xrange(10):
    print >>f, 'cc'
  with CP16623_LOCK:
   FINISHED_COUNTER += 1
 for i in xrange(total_threads):
  thread.start_new_thread(write_stuff, ())
 for i in xrange(total_threads):
  if (FINISHED_COUNTER != total_threads):
   print '*',
   time.sleep(1)
  else:
   break
 AreEqual(FINISHED_COUNTER, total_threads)
 f.close()

@ skip ( 'win32' ) def test_cp16623 ( ) : global FINISHED_COUNTER FINISHED_COUNTER = 0 import time expected_lines = [ 'a' , ( 'bbb' * 100 ) , 'cc' ] total_threads = 50 file_name = path_combine ( testpath . temporary_dir , 'cp16623.txt' ) f = open ( file_name , 'w' ) def write_stuff ( ) : global FINISHED_COUNTER global CP16623_LOCK for j in xrange ( 100 ) : for i in xrange ( 50 ) : print >> f , 'a' print >> f , ( 'bbb' * 1000 ) for i in xrange ( 10 ) : print >> f , 'cc' with CP16623_LOCK : FINISHED_COUNTER += 1 for i in xrange ( total_threads ) : thread . start_new_thread ( write_stuff , ( ) ) for i in xrange ( total_threads ) : if ( FINISHED_COUNTER != total_threads ) : print '*' , time . sleep ( 1 ) else : break AreEqual ( FINISHED_COUNTER , total_threads ) f . close ( )
if this test ever fails randomly .