<s> """<STR_LIT>""" <EOL> from __future__ import nested_scopes <EOL> __revision__ = "<STR_LIT>" <EOL> import binascii <EOL> import pprint <EOL> import unittest <EOL> import os <EOL> import time <EOL> import sys <EOL> if sys . version_info [ <NUM_LIT:0> ] == <NUM_LIT:2> and sys . version_info [ <NUM_LIT:1> ] == <NUM_LIT:1> : <EOL> from Crypto . Util . py21compat import * <EOL> from Crypto . Util . py3compat import * <EOL> try : <EOL> import multiprocessing <EOL> except ImportError : <EOL> multiprocessing = None <EOL> import Crypto . Random . _UserFriendlyRNG <EOL> import Crypto . Random . random <EOL> class RNGForkTest ( unittest . TestCase ) : <EOL> def _get_reseed_count ( self ) : <EOL> """<STR_LIT>""" <EOL> rng_singleton = Crypto . Random . _UserFriendlyRNG . _get_singleton ( ) <EOL> rng_singleton . _lock . acquire ( ) <EOL> try : <EOL> return rng_singleton . _fa . reseed_count <EOL> finally : <EOL> rng_singleton . _lock . release ( ) <EOL> def runTest ( self ) : <EOL> if sys . platform . startswith ( '<STR_LIT>' ) : <EOL> assert not hasattr ( os , '<STR_LIT>' ) <EOL> return <EOL> time . sleep ( <NUM_LIT> ) <EOL> reseed_count_before = self . _get_reseed_count ( ) <EOL> Crypto . Random . _UserFriendlyRNG . _get_singleton ( ) . reinit ( ) <EOL> Crypto . Random . get_random_bytes ( <NUM_LIT:1> ) <EOL> reseed_count_after = self . _get_reseed_count ( ) <EOL> self . assertNotEqual ( reseed_count_before , reseed_count_after ) <EOL> rfiles = [ ] <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> rfd , wfd = os . pipe ( ) <EOL> if os . fork ( ) == <NUM_LIT:0> : <EOL> os . close ( rfd ) <EOL> f = os . fdopen ( wfd , "<STR_LIT:wb>" ) <EOL> Crypto . Random . atfork ( ) <EOL> data = Crypto . Random . get_random_bytes ( <NUM_LIT:16> ) <EOL> f . write ( data ) <EOL> f . close ( ) <EOL> os . _exit ( <NUM_LIT:0> ) <EOL> os . close ( wfd ) <EOL> rfiles . append ( os . fdopen ( rfd , "<STR_LIT:rb>" ) ) <EOL> results = [ ] <EOL> results_dict = { } <EOL> for f in rfiles : <EOL> data = binascii . hexlify ( f . read ( ) ) <EOL> results . append ( data ) <EOL> results_dict [ data ] = <NUM_LIT:1> <EOL> f . close ( ) <EOL> if len ( results ) != len ( results_dict . keys ( ) ) : <EOL> raise AssertionError ( "<STR_LIT>" % <EOL> ( pprint . pformat ( results ) ) ) <EOL> def _task_main ( q ) : <EOL> a = Crypto . Random . get_random_bytes ( <NUM_LIT:16> ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> b = Crypto . Random . get_random_bytes ( <NUM_LIT:16> ) <EOL> q . put ( binascii . b2a_hex ( a ) ) <EOL> q . put ( binascii . b2a_hex ( b ) ) <EOL> q . put ( None ) <EOL> class RNGMultiprocessingForkTest ( unittest . TestCase ) : <EOL> def runTest ( self ) : <EOL> n_procs = <NUM_LIT:5> <EOL> manager = multiprocessing . Manager ( ) <EOL> queues = [ manager . Queue ( <NUM_LIT:1> ) for i in range ( n_procs ) ] <EOL> time . sleep ( <NUM_LIT> ) <EOL> Crypto . Random . _UserFriendlyRNG . _get_singleton ( ) . reinit ( ) <EOL> Crypto . Random . get_random_bytes ( <NUM_LIT:1> ) <EOL> pool = multiprocessing . Pool ( processes = n_procs , initializer = Crypto . Random . atfork ) <EOL> map_result = pool . map_async ( _task_main , queues ) <EOL> aa = [ queues [ i ] . get ( <NUM_LIT:30> ) for i in range ( n_procs ) ] <EOL> bb = [ queues [ i ] . get ( <NUM_LIT:30> ) for i in range ( n_procs ) ] <EOL> res = list ( zip ( aa , bb ) ) <EOL> map_result . get ( <NUM_LIT:30> ) <EOL> pool . close ( ) <EOL> pool . join ( ) <EOL> if len ( set ( aa ) ) != len ( aa ) or len ( set ( res ) ) != len ( res ) : <EOL> raise AssertionError ( "<STR_LIT>" % <EOL> ( pprint . pformat ( res ) , ) ) <EOL> def get_tests ( config = { } ) : <EOL> tests = [ ] <EOL> tests += [ RNGForkTest ( ) ] <EOL> if multiprocessing is not None : <EOL> tests += [ RNGMultiprocessingForkTest ( ) ] <EOL> return tests <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> suite = lambda : unittest . TestSuite ( get_tests ( ) ) <EOL> unittest . main ( defaultTest = '<STR_LIT>' ) </s>
