<s> import jsonschema <EOL> import mock <EOL> from rally . plugins . common . runners import constant <EOL> from rally . task import runner <EOL> from tests . unit import fakes <EOL> from tests . unit import test <EOL> RUNNERS_BASE = "<STR_LIT>" <EOL> RUNNERS = "<STR_LIT>" <EOL> class ConstantScenarioRunnerTestCase ( test . TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( ConstantScenarioRunnerTestCase , self ) . setUp ( ) <EOL> self . config = { "<STR_LIT>" : <NUM_LIT:4> , "<STR_LIT>" : <NUM_LIT:2> , <EOL> "<STR_LIT>" : <NUM_LIT:2> , "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT:2> } <EOL> self . context = fakes . FakeContext ( { "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } ) . context <EOL> self . args = { "<STR_LIT:a>" : <NUM_LIT:1> } <EOL> self . task = mock . MagicMock ( ) <EOL> def test_validate ( self ) : <EOL> constant . ConstantScenarioRunner . validate ( self . config ) <EOL> def test_validate_failed ( self ) : <EOL> self . config [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> self . assertRaises ( jsonschema . ValidationError , <EOL> runner . ScenarioRunner . validate , <EOL> self . config ) <EOL> @ mock . patch ( RUNNERS + "<STR_LIT>" ) <EOL> def test__run_scenario_once_with_unpack_args ( self , mock_runner ) : <EOL> result = constant . _run_scenario_once_with_unpack_args ( <EOL> ( "<STR_LIT>" , ( "<STR_LIT>" , "<STR_LIT>" ) ) ) <EOL> self . assertEqual ( mock_runner . _run_scenario_once . return_value , result ) <EOL> mock_runner . _run_scenario_once . assert_called_once_with ( <EOL> "<STR_LIT>" , ( "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> @ mock . patch ( RUNNERS + "<STR_LIT>" ) <EOL> @ mock . patch ( RUNNERS + "<STR_LIT>" ) <EOL> @ mock . patch ( RUNNERS + "<STR_LIT>" ) <EOL> @ mock . patch ( RUNNERS + "<STR_LIT>" ) <EOL> def test__worker_process ( self , mock_runner , mock_queue , mock_thread , <EOL> mock_time ) : <EOL> mock_thread_instance = mock . MagicMock ( <EOL> isAlive = mock . MagicMock ( return_value = False ) ) <EOL> mock_thread . return_value = mock_thread_instance <EOL> mock_event = mock . MagicMock ( <EOL> is_set = mock . MagicMock ( return_value = False ) ) <EOL> times = <NUM_LIT:4> <EOL> fake_ram_int = iter ( range ( <NUM_LIT:10> ) ) <EOL> context = { "<STR_LIT>" : [ { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:id>" : "<STR_LIT>" } ] } <EOL> info = { "<STR_LIT>" : <NUM_LIT:1> , "<STR_LIT>" : <NUM_LIT:1> } <EOL> constant . _worker_process ( mock_queue , fake_ram_int , <NUM_LIT:1> , <NUM_LIT:2> , times , <EOL> context , "<STR_LIT>" , "<STR_LIT>" , ( ) , mock_event , <EOL> info ) <EOL> self . assertEqual ( times + <NUM_LIT:1> , mock_thread . call_count ) <EOL> self . assertEqual ( times + <NUM_LIT:1> , mock_thread_instance . start . call_count ) <EOL> self . assertEqual ( times + <NUM_LIT:1> , mock_thread_instance . join . call_count ) <EOL> self . assertEqual ( times , mock_runner . _get_scenario_context . call_count ) <EOL> for i in range ( times ) : <EOL> scenario_context = mock_runner . _get_scenario_context ( i , context ) <EOL> call = mock . call ( <EOL> args = ( mock_queue , "<STR_LIT>" , "<STR_LIT>" , scenario_context , ( ) ) , <EOL> target = mock_runner . _worker_thread , <EOL> ) <EOL> self . assertIn ( call , mock_thread . mock_calls ) <EOL> @ mock . patch ( RUNNERS_BASE + "<STR_LIT>" ) <EOL> def test__worker_thread ( self , mock__run_scenario_once ) : <EOL> mock_queue = mock . MagicMock ( ) <EOL> args = ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , { } ) <EOL> runner . _worker_thread ( mock_queue , * args ) <EOL> self . assertEqual ( <NUM_LIT:1> , mock_queue . put . call_count ) <EOL> expected_calls = [ mock . call ( * args ) ] <EOL> self . assertEqual ( expected_calls , mock__run_scenario_once . mock_calls ) <EOL> def test__run_scenario ( self ) : <EOL> runner_obj = constant . ConstantScenarioRunner ( self . task , self . config ) <EOL> runner_obj . _run_scenario ( <EOL> fakes . FakeScenario , "<STR_LIT>" , self . context , self . args ) <EOL> self . assertEqual ( len ( runner_obj . result_queue ) , self . config [ "<STR_LIT>" ] ) <EOL> for result_batch in runner_obj . result_queue : <EOL> for result in result_batch : <EOL> self . assertIsNotNone ( result ) <EOL> def test__run_scenario_exception ( self ) : <EOL> runner_obj = constant . ConstantScenarioRunner ( self . task , self . config ) <EOL> runner_obj . _run_scenario ( fakes . FakeScenario , "<STR_LIT>" , <EOL> self . context , self . args ) <EOL> self . assertEqual ( len ( runner_obj . result_queue ) , self . config [ "<STR_LIT>" ] ) <EOL> for result_batch in runner_obj . result_queue : <EOL> for result in result_batch : <EOL> self . assertIsNotNone ( result ) <EOL> self . assertIn ( "<STR_LIT:error>" , runner_obj . result_queue [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) <EOL> def test__run_scenario_aborted ( self ) : <EOL> runner_obj = constant . ConstantScenarioRunner ( self . task , self . config ) <EOL> runner_obj . abort ( ) <EOL> runner_obj . _run_scenario ( fakes . FakeScenario , "<STR_LIT>" , self . context , <EOL> self . args ) <EOL> self . assertEqual ( len ( runner_obj . result_queue ) , <NUM_LIT:0> ) <EOL> @ mock . patch ( RUNNERS + "<STR_LIT>" ) <EOL> @ mock . patch ( RUNNERS + "<STR_LIT>" ) <EOL> @ mock . patch ( RUNNERS + "<STR_LIT>" ) <EOL> @ mock . patch ( RUNNERS + <EOL> "<STR_LIT>" ) <EOL> @ mock . patch ( RUNNERS + "<STR_LIT>" ) <EOL> def test_that_cpu_count_is_adjusted_properly ( <EOL> self , <EOL> mock__join_processes , <EOL> mock__create_process_pool , <EOL> mock__log_debug_info , <EOL> mock_cpu_count , mock_queue ) : <EOL> samples = [ <EOL> { <EOL> "<STR_LIT:input>" : { "<STR_LIT>" : <NUM_LIT:20> , "<STR_LIT>" : <NUM_LIT:20> , "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT:1> } , <EOL> "<STR_LIT>" : <NUM_LIT:2> , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : <NUM_LIT:20> , <EOL> "<STR_LIT>" : <NUM_LIT:0> , <EOL> } <EOL> } , <EOL> { <EOL> "<STR_LIT:input>" : { "<STR_LIT>" : <NUM_LIT:20> , "<STR_LIT>" : <NUM_LIT:15> , "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT:3> } , <EOL> "<STR_LIT>" : <NUM_LIT:2> , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : <NUM_LIT:2> , <EOL> "<STR_LIT>" : <NUM_LIT:2> , <EOL> "<STR_LIT>" : <NUM_LIT:7> , <EOL> "<STR_LIT>" : <NUM_LIT:1> , <EOL> } <EOL> } , <EOL> { <EOL> "<STR_LIT:input>" : { "<STR_LIT>" : <NUM_LIT:20> , "<STR_LIT>" : <NUM_LIT:1> , "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT:3> } , <EOL> "<STR_LIT>" : <NUM_LIT:2> , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : <NUM_LIT:2> , <EOL> "<STR_LIT>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : <NUM_LIT:0> , <EOL> } <EOL> } , <EOL> { <EOL> "<STR_LIT:input>" : { "<STR_LIT>" : <NUM_LIT:2> , "<STR_LIT>" : <NUM_LIT:5> , "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT:4> } , <EOL> "<STR_LIT>" : <NUM_LIT:4> , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : <NUM_LIT:4> , <EOL> "<STR_LIT>" : <NUM_LIT:2> , <EOL> "<STR_LIT>" : <NUM_LIT:2> , <EOL> "<STR_LIT>" : <NUM_LIT:1> , <EOL> } <EOL> } <EOL> ] <EOL> for sample in samples : <EOL> mock__log_debug_info . reset_mock ( ) <EOL> mock_cpu_count . reset_mock ( ) <EOL> mock__create_process_pool . reset_mock ( ) <EOL> mock__join_processes . reset_mock ( ) <EOL> mock_queue . reset_mock ( ) <EOL> mock_cpu_count . return_value = sample [ "<STR_LIT>" ] <EOL> runner_obj = constant . ConstantScenarioRunner ( self . task , <EOL> sample [ "<STR_LIT:input>" ] ) <EOL> runner_obj . _run_scenario ( fakes . FakeScenario , "<STR_LIT>" , self . context , <EOL> self . args ) <EOL> mock_cpu_count . assert_called_once_with ( ) <EOL> mock__log_debug_info . assert_called_once_with ( <EOL> times = sample [ "<STR_LIT:input>" ] [ "<STR_LIT>" ] , <EOL> concurrency = sample [ "<STR_LIT:input>" ] [ "<STR_LIT>" ] , <EOL> timeout = <NUM_LIT:0> , <EOL> max_cpu_used = sample [ "<STR_LIT>" ] [ "<STR_LIT>" ] , <EOL> processes_to_start = sample [ "<STR_LIT>" ] [ "<STR_LIT>" ] , <EOL> concurrency_per_worker = ( <EOL> sample [ "<STR_LIT>" ] [ "<STR_LIT>" ] ) , <EOL> concurrency_overhead = ( <EOL> sample [ "<STR_LIT>" ] [ "<STR_LIT>" ] ) ) <EOL> args , kwargs = mock__create_process_pool . call_args <EOL> self . assertIn ( sample [ "<STR_LIT>" ] [ "<STR_LIT>" ] , args ) <EOL> self . assertIn ( constant . _worker_process , args ) <EOL> mock__join_processes . assert_called_once_with ( <EOL> mock__create_process_pool . return_value , <EOL> mock_queue . return_value ) <EOL> def test_abort ( self ) : <EOL> runner_obj = constant . ConstantScenarioRunner ( self . task , self . config ) <EOL> self . assertFalse ( runner_obj . aborted . is_set ( ) ) <EOL> runner_obj . abort ( ) <EOL> self . assertTrue ( runner_obj . aborted . is_set ( ) ) <EOL> class ConstantForDurationScenarioRunnerTestCase ( test . TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( ConstantForDurationScenarioRunnerTestCase , self ) . setUp ( ) <EOL> self . config = { "<STR_LIT>" : <NUM_LIT:0> , "<STR_LIT>" : <NUM_LIT:2> , <EOL> "<STR_LIT>" : <NUM_LIT:2> , "<STR_LIT:type>" : "<STR_LIT>" } <EOL> self . context = fakes . FakeContext ( { "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } ) . context <EOL> self . context [ "<STR_LIT>" ] = <NUM_LIT> <EOL> self . args = { "<STR_LIT:a>" : <NUM_LIT:1> } <EOL> def test_validate ( self ) : <EOL> constant . ConstantForDurationScenarioRunner . validate ( self . config ) <EOL> def test_validate_failed ( self ) : <EOL> self . config [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> self . assertRaises ( jsonschema . ValidationError , <EOL> runner . ScenarioRunner . validate , <EOL> self . config ) <EOL> def test_run_scenario_constantly_for_duration ( self ) : <EOL> runner_obj = constant . ConstantForDurationScenarioRunner ( <EOL> None , self . config ) <EOL> runner_obj . _run_scenario ( fakes . FakeScenario , "<STR_LIT>" , <EOL> self . context , self . args ) <EOL> expected_times = <NUM_LIT:1> <EOL> self . assertEqual ( len ( runner_obj . result_queue ) , expected_times ) <EOL> for result_batch in runner_obj . result_queue : <EOL> for result in result_batch : <EOL> self . assertIsNotNone ( result ) <EOL> def test_run_scenario_constantly_for_duration_exception ( self ) : <EOL> runner_obj = constant . ConstantForDurationScenarioRunner ( <EOL> None , self . config ) <EOL> runner_obj . _run_scenario ( fakes . FakeScenario , "<STR_LIT>" , <EOL> self . context , self . args ) <EOL> expected_times = <NUM_LIT:1> <EOL> self . assertEqual ( len ( runner_obj . result_queue ) , expected_times ) <EOL> for result_batch in runner_obj . result_queue : <EOL> for result in result_batch : <EOL> self . assertIsNotNone ( result ) <EOL> self . assertIn ( "<STR_LIT:error>" , runner_obj . result_queue [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) <EOL> def test_run_scenario_constantly_for_duration_timeout ( self ) : <EOL> runner_obj = constant . ConstantForDurationScenarioRunner ( <EOL> None , self . config ) <EOL> runner_obj . _run_scenario ( fakes . FakeScenario , "<STR_LIT>" , <EOL> self . context , self . args ) <EOL> expected_times = <NUM_LIT:1> <EOL> self . assertEqual ( len ( runner_obj . result_queue ) , expected_times ) <EOL> for result_batch in runner_obj . result_queue : <EOL> for result in result_batch : <EOL> self . assertIsNotNone ( result ) <EOL> self . assertIn ( "<STR_LIT:error>" , runner_obj . result_queue [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) <EOL> def test__run_scenario_constantly_aborted ( self ) : <EOL> runner_obj = constant . ConstantForDurationScenarioRunner ( None , <EOL> self . config ) <EOL> runner_obj . abort ( ) <EOL> runner_obj . _run_scenario ( fakes . FakeScenario , "<STR_LIT>" , <EOL> self . context , self . args ) <EOL> self . assertEqual ( len ( runner_obj . result_queue ) , <NUM_LIT:0> ) <EOL> def test_abort ( self ) : <EOL> runner_obj = constant . ConstantForDurationScenarioRunner ( None , <EOL> self . config ) <EOL> self . assertFalse ( runner_obj . aborted . is_set ( ) ) <EOL> runner_obj . abort ( ) <EOL> self . assertTrue ( runner_obj . aborted . is_set ( ) ) </s>
