<s> import unittest <EOL> from locust . core import HttpLocust , Locust , TaskSet , task , events <EOL> from locust import ResponseError , InterruptTaskSet <EOL> from locust . exception import CatchResponseError , RescheduleTask , RescheduleTaskImmediately , LocustError <EOL> from testcases import LocustTestCase , WebserverTestCase <EOL> class TestTaskSet ( LocustTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestTaskSet , self ) . setUp ( ) <EOL> class User ( Locust ) : <EOL> host = "<STR_LIT:127.0.0.1>" <EOL> self . locust = User ( ) <EOL> def test_task_ratio ( self ) : <EOL> t1 = lambda l : None <EOL> t2 = lambda l : None <EOL> class MyTasks ( TaskSet ) : <EOL> tasks = { t1 : <NUM_LIT:5> , t2 : <NUM_LIT:2> } <EOL> l = MyTasks ( self . locust ) <EOL> t1_count = len ( [ t for t in l . tasks if t == t1 ] ) <EOL> t2_count = len ( [ t for t in l . tasks if t == t2 ] ) <EOL> self . assertEqual ( t1_count , <NUM_LIT:5> ) <EOL> self . assertEqual ( t2_count , <NUM_LIT:2> ) <EOL> def test_task_decorator_ratio ( self ) : <EOL> t1 = lambda l : None <EOL> t2 = lambda l : None <EOL> class MyTasks ( TaskSet ) : <EOL> tasks = { t1 : <NUM_LIT:5> , t2 : <NUM_LIT:2> } <EOL> host = "<STR_LIT>" <EOL> @ task ( <NUM_LIT:3> ) <EOL> def t3 ( self ) : <EOL> pass <EOL> @ task ( <NUM_LIT> ) <EOL> def t4 ( self ) : <EOL> pass <EOL> l = MyTasks ( self . locust ) <EOL> t1_count = len ( [ t for t in l . tasks if t == t1 ] ) <EOL> t2_count = len ( [ t for t in l . tasks if t == t2 ] ) <EOL> t3_count = len ( [ t for t in l . tasks if t . __name__ == MyTasks . t3 . __name__ ] ) <EOL> t4_count = len ( [ t for t in l . tasks if t . __name__ == MyTasks . t4 . __name__ ] ) <EOL> self . assertEqual ( t1_count , <NUM_LIT:5> ) <EOL> self . assertEqual ( t2_count , <NUM_LIT:2> ) <EOL> self . assertEqual ( t3_count , <NUM_LIT:3> ) <EOL> self . assertEqual ( t4_count , <NUM_LIT> ) <EOL> def test_on_start ( self ) : <EOL> class MyTasks ( TaskSet ) : <EOL> t1_executed = False <EOL> t2_executed = False <EOL> def on_start ( self ) : <EOL> self . t1 ( ) <EOL> def t1 ( self ) : <EOL> self . t1_executed = True <EOL> @ task <EOL> def t2 ( self ) : <EOL> self . t2_executed = True <EOL> raise InterruptTaskSet ( reschedule = False ) <EOL> l = MyTasks ( self . locust ) <EOL> self . assertRaises ( RescheduleTask , lambda : l . run ( ) ) <EOL> self . assertTrue ( l . t1_executed ) <EOL> self . assertTrue ( l . t2_executed ) <EOL> def test_schedule_task ( self ) : <EOL> self . t1_executed = False <EOL> self . t2_arg = None <EOL> def t1 ( l ) : <EOL> self . t1_executed = True <EOL> def t2 ( l , arg ) : <EOL> self . t2_arg = arg <EOL> class MyTasks ( TaskSet ) : <EOL> tasks = [ t1 , t2 ] <EOL> taskset = MyTasks ( self . locust ) <EOL> taskset . schedule_task ( t1 ) <EOL> taskset . execute_next_task ( ) <EOL> self . assertTrue ( self . t1_executed ) <EOL> taskset . schedule_task ( t2 , args = [ "<STR_LIT>" ] ) <EOL> taskset . execute_next_task ( ) <EOL> self . assertEqual ( "<STR_LIT>" , self . t2_arg ) <EOL> def test_schedule_task_with_kwargs ( self ) : <EOL> class MyTasks ( TaskSet ) : <EOL> @ task <EOL> def t1 ( self ) : <EOL> self . t1_executed = True <EOL> @ task <EOL> def t2 ( self , * args , ** kwargs ) : <EOL> self . t2_args = args <EOL> self . t2_kwargs = kwargs <EOL> loc = MyTasks ( self . locust ) <EOL> loc . schedule_task ( loc . t2 , [ <NUM_LIT> ] , { "<STR_LIT>" : "<STR_LIT:hello>" } ) <EOL> loc . execute_next_task ( ) <EOL> self . assertEqual ( ( <NUM_LIT> , ) , loc . t2_args ) <EOL> self . assertEqual ( { "<STR_LIT>" : "<STR_LIT:hello>" } , loc . t2_kwargs ) <EOL> loc . schedule_task ( loc . t2 , args = [ <NUM_LIT:10> , <NUM_LIT:4> ] , kwargs = { "<STR_LIT>" : <NUM_LIT:1> , "<STR_LIT>" : <NUM_LIT:2> } ) <EOL> loc . execute_next_task ( ) <EOL> self . assertEqual ( ( <NUM_LIT:10> , <NUM_LIT:4> ) , loc . t2_args ) <EOL> self . assertEqual ( { "<STR_LIT>" : <NUM_LIT:1> , "<STR_LIT>" : <NUM_LIT:2> } , loc . t2_kwargs ) <EOL> def test_schedule_task_bound_method ( self ) : <EOL> class MyTasks ( TaskSet ) : <EOL> host = "<STR_LIT>" <EOL> @ task ( ) <EOL> def t1 ( self ) : <EOL> self . t1_executed = True <EOL> self . schedule_task ( self . t2 ) <EOL> def t2 ( self ) : <EOL> self . t2_executed = True <EOL> taskset = MyTasks ( self . locust ) <EOL> taskset . schedule_task ( taskset . get_next_task ( ) ) <EOL> taskset . execute_next_task ( ) <EOL> self . assertTrue ( taskset . t1_executed ) <EOL> taskset . execute_next_task ( ) <EOL> self . assertTrue ( taskset . t2_executed ) <EOL> def test_taskset_inheritance ( self ) : <EOL> def t1 ( l ) : <EOL> pass <EOL> class MyBaseTaskSet ( TaskSet ) : <EOL> tasks = [ t1 ] <EOL> host = "<STR_LIT>" <EOL> class MySubTaskSet ( MyBaseTaskSet ) : <EOL> @ task <EOL> def t2 ( self ) : <EOL> pass <EOL> l = MySubTaskSet ( self . locust ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( l . tasks ) ) <EOL> self . assertEqual ( [ t1 , MySubTaskSet . t2 . __func__ ] , l . tasks ) <EOL> def test_task_decorator_with_or_without_argument ( self ) : <EOL> class MyTaskSet ( TaskSet ) : <EOL> @ task <EOL> def t1 ( self ) : <EOL> pass <EOL> taskset = MyTaskSet ( self . locust ) <EOL> self . assertEqual ( len ( taskset . tasks ) , <NUM_LIT:1> ) <EOL> class MyTaskSet2 ( TaskSet ) : <EOL> @ task ( ) <EOL> def t1 ( self ) : <EOL> pass <EOL> taskset = MyTaskSet2 ( self . locust ) <EOL> self . assertEqual ( len ( taskset . tasks ) , <NUM_LIT:1> ) <EOL> class MyTaskSet3 ( TaskSet ) : <EOL> @ task ( <NUM_LIT:3> ) <EOL> def t1 ( self ) : <EOL> pass <EOL> taskset = MyTaskSet3 ( self . locust ) <EOL> self . assertEqual ( len ( taskset . tasks ) , <NUM_LIT:3> ) <EOL> def test_sub_taskset ( self ) : <EOL> class MySubTaskSet ( TaskSet ) : <EOL> min_wait = <NUM_LIT:1> <EOL> max_wait = <NUM_LIT:1> <EOL> @ task ( ) <EOL> def a_task ( self ) : <EOL> self . locust . sub_locust_task_executed = True <EOL> self . interrupt ( ) <EOL> class MyTaskSet ( TaskSet ) : <EOL> tasks = [ MySubTaskSet ] <EOL> self . sub_locust_task_executed = False <EOL> loc = MyTaskSet ( self . locust ) <EOL> loc . schedule_task ( loc . get_next_task ( ) ) <EOL> self . assertRaises ( RescheduleTaskImmediately , lambda : loc . execute_next_task ( ) ) <EOL> self . assertTrue ( self . locust . sub_locust_task_executed ) <EOL> def test_sub_taskset_tasks_decorator ( self ) : <EOL> class MyTaskSet ( TaskSet ) : <EOL> @ task <EOL> class MySubTaskSet ( TaskSet ) : <EOL> min_wait = <NUM_LIT:1> <EOL> max_wait = <NUM_LIT:1> <EOL> @ task ( ) <EOL> def a_task ( self ) : <EOL> self . locust . sub_locust_task_executed = True <EOL> self . interrupt ( ) <EOL> self . sub_locust_task_executed = False <EOL> loc = MyTaskSet ( self . locust ) <EOL> loc . schedule_task ( loc . get_next_task ( ) ) <EOL> self . assertRaises ( RescheduleTaskImmediately , lambda : loc . execute_next_task ( ) ) <EOL> self . assertTrue ( self . locust . sub_locust_task_executed ) <EOL> def test_sub_taskset_arguments ( self ) : <EOL> class MySubTaskSet ( TaskSet ) : <EOL> min_wait = <NUM_LIT:1> <EOL> max_wait = <NUM_LIT:1> <EOL> @ task ( ) <EOL> def a_task ( self ) : <EOL> self . locust . sub_taskset_args = self . args <EOL> self . locust . sub_taskset_kwargs = self . kwargs <EOL> self . interrupt ( ) <EOL> class MyTaskSet ( TaskSet ) : <EOL> sub_locust_args = None <EOL> sub_locust_kwargs = None <EOL> tasks = [ MySubTaskSet ] <EOL> self . locust . sub_taskset_args = None <EOL> self . locust . sub_taskset_kwargs = None <EOL> loc = MyTaskSet ( self . locust ) <EOL> loc . schedule_task ( MySubTaskSet , args = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] , kwargs = { "<STR_LIT:hello>" : "<STR_LIT>" } ) <EOL> self . assertRaises ( RescheduleTaskImmediately , lambda : loc . execute_next_task ( ) ) <EOL> self . assertEqual ( ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ) , self . locust . sub_taskset_args ) <EOL> self . assertEqual ( { "<STR_LIT:hello>" : "<STR_LIT>" } , self . locust . sub_taskset_kwargs ) <EOL> def test_interrupt_taskset_in_main_taskset ( self ) : <EOL> class MyTaskSet ( TaskSet ) : <EOL> @ task <EOL> def interrupted_task ( self ) : <EOL> raise InterruptTaskSet ( reschedule = False ) <EOL> class MyLocust ( Locust ) : <EOL> host = "<STR_LIT>" <EOL> task_set = MyTaskSet <EOL> class MyTaskSet2 ( TaskSet ) : <EOL> @ task <EOL> def interrupted_task ( self ) : <EOL> self . interrupt ( ) <EOL> class MyLocust2 ( Locust ) : <EOL> host = "<STR_LIT>" <EOL> task_set = MyTaskSet2 <EOL> l = MyLocust ( ) <EOL> l2 = MyLocust2 ( ) <EOL> self . assertRaises ( LocustError , lambda : l . run ( ) ) <EOL> self . assertRaises ( LocustError , lambda : l2 . run ( ) ) <EOL> try : <EOL> l . run ( ) <EOL> except LocustError as e : <EOL> self . assertTrue ( "<STR_LIT>" in e . args [ <NUM_LIT:0> ] , "<STR_LIT>" ) <EOL> self . assertTrue ( "<STR_LIT>" in e . args [ <NUM_LIT:0> ] , "<STR_LIT>" ) <EOL> except : <EOL> raise <EOL> try : <EOL> l2 . run ( ) <EOL> except LocustError as e : <EOL> self . assertTrue ( "<STR_LIT>" in e . args [ <NUM_LIT:0> ] , "<STR_LIT>" ) <EOL> self . assertTrue ( "<STR_LIT>" in e . args [ <NUM_LIT:0> ] , "<STR_LIT>" ) <EOL> except : <EOL> raise <EOL> def test_on_start_interrupt ( self ) : <EOL> class SubTaskSet ( TaskSet ) : <EOL> def on_start ( self ) : <EOL> if self . kwargs [ "<STR_LIT>" ] : <EOL> self . interrupt ( reschedule = True ) <EOL> else : <EOL> self . interrupt ( reschedule = False ) <EOL> class MyLocust ( Locust ) : <EOL> host = "<STR_LIT>" <EOL> task_set = SubTaskSet <EOL> l = MyLocust ( ) <EOL> task_set = SubTaskSet ( l ) <EOL> self . assertRaises ( RescheduleTaskImmediately , lambda : task_set . run ( reschedule = True ) ) <EOL> self . assertRaises ( RescheduleTask , lambda : task_set . run ( reschedule = False ) ) <EOL> def test_parent_attribute ( self ) : <EOL> from locust . exception import StopLocust <EOL> parents = { } <EOL> class SubTaskSet ( TaskSet ) : <EOL> def on_start ( self ) : <EOL> parents [ "<STR_LIT>" ] = self . parent <EOL> @ task <EOL> class SubSubTaskSet ( TaskSet ) : <EOL> def on_start ( self ) : <EOL> parents [ "<STR_LIT>" ] = self . parent <EOL> @ task <EOL> def stop ( self ) : <EOL> raise StopLocust ( ) <EOL> class RootTaskSet ( TaskSet ) : <EOL> tasks = [ SubTaskSet ] <EOL> class MyLocust ( Locust ) : <EOL> host = "<STR_LIT>" <EOL> task_set = RootTaskSet <EOL> l = MyLocust ( ) <EOL> l . run ( ) <EOL> self . assertTrue ( isinstance ( parents [ "<STR_LIT>" ] , RootTaskSet ) ) <EOL> self . assertTrue ( isinstance ( parents [ "<STR_LIT>" ] , SubTaskSet ) ) <EOL> class TestWebLocustClass ( WebserverTestCase ) : <EOL> def test_get_request ( self ) : <EOL> self . response = "<STR_LIT>" <EOL> def t1 ( l ) : <EOL> self . response = l . client . get ( "<STR_LIT>" ) <EOL> class MyLocust ( HttpLocust ) : <EOL> tasks = [ t1 ] <EOL> host = "<STR_LIT>" % self . port <EOL> my_locust = MyLocust ( ) <EOL> t1 ( my_locust ) <EOL> self . assertEqual ( self . response . content , "<STR_LIT>" ) <EOL> def test_client_request_headers ( self ) : <EOL> class MyLocust ( HttpLocust ) : <EOL> host = "<STR_LIT>" % self . port <EOL> locust = MyLocust ( ) <EOL> self . assertEqual ( "<STR_LIT:hello>" , locust . client . get ( "<STR_LIT>" , headers = { "<STR_LIT>" : "<STR_LIT:hello>" } ) . content ) <EOL> def test_client_get ( self ) : <EOL> class MyLocust ( HttpLocust ) : <EOL> host = "<STR_LIT>" % self . port <EOL> locust = MyLocust ( ) <EOL> self . assertEqual ( "<STR_LIT:GET>" , locust . client . get ( "<STR_LIT>" ) . content ) <EOL> def test_client_get_absolute_url ( self ) : <EOL> class MyLocust ( HttpLocust ) : <EOL> host = "<STR_LIT>" % self . port <EOL> locust = MyLocust ( ) <EOL> self . assertEqual ( "<STR_LIT:GET>" , locust . client . get ( "<STR_LIT>" % self . port ) . content ) <EOL> def test_client_post ( self ) : <EOL> class MyLocust ( HttpLocust ) : <EOL> host = "<STR_LIT>" % self . port <EOL> locust = MyLocust ( ) <EOL> self . assertEqual ( "<STR_LIT:POST>" , locust . client . post ( "<STR_LIT>" , { "<STR_LIT>" : "<STR_LIT>" } ) . content ) <EOL> self . assertEqual ( "<STR_LIT>" , locust . client . post ( "<STR_LIT>" , { "<STR_LIT>" : "<STR_LIT>" } ) . content ) <EOL> def test_client_put ( self ) : <EOL> class MyLocust ( HttpLocust ) : <EOL> host = "<STR_LIT>" % self . port <EOL> locust = MyLocust ( ) <EOL> self . assertEqual ( "<STR_LIT>" , locust . client . put ( "<STR_LIT>" , { "<STR_LIT>" : "<STR_LIT>" } ) . content ) <EOL> self . assertEqual ( "<STR_LIT>" , locust . client . put ( "<STR_LIT>" , { "<STR_LIT>" : "<STR_LIT>" } ) . content ) <EOL> def test_client_delete ( self ) : <EOL> class MyLocust ( HttpLocust ) : <EOL> host = "<STR_LIT>" % self . port <EOL> locust = MyLocust ( ) <EOL> self . assertEqual ( "<STR_LIT>" , locust . client . delete ( "<STR_LIT>" ) . content ) <EOL> self . assertEqual ( <NUM_LIT:200> , locust . client . delete ( "<STR_LIT>" ) . status_code ) <EOL> def test_client_head ( self ) : <EOL> class MyLocust ( HttpLocust ) : <EOL> host = "<STR_LIT>" % self . port <EOL> locust = MyLocust ( ) <EOL> self . assertEqual ( <NUM_LIT:200> , locust . client . head ( "<STR_LIT>" ) . status_code ) <EOL> def test_client_basic_auth ( self ) : <EOL> class MyLocust ( HttpLocust ) : <EOL> host = "<STR_LIT>" % self . port <EOL> class MyAuthorizedLocust ( HttpLocust ) : <EOL> host = "<STR_LIT>" % self . port <EOL> class MyUnauthorizedLocust ( HttpLocust ) : <EOL> host = "<STR_LIT>" % self . port <EOL> locust = MyLocust ( ) <EOL> unauthorized = MyUnauthorizedLocust ( ) <EOL> authorized = MyAuthorizedLocust ( ) <EOL> self . assertEqual ( "<STR_LIT>" , authorized . client . get ( "<STR_LIT>" ) . content ) <EOL> self . assertFalse ( locust . client . get ( "<STR_LIT>" ) ) <EOL> self . assertFalse ( unauthorized . client . get ( "<STR_LIT>" ) ) <EOL> def test_log_request_name_argument ( self ) : <EOL> from locust . stats import RequestStats , global_stats <EOL> self . response = "<STR_LIT>" <EOL> class MyLocust ( HttpLocust ) : <EOL> tasks = [ ] <EOL> host = "<STR_LIT>" % self . port <EOL> @ task ( ) <EOL> def t1 ( l ) : <EOL> self . response = l . client . get ( "<STR_LIT>" , name = "<STR_LIT>" ) <EOL> my_locust = MyLocust ( ) <EOL> my_locust . t1 ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , global_stats . get ( "<STR_LIT>" , "<STR_LIT:GET>" ) . num_requests ) <EOL> self . assertEqual ( <NUM_LIT:0> , global_stats . get ( "<STR_LIT>" , "<STR_LIT:GET>" ) . num_requests ) <EOL> def test_locust_client_error ( self ) : <EOL> class MyTaskSet ( TaskSet ) : <EOL> @ task <EOL> def t1 ( self ) : <EOL> self . client . get ( "<STR_LIT:/>" ) <EOL> self . interrupt ( ) <EOL> class MyLocust ( Locust ) : <EOL> host = "<STR_LIT>" % self . port <EOL> task_set = MyTaskSet <EOL> my_locust = MyLocust ( ) <EOL> self . assertRaises ( LocustError , lambda : my_locust . client . get ( "<STR_LIT:/>" ) ) <EOL> my_taskset = MyTaskSet ( my_locust ) <EOL> self . assertRaises ( LocustError , lambda : my_taskset . client . get ( "<STR_LIT:/>" ) ) <EOL> def test_redirect_url_original_path_as_name ( self ) : <EOL> class MyLocust ( HttpLocust ) : <EOL> host = "<STR_LIT>" % self . port <EOL> l = MyLocust ( ) <EOL> l . client . get ( "<STR_LIT>" ) <EOL> from locust . stats import global_stats <EOL> self . assertEqual ( <NUM_LIT:1> , len ( global_stats . entries ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , global_stats . get ( "<STR_LIT>" , "<STR_LIT:GET>" ) . num_requests ) <EOL> self . assertEqual ( <NUM_LIT:0> , global_stats . get ( "<STR_LIT>" , "<STR_LIT:GET>" ) . num_requests ) <EOL> class TestCatchResponse ( WebserverTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestCatchResponse , self ) . setUp ( ) <EOL> class MyLocust ( HttpLocust ) : <EOL> host = "<STR_LIT>" % self . port <EOL> self . locust = MyLocust ( ) <EOL> self . num_failures = <NUM_LIT:0> <EOL> self . num_success = <NUM_LIT:0> <EOL> def on_failure ( request_type , name , response_time , exception ) : <EOL> self . num_failures += <NUM_LIT:1> <EOL> self . last_failure_exception = exception <EOL> def on_success ( ** kwargs ) : <EOL> self . num_success += <NUM_LIT:1> <EOL> events . request_failure += on_failure <EOL> events . request_success += on_success <EOL> def test_catch_response ( self ) : <EOL> self . assertEqual ( <NUM_LIT> , self . locust . client . get ( "<STR_LIT>" ) . status_code ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . num_failures ) <EOL> self . assertEqual ( <NUM_LIT:0> , self . num_success ) <EOL> with self . locust . client . get ( "<STR_LIT>" , catch_response = True ) as response : pass <EOL> self . assertEqual ( <NUM_LIT:1> , self . num_failures ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . num_success ) <EOL> with self . locust . client . get ( "<STR_LIT>" , catch_response = True ) as response : <EOL> raise ResponseError ( "<STR_LIT>" ) <EOL> self . assertEqual ( <NUM_LIT:2> , self . num_failures ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . num_success ) <EOL> def test_catch_response_http_fail ( self ) : <EOL> with self . locust . client . get ( "<STR_LIT>" , catch_response = True ) as response : pass <EOL> self . assertEqual ( <NUM_LIT:1> , self . num_failures ) <EOL> self . assertEqual ( <NUM_LIT:0> , self . num_success ) <EOL> def test_catch_response_http_manual_fail ( self ) : <EOL> with self . locust . client . get ( "<STR_LIT>" , catch_response = True ) as response : <EOL> response . failure ( "<STR_LIT>" ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . num_failures ) <EOL> self . assertEqual ( <NUM_LIT:0> , self . num_success ) <EOL> self . assertTrue ( <EOL> isinstance ( self . last_failure_exception , CatchResponseError ) , <EOL> "<STR_LIT>" <EOL> ) <EOL> def test_catch_response_http_manual_success ( self ) : <EOL> with self . locust . client . get ( "<STR_LIT>" , catch_response = True ) as response : <EOL> response . success ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , self . num_failures ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . num_success ) <EOL> def test_catch_response_allow_404 ( self ) : <EOL> with self . locust . client . get ( "<STR_LIT>" , catch_response = True ) as response : <EOL> self . assertEqual ( <NUM_LIT> , response . status_code ) <EOL> if response . status_code == <NUM_LIT> : <EOL> response . success ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , self . num_failures ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . num_success ) <EOL> def test_interrupt_taskset_with_catch_response ( self ) : <EOL> class MyTaskSet ( TaskSet ) : <EOL> @ task <EOL> def interrupted_task ( self ) : <EOL> with self . client . get ( "<STR_LIT>" , catch_response = True ) as r : <EOL> raise InterruptTaskSet ( ) <EOL> class MyLocust ( HttpLocust ) : <EOL> host = "<STR_LIT>" % self . port <EOL> task_set = MyTaskSet <EOL> l = MyLocust ( ) <EOL> ts = MyTaskSet ( l ) <EOL> self . assertRaises ( InterruptTaskSet , lambda : ts . interrupted_task ( ) ) <EOL> self . assertEqual ( <NUM_LIT:0> , self . num_failures ) <EOL> self . assertEqual ( <NUM_LIT:0> , self . num_success ) <EOL> def test_catch_response_connection_error_success ( self ) : <EOL> class MyLocust ( HttpLocust ) : <EOL> host = "<STR_LIT>" <EOL> l = MyLocust ( ) <EOL> with l . client . get ( "<STR_LIT:/>" , catch_response = True ) as r : <EOL> self . assertEqual ( r . status_code , <NUM_LIT:0> ) <EOL> self . assertEqual ( None , r . content ) <EOL> r . success ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . num_success ) <EOL> self . assertEqual ( <NUM_LIT:0> , self . num_failures ) <EOL> def test_catch_response_connection_error_fail ( self ) : <EOL> class MyLocust ( HttpLocust ) : <EOL> host = "<STR_LIT>" <EOL> l = MyLocust ( ) <EOL> with l . client . get ( "<STR_LIT:/>" , catch_response = True ) as r : <EOL> self . assertEqual ( r . status_code , <NUM_LIT:0> ) <EOL> self . assertEqual ( None , r . content ) <EOL> r . success ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . num_success ) <EOL> self . assertEqual ( <NUM_LIT:0> , self . num_failures ) </s>
