<s> __author__ = '<STR_LIT>' <EOL> import logging <EOL> import random <EOL> import os <EOL> from pypet import pypetconstants <EOL> from pypet . environment import Environment <EOL> from pypet . tests . integration . environment_test import EnvironmentTest , ResultSortTest , TestOtherHDF5Settings2 , multiply <EOL> from pypet . tests . testutils . ioutils import run_suite , make_temp_dir , make_trajectory_name , parse_args , get_log_config , unittest , get_random_port_url <EOL> from pypet . tests . testutils . data import create_param_dict , add_params <EOL> import pypet . compat as compat <EOL> import platform <EOL> try : <EOL> import psutil <EOL> except ImportError : <EOL> psutil = None <EOL> try : <EOL> import zmq <EOL> except ImportError : <EOL> zmq = None <EOL> def check_nice ( nice ) : <EOL> """<STR_LIT>""" <EOL> if hasattr ( os , '<STR_LIT>' ) : <EOL> return nice <EOL> elif psutil is not None : <EOL> return psutil . ABOVE_NORMAL_PRIORITY_CLASS <EOL> else : <EOL> return None <EOL> class MultiprocPoolQueueTest ( TestOtherHDF5Settings2 ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> def set_mode ( self ) : <EOL> super ( MultiprocPoolQueueTest , self ) . set_mode ( ) <EOL> self . mode = pypetconstants . WRAP_MODE_QUEUE <EOL> self . multiproc = True <EOL> self . ncores = <NUM_LIT:4> <EOL> self . use_pool = True <EOL> self . niceness = check_nice ( <NUM_LIT:4> ) <EOL> class MultiprocPoolSortQueueTest ( ResultSortTest ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> def set_mode ( self ) : <EOL> super ( MultiprocPoolSortQueueTest , self ) . set_mode ( ) <EOL> self . mode = pypetconstants . WRAP_MODE_QUEUE <EOL> self . multiproc = True <EOL> self . ncores = <NUM_LIT:3> <EOL> self . use_pool = True <EOL> class MultiprocPoolSortLockTest ( ResultSortTest ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> def set_mode ( self ) : <EOL> super ( MultiprocPoolSortLockTest , self ) . set_mode ( ) <EOL> self . mode = pypetconstants . WRAP_MODE_LOCK <EOL> self . multiproc = True <EOL> self . ncores = <NUM_LIT:4> <EOL> self . use_pool = True <EOL> class MultiprocPoolSortLocalTest ( ResultSortTest ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> def set_mode ( self ) : <EOL> super ( MultiprocPoolSortLocalTest , self ) . set_mode ( ) <EOL> self . mode = pypetconstants . WRAP_MODE_LOCAL <EOL> self . multiproc = True <EOL> self . ncores = <NUM_LIT:4> <EOL> self . use_pool = True <EOL> self . graceful_exit = True <EOL> @ unittest . skipIf ( platform . system ( ) == '<STR_LIT>' , '<STR_LIT>' ) <EOL> class MultiprocPoolSortPipeTest ( ResultSortTest ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> def set_mode ( self ) : <EOL> super ( MultiprocPoolSortPipeTest , self ) . set_mode ( ) <EOL> self . mode = pypetconstants . WRAP_MODE_PIPE <EOL> self . multiproc = True <EOL> self . ncores = <NUM_LIT:4> <EOL> self . use_pool = True <EOL> class MultiprocNoPoolSortLocalTest ( ResultSortTest ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> def set_mode ( self ) : <EOL> super ( MultiprocNoPoolSortLocalTest , self ) . set_mode ( ) <EOL> self . mode = pypetconstants . WRAP_MODE_LOCAL <EOL> self . multiproc = True <EOL> self . ncores = <NUM_LIT:2> <EOL> self . use_pool = False <EOL> self . gc_interval = <NUM_LIT:2> <EOL> class MultiprocNoPoolLockTest ( EnvironmentTest ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> def set_mode ( self ) : <EOL> super ( MultiprocNoPoolLockTest , self ) . set_mode ( ) <EOL> self . mode = pypetconstants . WRAP_MODE_LOCK <EOL> self . multiproc = True <EOL> self . ncores = <NUM_LIT:2> <EOL> self . use_pool = False <EOL> self . niceness = check_nice ( <NUM_LIT> ) <EOL> class MultiprocNoPoolSortQueueTest ( ResultSortTest ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> def set_mode ( self ) : <EOL> super ( MultiprocNoPoolSortQueueTest , self ) . set_mode ( ) <EOL> self . mode = pypetconstants . WRAP_MODE_QUEUE <EOL> self . multiproc = True <EOL> self . ncores = <NUM_LIT:3> <EOL> self . use_pool = False <EOL> class MultiprocNoPoolSortLockTest ( ResultSortTest ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> def set_mode ( self ) : <EOL> super ( MultiprocNoPoolSortLockTest , self ) . set_mode ( ) <EOL> self . mode = pypetconstants . WRAP_MODE_LOCK <EOL> self . multiproc = True <EOL> self . ncores = <NUM_LIT:3> <EOL> self . use_pool = False <EOL> @ unittest . skipIf ( platform . system ( ) == '<STR_LIT>' , '<STR_LIT>' ) <EOL> class MultiprocNoPoolSortPipeTest ( ResultSortTest ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> def set_mode ( self ) : <EOL> super ( MultiprocNoPoolSortPipeTest , self ) . set_mode ( ) <EOL> self . mode = pypetconstants . WRAP_MODE_PIPE <EOL> self . multiproc = True <EOL> self . ncores = <NUM_LIT:3> <EOL> self . use_pool = False <EOL> class MultiprocFrozenPoolQueueTest ( TestOtherHDF5Settings2 ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> def set_mode ( self ) : <EOL> super ( MultiprocFrozenPoolQueueTest , self ) . set_mode ( ) <EOL> self . mode = pypetconstants . WRAP_MODE_QUEUE <EOL> self . multiproc = True <EOL> self . freeze_input = True <EOL> self . ncores = <NUM_LIT:4> <EOL> self . niceness = check_nice ( <NUM_LIT:1> ) <EOL> self . use_pool = True <EOL> self . gc_interval = <NUM_LIT:2> <EOL> self . graceful_exit = True <EOL> def new_multiply ( traj ) : <EOL> if traj . v_full_copy : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> return multiply ( traj ) <EOL> class MultiprocFrozenPoolSortQueueTest ( ResultSortTest ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> def set_mode ( self ) : <EOL> super ( MultiprocFrozenPoolSortQueueTest , self ) . set_mode ( ) <EOL> self . mode = pypetconstants . WRAP_MODE_QUEUE <EOL> self . multiproc = True <EOL> self . freeze_input = True <EOL> self . ncores = <NUM_LIT:3> <EOL> self . use_pool = True <EOL> self . gc_interval = <NUM_LIT:2> <EOL> def test_if_full_copy_is_old_value ( self ) : <EOL> self . explore ( self . traj ) <EOL> self . traj . v_full_copy = False <EOL> self . env . f_run ( new_multiply ) <EOL> traj = self . traj <EOL> self . assertTrue ( len ( traj ) == len ( compat . listvalues ( self . explore_dict ) [ <NUM_LIT:0> ] ) ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . check_if_z_is_correct ( traj ) <EOL> newtraj = self . load_trajectory ( trajectory_name = self . traj . v_name , as_new = False ) <EOL> self . traj . f_load_skeleton ( ) <EOL> self . traj . f_load_items ( self . traj . f_to_dict ( ) . keys ( ) , only_empties = True ) <EOL> self . compare_trajectories ( self . traj , newtraj ) <EOL> @ unittest . skipIf ( platform . system ( ) == '<STR_LIT>' , '<STR_LIT>' ) <EOL> class MultiprocFrozenPoolPipeTest ( EnvironmentTest ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> def set_mode ( self ) : <EOL> super ( MultiprocFrozenPoolPipeTest , self ) . set_mode ( ) <EOL> self . mode = pypetconstants . WRAP_MODE_PIPE <EOL> self . multiproc = True <EOL> self . freeze_input = True <EOL> self . ncores = <NUM_LIT:4> <EOL> self . use_pool = True <EOL> self . niceness = check_nice ( <NUM_LIT:10> ) <EOL> class MultiprocFrozenPoolLocalTest ( EnvironmentTest ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> def set_mode ( self ) : <EOL> super ( MultiprocFrozenPoolLocalTest , self ) . set_mode ( ) <EOL> self . mode = pypetconstants . WRAP_MODE_LOCAL <EOL> self . multiproc = True <EOL> self . freeze_input = True <EOL> self . ncores = <NUM_LIT:4> <EOL> self . gc_interval = <NUM_LIT:3> <EOL> self . niceness = check_nice ( <NUM_LIT:1> ) <EOL> self . use_pool = True <EOL> class MultiprocFrozenPoolSortLockTest ( ResultSortTest ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> def set_mode ( self ) : <EOL> super ( MultiprocFrozenPoolSortLockTest , self ) . set_mode ( ) <EOL> self . mode = pypetconstants . WRAP_MODE_LOCK <EOL> self . freeze_input = True <EOL> self . multiproc = True <EOL> self . ncores = <NUM_LIT:4> <EOL> self . use_pool = True <EOL> @ unittest . skipIf ( platform . system ( ) == '<STR_LIT>' , '<STR_LIT>' ) <EOL> class MultiprocFrozenPoolSortPipeTest ( ResultSortTest ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> def set_mode ( self ) : <EOL> super ( MultiprocFrozenPoolSortPipeTest , self ) . set_mode ( ) <EOL> self . mode = pypetconstants . WRAP_MODE_PIPE <EOL> self . freeze_input = True <EOL> self . multiproc = True <EOL> self . ncores = <NUM_LIT:4> <EOL> self . use_pool = True <EOL> self . gc_interval = <NUM_LIT:3> <EOL> @ unittest . skipIf ( psutil is None , '<STR_LIT>' ) <EOL> class CapTest ( EnvironmentTest ) : <EOL> tags = '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> cap_count = <NUM_LIT:0> <EOL> def setUp ( self ) : <EOL> self . multiproc = True <EOL> self . mode = '<STR_LIT>' <EOL> self . trajname = make_trajectory_name ( self ) <EOL> self . filename = make_temp_dir ( os . path . join ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' % self . trajname ) ) <EOL> self . logfolder = make_temp_dir ( os . path . join ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> random . seed ( ) <EOL> cap_dicts = ( dict ( cpu_cap = <NUM_LIT> ) , <EOL> dict ( memory_cap = ( <NUM_LIT> , <NUM_LIT> ) ) , <EOL> dict ( swap_cap = <NUM_LIT> , ) ) <EOL> cap_dict = cap_dicts [ CapTest . cap_count ] <EOL> env = Environment ( trajectory = self . trajname , filename = self . filename , <EOL> file_title = self . trajname , log_folder = self . logfolder , <EOL> logger_names = ( '<STR_LIT>' , '<STR_LIT:test>' ) , log_levels = '<STR_LIT>' , <EOL> log_stdout = False , <EOL> results_per_run = <NUM_LIT:5> , <EOL> derived_parameters_per_run = <NUM_LIT:5> , <EOL> multiproc = True , <EOL> ncores = <NUM_LIT:4> , <EOL> use_pool = False , <EOL> niceness = check_nice ( <NUM_LIT:11> ) , <EOL> ** cap_dict ) <EOL> logging . getLogger ( '<STR_LIT:test>' ) . error ( '<STR_LIT>' % ( str ( cap_dict ) , <EOL> str ( self . filename ) ) ) <EOL> CapTest . cap_count += <NUM_LIT:1> <EOL> CapTest . cap_count = CapTest . cap_count % len ( cap_dicts ) <EOL> traj = env . v_trajectory <EOL> self . param_dict = { } <EOL> create_param_dict ( self . param_dict ) <EOL> add_params ( traj , self . param_dict ) <EOL> self . traj = traj <EOL> self . env = env <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> opt_args = parse_args ( ) <EOL> run_suite ( ** opt_args ) </s>
