@pytest.mark.multithread
def test_num_reads_threads():
 import threading
 def count_reads(rparser):
  for _ in rparser:
   pass
 n_threads = 4
 threads = []
 rparser = ReadParser(utils.get_test_data('100-reads.fq.gz'))
 for _ in range(n_threads):
  thr = threading.Thread(target=count_reads, args=[rparser])
  threads.append(thr)
  thr.start()
 for thr in threads:
  thr.join()
 assert (rparser.num_reads == 100)

@ pytest . mark . multithread def test_num_reads_threads ( ) : import threading def count_reads ( rparser ) : for _ in rparser : pass n_threads = 4 threads = [ ] rparser = ReadParser ( utils . get_test_data ( '100-reads.fq.gz' ) ) for _ in range ( n_threads ) : thr = threading . Thread ( target = count_reads , args = [ rparser ] ) threads . append ( thr ) thr . start ( ) for thr in threads : thr . join ( ) assert ( rparser . num_reads == 100 )
test threadsaftey of readparsers read counting .