<s> import threading <EOL> __author__ = '<STR_LIT>' <EOL> import os <EOL> import tempfile <EOL> import logging <EOL> import sys <EOL> root = logging . getLogger ( ) <EOL> root . setLevel ( logging . DEBUG ) <EOL> ch = logging . StreamHandler ( sys . stdout ) <EOL> ch . setLevel ( logging . DEBUG ) <EOL> formatter = logging . Formatter ( '<STR_LIT>' ) <EOL> ch . setFormatter ( formatter ) <EOL> root . addHandler ( ch ) <EOL> from scalyr_agent . configuration import Configuration <EOL> from scalyr_agent . copying_manager import CopyingParameters , CopyingManager <EOL> from scalyr_agent . platform_controller import DefaultPaths <EOL> from scalyr_agent . scalyr_client import AddEventsRequest <EOL> from scalyr_agent . test_base import ScalyrTestCase <EOL> from scalyr_agent . json_lib import JsonObject , JsonArray <EOL> from scalyr_agent import json_lib <EOL> ONE_MB = <NUM_LIT> * <NUM_LIT> <EOL> class CopyingParamsTest ( ScalyrTestCase ) : <EOL> def setUp ( self ) : <EOL> self . __config_dir = tempfile . mkdtemp ( ) <EOL> self . __config_file = os . path . join ( self . __config_dir , '<STR_LIT>' ) <EOL> self . __config_fragments_dir = os . path . join ( self . __config_dir , '<STR_LIT>' ) <EOL> os . makedirs ( self . __config_fragments_dir ) <EOL> fp = open ( self . __config_file , '<STR_LIT:w>' ) <EOL> fp . write ( '<STR_LIT>' ) <EOL> fp . close ( ) <EOL> config = self . __create_test_configuration_instance ( ) <EOL> config . parse ( ) <EOL> self . test_params = CopyingParameters ( config ) <EOL> def test_initial_settings ( self ) : <EOL> self . assertEquals ( self . test_params . current_bytes_allowed_to_send , ONE_MB ) <EOL> self . assertEquals ( self . test_params . current_sleep_interval , <NUM_LIT> ) <EOL> def test_no_events_being_sent ( self ) : <EOL> for i in range ( <NUM_LIT:0> , <NUM_LIT:5> ) : <EOL> self . test_params . update_params ( '<STR_LIT:success>' , <NUM_LIT:0> ) <EOL> self . assertEquals ( self . test_params . current_bytes_allowed_to_send , ONE_MB ) <EOL> self . assertEquals ( self . test_params . current_sleep_interval , <NUM_LIT> ) <EOL> def test_small_events_being_sent ( self ) : <EOL> self . test_params . current_sleep_interval = <NUM_LIT:1> <EOL> self . run_test_case ( '<STR_LIT:success>' , <NUM_LIT:10> * <NUM_LIT> , [ <NUM_LIT> , ONE_MB ] , [ <NUM_LIT> , ONE_MB ] , [ <NUM_LIT> , ONE_MB ] , [ <NUM_LIT:5> , ONE_MB ] ) <EOL> def test_too_many_events_being_sent ( self ) : <EOL> self . test_params . current_sleep_interval = <NUM_LIT:5> <EOL> self . run_test_case ( '<STR_LIT:success>' , <NUM_LIT:200> * <NUM_LIT> , [ <NUM_LIT> , ONE_MB ] , [ <NUM_LIT> , ONE_MB ] , [ <NUM_LIT> , ONE_MB ] , [ <NUM_LIT:1> , ONE_MB ] ) <EOL> def test_request_too_big ( self ) : <EOL> self . test_params . current_sleep_interval = <NUM_LIT:1> <EOL> self . test_params . update_params ( '<STR_LIT>' , <NUM_LIT> * <NUM_LIT> ) <EOL> self . assertAlmostEquals ( self . test_params . current_bytes_allowed_to_send , <NUM_LIT> * <NUM_LIT> ) <EOL> self . test_params . update_params ( '<STR_LIT>' , <NUM_LIT> * <NUM_LIT> ) <EOL> self . assertAlmostEquals ( self . test_params . current_bytes_allowed_to_send , <NUM_LIT:100> * <NUM_LIT> ) <EOL> def test_error_back_off ( self ) : <EOL> self . test_params . current_sleep_interval = <NUM_LIT:3> <EOL> self . run_test_case ( '<STR_LIT:error>' , <NUM_LIT:200> * <NUM_LIT> , [ <NUM_LIT> , ONE_MB ] , [ <NUM_LIT> , ONE_MB ] , [ <NUM_LIT> , ONE_MB ] , [ <NUM_LIT> , ONE_MB ] , <EOL> [ <NUM_LIT> , ONE_MB ] , [ <NUM_LIT:30> , ONE_MB ] ) <EOL> def run_test_case ( self , status , bytes_sent , * expected_sleep_interval_allowed_bytes ) : <EOL> """<STR_LIT>""" <EOL> for expected_result in expected_sleep_interval_allowed_bytes : <EOL> self . test_params . update_params ( status , bytes_sent ) <EOL> self . assertAlmostEquals ( self . test_params . current_sleep_interval , expected_result [ <NUM_LIT:0> ] ) <EOL> self . assertAlmostEquals ( self . test_params . current_bytes_allowed_to_send , expected_result [ <NUM_LIT:1> ] ) <EOL> class LogObject ( object ) : <EOL> def __init__ ( self , config ) : <EOL> self . config = config <EOL> self . log_path = config [ '<STR_LIT:path>' ] <EOL> class MonitorObject ( object ) : <EOL> def __init__ ( self , config ) : <EOL> self . module_name = config [ '<STR_LIT>' ] <EOL> self . config = config <EOL> self . log_config = { '<STR_LIT:path>' : self . module_name . split ( '<STR_LIT:.>' ) [ - <NUM_LIT:1> ] + '<STR_LIT>' } <EOL> def __create_test_configuration_instance ( self ) : <EOL> default_paths = DefaultPaths ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> return Configuration ( self . __config_file , default_paths ) <EOL> class CopyingManagerInitializationTest ( ScalyrTestCase ) : <EOL> def test_from_config_file ( self ) : <EOL> test_manager = self . __create_test_instance ( [ <EOL> { <EOL> '<STR_LIT:path>' : '<STR_LIT>' <EOL> } <EOL> ] , [ ] ) <EOL> self . assertEquals ( len ( test_manager . log_matchers ) , <NUM_LIT:2> ) <EOL> self . assertEquals ( test_manager . log_matchers [ <NUM_LIT:0> ] . config [ '<STR_LIT:path>' ] , '<STR_LIT>' ) <EOL> self . assertEquals ( test_manager . log_matchers [ <NUM_LIT:1> ] . config [ '<STR_LIT:path>' ] , '<STR_LIT>' ) <EOL> def test_from_monitors ( self ) : <EOL> test_manager = self . __create_test_instance ( [ <EOL> ] , [ <EOL> { <EOL> '<STR_LIT:path>' : '<STR_LIT>' , <EOL> } <EOL> ] ) <EOL> self . assertEquals ( len ( test_manager . log_matchers ) , <NUM_LIT:2> ) <EOL> self . assertEquals ( test_manager . log_matchers [ <NUM_LIT:0> ] . config [ '<STR_LIT:path>' ] , '<STR_LIT>' ) <EOL> self . assertEquals ( test_manager . log_matchers [ <NUM_LIT:1> ] . config [ '<STR_LIT:path>' ] , '<STR_LIT>' ) <EOL> self . assertEquals ( test_manager . log_matchers [ <NUM_LIT:1> ] . config [ '<STR_LIT>' ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_multiple_monitors_for_same_file ( self ) : <EOL> test_manager = self . __create_test_instance ( [ <EOL> ] , [ <EOL> { '<STR_LIT:path>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:path>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:path>' : '<STR_LIT>' } <EOL> ] ) <EOL> self . assertEquals ( len ( test_manager . log_matchers ) , <NUM_LIT:3> ) <EOL> self . assertEquals ( test_manager . log_matchers [ <NUM_LIT:0> ] . config [ '<STR_LIT:path>' ] , '<STR_LIT>' ) <EOL> self . assertEquals ( test_manager . log_matchers [ <NUM_LIT:1> ] . config [ '<STR_LIT:path>' ] , '<STR_LIT>' ) <EOL> self . assertEquals ( test_manager . log_matchers [ <NUM_LIT:1> ] . config [ '<STR_LIT>' ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEquals ( test_manager . log_matchers [ <NUM_LIT:2> ] . config [ '<STR_LIT:path>' ] , '<STR_LIT>' ) <EOL> self . assertEquals ( test_manager . log_matchers [ <NUM_LIT:2> ] . config [ '<STR_LIT>' ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_monitor_log_config_updated ( self ) : <EOL> test_manager = self . __create_test_instance ( [ <EOL> ] , [ <EOL> { '<STR_LIT:path>' : '<STR_LIT>' } , <EOL> ] ) <EOL> self . assertEquals ( len ( test_manager . log_matchers ) , <NUM_LIT:2> ) <EOL> self . assertEquals ( test_manager . log_matchers [ <NUM_LIT:0> ] . config [ '<STR_LIT:path>' ] , '<STR_LIT>' ) <EOL> self . assertEquals ( test_manager . log_matchers [ <NUM_LIT:1> ] . config [ '<STR_LIT:path>' ] , '<STR_LIT>' ) <EOL> self . assertEquals ( self . __monitor_fake_instances [ <NUM_LIT:0> ] . log_config [ '<STR_LIT:path>' ] , '<STR_LIT>' ) <EOL> def __create_test_instance ( self , configuration_logs_entry , monitors_log_configs ) : <EOL> config_dir = tempfile . mkdtemp ( ) <EOL> config_file = os . path . join ( config_dir , '<STR_LIT>' ) <EOL> config_fragments_dir = os . path . join ( config_dir , '<STR_LIT>' ) <EOL> os . makedirs ( config_fragments_dir ) <EOL> logs_json_array = JsonArray ( ) <EOL> for entry in configuration_logs_entry : <EOL> logs_json_array . add ( JsonObject ( content = entry ) ) <EOL> fp = open ( config_file , '<STR_LIT:w>' ) <EOL> fp . write ( json_lib . serialize ( JsonObject ( api_key = '<STR_LIT>' , logs = logs_json_array ) ) ) <EOL> fp . close ( ) <EOL> default_paths = DefaultPaths ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> config = Configuration ( config_file , default_paths ) <EOL> config . parse ( ) <EOL> self . __monitor_fake_instances = [ ] <EOL> for monitor_log_config in monitors_log_configs : <EOL> self . __monitor_fake_instances . append ( FakeMonitor ( monitor_log_config ) ) <EOL> return CopyingManager ( config , self . __monitor_fake_instances ) <EOL> class CopyingManagerEnd2EndTest ( ScalyrTestCase ) : <EOL> def setUp ( self ) : <EOL> self . _controller = None <EOL> def tearDown ( self ) : <EOL> if self . _controller is not None : <EOL> self . _controller . stop ( ) <EOL> def test_single_log_file ( self ) : <EOL> controller = self . __create_test_instance ( ) <EOL> self . __append_log_lines ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> ( request , responder_callback ) = controller . wait_for_rpc ( ) <EOL> lines = self . __extract_lines ( request ) <EOL> self . assertEquals ( <NUM_LIT:2> , len ( lines ) ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:0> ] ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:1> ] ) <EOL> responder_callback ( '<STR_LIT:success>' ) <EOL> def test_multiple_scans_of_log_file ( self ) : <EOL> controller = self . __create_test_instance ( ) <EOL> self . __append_log_lines ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> ( request , responder_callback ) = controller . wait_for_rpc ( ) <EOL> lines = self . __extract_lines ( request ) <EOL> self . assertEquals ( <NUM_LIT:2> , len ( lines ) ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:0> ] ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:1> ] ) <EOL> responder_callback ( '<STR_LIT:success>' ) <EOL> self . __append_log_lines ( '<STR_LIT>' ) <EOL> ( request , responder_callback ) = controller . wait_for_rpc ( ) <EOL> lines = self . __extract_lines ( request ) <EOL> self . assertEquals ( <NUM_LIT:1> , len ( lines ) ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:0> ] ) <EOL> def test_normal_error ( self ) : <EOL> controller = self . __create_test_instance ( ) <EOL> self . __append_log_lines ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> ( request , responder_callback ) = controller . wait_for_rpc ( ) <EOL> lines = self . __extract_lines ( request ) <EOL> self . assertEquals ( <NUM_LIT:2> , len ( lines ) ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:0> ] ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:1> ] ) <EOL> responder_callback ( '<STR_LIT:error>' ) <EOL> self . __append_log_lines ( '<STR_LIT>' ) <EOL> ( request , responder_callback ) = controller . wait_for_rpc ( ) <EOL> lines = self . __extract_lines ( request ) <EOL> self . assertEquals ( <NUM_LIT:2> , len ( lines ) ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:0> ] ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:1> ] ) <EOL> def test_drop_request_due_to_error ( self ) : <EOL> controller = self . __create_test_instance ( ) <EOL> self . __append_log_lines ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> ( request , responder_callback ) = controller . wait_for_rpc ( ) <EOL> lines = self . __extract_lines ( request ) <EOL> self . assertEquals ( <NUM_LIT:2> , len ( lines ) ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:0> ] ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:1> ] ) <EOL> responder_callback ( '<STR_LIT>' ) <EOL> self . __append_log_lines ( '<STR_LIT>' ) <EOL> ( request , responder_callback ) = controller . wait_for_rpc ( ) <EOL> lines = self . __extract_lines ( request ) <EOL> self . assertEquals ( <NUM_LIT:1> , len ( lines ) ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:0> ] ) <EOL> def test_request_too_large_error ( self ) : <EOL> controller = self . __create_test_instance ( ) <EOL> self . __append_log_lines ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> ( request , responder_callback ) = controller . wait_for_rpc ( ) <EOL> lines = self . __extract_lines ( request ) <EOL> self . assertEquals ( <NUM_LIT:2> , len ( lines ) ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:0> ] ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:1> ] ) <EOL> responder_callback ( '<STR_LIT>' ) <EOL> self . __append_log_lines ( '<STR_LIT>' ) <EOL> ( request , responder_callback ) = controller . wait_for_rpc ( ) <EOL> lines = self . __extract_lines ( request ) <EOL> self . assertEquals ( <NUM_LIT:3> , len ( lines ) ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:0> ] ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:1> ] ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:2> ] ) <EOL> def test_pipelined_requests ( self ) : <EOL> controller = self . __create_test_instance ( use_pipelining = True ) <EOL> self . __append_log_lines ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> controller . perform_scan ( ) <EOL> self . __append_log_lines ( '<STR_LIT>' ) <EOL> controller . perform_pipeline_scan ( ) <EOL> ( request , responder_callback ) = controller . wait_for_rpc ( ) <EOL> self . assertFalse ( self . __was_pipelined ( request ) ) <EOL> lines = self . __extract_lines ( request ) <EOL> self . assertEquals ( <NUM_LIT:2> , len ( lines ) ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:0> ] ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:1> ] ) <EOL> responder_callback ( '<STR_LIT:success>' ) <EOL> ( request , responder_callback ) = controller . wait_for_rpc ( ) <EOL> self . assertTrue ( self . __was_pipelined ( request ) ) <EOL> lines = self . __extract_lines ( request ) <EOL> self . assertEquals ( <NUM_LIT:1> , len ( lines ) ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:0> ] ) <EOL> responder_callback ( '<STR_LIT:success>' ) <EOL> def test_pipelined_requests_with_normal_error ( self ) : <EOL> controller = self . __create_test_instance ( use_pipelining = True ) <EOL> self . __append_log_lines ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> controller . perform_scan ( ) <EOL> self . __append_log_lines ( '<STR_LIT>' ) <EOL> controller . perform_pipeline_scan ( ) <EOL> ( request , responder_callback ) = controller . wait_for_rpc ( ) <EOL> self . assertFalse ( self . __was_pipelined ( request ) ) <EOL> lines = self . __extract_lines ( request ) <EOL> self . assertEquals ( <NUM_LIT:2> , len ( lines ) ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:0> ] ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:1> ] ) <EOL> responder_callback ( '<STR_LIT:error>' ) <EOL> ( request , responder_callback ) = controller . wait_for_rpc ( ) <EOL> self . assertFalse ( self . __was_pipelined ( request ) ) <EOL> lines = self . __extract_lines ( request ) <EOL> self . assertEquals ( <NUM_LIT:2> , len ( lines ) ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:0> ] ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:1> ] ) <EOL> responder_callback ( '<STR_LIT:success>' ) <EOL> ( request , responder_callback ) = controller . wait_for_rpc ( ) <EOL> self . assertTrue ( self . __was_pipelined ( request ) ) <EOL> lines = self . __extract_lines ( request ) <EOL> self . assertEquals ( <NUM_LIT:1> , len ( lines ) ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:0> ] ) <EOL> responder_callback ( '<STR_LIT:success>' ) <EOL> def test_pipelined_requests_with_retry_error ( self ) : <EOL> controller = self . __create_test_instance ( use_pipelining = True ) <EOL> self . __append_log_lines ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> controller . perform_scan ( ) <EOL> self . __append_log_lines ( '<STR_LIT>' ) <EOL> controller . perform_pipeline_scan ( ) <EOL> ( request , responder_callback ) = controller . wait_for_rpc ( ) <EOL> self . assertFalse ( self . __was_pipelined ( request ) ) <EOL> lines = self . __extract_lines ( request ) <EOL> self . assertEquals ( <NUM_LIT:2> , len ( lines ) ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:0> ] ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:1> ] ) <EOL> responder_callback ( '<STR_LIT>' ) <EOL> ( request , responder_callback ) = controller . wait_for_rpc ( ) <EOL> self . assertFalse ( self . __was_pipelined ( request ) ) <EOL> lines = self . __extract_lines ( request ) <EOL> self . assertEquals ( <NUM_LIT:3> , len ( lines ) ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:0> ] ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:1> ] ) <EOL> self . assertEquals ( '<STR_LIT>' , lines [ <NUM_LIT:2> ] ) <EOL> responder_callback ( '<STR_LIT:success>' ) <EOL> def __extract_lines ( self , request ) : <EOL> parsed_request = json_lib . parse ( request . get_payload ( ) ) <EOL> lines = [ ] <EOL> if '<STR_LIT>' in parsed_request : <EOL> for event in parsed_request . get_json_array ( '<STR_LIT>' ) : <EOL> if '<STR_LIT>' in event : <EOL> attrs = event . get_json_object ( '<STR_LIT>' ) <EOL> if '<STR_LIT:message>' in attrs : <EOL> lines . append ( attrs . get_string ( '<STR_LIT:message>' ) . strip ( ) ) <EOL> return lines <EOL> def __was_pipelined ( self , request ) : <EOL> return '<STR_LIT>' in request . get_timing_data ( ) <EOL> def __create_test_instance ( self , use_pipelining = False ) : <EOL> tmp_dir = tempfile . mkdtemp ( ) <EOL> config_dir = os . path . join ( tmp_dir , '<STR_LIT>' ) <EOL> data_dir = os . path . join ( tmp_dir , '<STR_LIT:data>' ) <EOL> log_dir = os . path . join ( tmp_dir , '<STR_LIT>' ) <EOL> os . mkdir ( data_dir ) <EOL> os . mkdir ( config_dir ) <EOL> os . mkdir ( log_dir ) <EOL> self . __test_log_file = os . path . join ( tmp_dir , '<STR_LIT>' ) <EOL> fp = open ( self . __test_log_file , '<STR_LIT:w>' ) <EOL> fp . close ( ) <EOL> config_file = os . path . join ( config_dir , '<STR_LIT>' ) <EOL> config_fragments_dir = os . path . join ( config_dir , '<STR_LIT>' ) <EOL> os . makedirs ( config_fragments_dir ) <EOL> logs_json_array = JsonArray ( ) <EOL> logs_json_array . add ( JsonObject ( path = self . __test_log_file ) ) <EOL> pipeline_threshold = <NUM_LIT> <EOL> if use_pipelining : <EOL> pipeline_threshold = <NUM_LIT:0.0> <EOL> fp = open ( config_file , '<STR_LIT:w>' ) <EOL> fp . write ( json_lib . serialize ( JsonObject ( api_key = '<STR_LIT>' , logs = logs_json_array , <EOL> pipeline_threshold = pipeline_threshold ) ) ) <EOL> fp . close ( ) <EOL> default_paths = DefaultPaths ( log_dir , config_file , data_dir ) <EOL> config = Configuration ( config_file , default_paths ) <EOL> config . parse ( ) <EOL> self . _controller = TestableCopyingManager ( config , [ ] ) . controller <EOL> return self . _controller <EOL> def __append_log_lines ( self , * args ) : <EOL> fp = open ( self . __test_log_file , '<STR_LIT:a>' ) <EOL> for l in args : <EOL> fp . write ( l ) <EOL> fp . write ( '<STR_LIT:\n>' ) <EOL> fp . close ( ) <EOL> class TestableCopyingManager ( CopyingManager ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , configuration , monitors ) : <EOL> CopyingManager . __init__ ( self , configuration , monitors ) <EOL> self . __test_state_cv = threading . Condition ( ) <EOL> self . __test_state = None <EOL> self . __test_state_changes = <NUM_LIT:0> <EOL> self . __test_stopping = False <EOL> self . __captured_request = None <EOL> self . __pending_response = None <EOL> self . __advance_requests_cv = threading . Condition ( ) <EOL> self . __advance_requests = <NUM_LIT:0> <EOL> self . __controller = TestableCopyingManager . TestController ( self ) <EOL> @ property <EOL> def controller ( self ) : <EOL> return self . __controller <EOL> def _sleep_but_awaken_if_stopped ( self , seconds ) : <EOL> """<STR_LIT>""" <EOL> self . __test_state_cv . acquire ( ) <EOL> self . __wait_until_advance_received ( '<STR_LIT>' ) <EOL> self . __test_state_cv . release ( ) <EOL> def _create_add_events_request ( self , session_info = None , max_size = None ) : <EOL> if session_info is None : <EOL> body = JsonObject ( server_attributes = session_info , token = '<STR_LIT>' ) <EOL> else : <EOL> body = JsonObject ( token = '<STR_LIT>' ) <EOL> return AddEventsRequest ( body , max_size = max_size ) <EOL> def _send_events ( self , add_events_task ) : <EOL> """<STR_LIT>""" <EOL> self . __test_state_cv . acquire ( ) <EOL> self . __wait_until_advance_received ( '<STR_LIT>' ) <EOL> self . __captured_request = add_events_task . add_events_request <EOL> self . __test_state_cv . release ( ) <EOL> def emit_response ( ) : <EOL> self . __test_state_cv . acquire ( ) <EOL> self . __wait_until_advance_received ( '<STR_LIT>' ) <EOL> result = self . __pending_response <EOL> self . __pending_response = None <EOL> self . __test_state_cv . release ( ) <EOL> if result is not None : <EOL> return result , <NUM_LIT:0> , '<STR_LIT>' <EOL> else : <EOL> return '<STR_LIT:success>' , <NUM_LIT:0> , '<STR_LIT>' <EOL> return emit_response <EOL> def __wait_until_advance_received ( self , new_state ) : <EOL> """<STR_LIT>""" <EOL> if self . __test_stopping : <EOL> return <EOL> self . __test_state_changes += <NUM_LIT:1> <EOL> self . __test_state = new_state <EOL> self . __test_state_cv . notifyAll ( ) <EOL> self . __test_state_cv . release ( ) <EOL> self . __advance_requests_cv . acquire ( ) <EOL> original_advance_requests = self . __advance_requests <EOL> while self . __advance_requests == original_advance_requests : <EOL> self . __advance_requests_cv . wait ( ) <EOL> self . __advance_requests_cv . release ( ) <EOL> self . __test_state_cv . acquire ( ) <EOL> self . __test_state = '<STR_LIT>' <EOL> def captured_request ( self ) : <EOL> """<STR_LIT>""" <EOL> self . __test_state_cv . acquire ( ) <EOL> try : <EOL> result = self . __captured_request <EOL> self . __captured_request = None <EOL> return result <EOL> finally : <EOL> self . __test_state_cv . release ( ) <EOL> def set_response ( self , status_message ) : <EOL> """<STR_LIT>""" <EOL> self . __test_state_cv . acquire ( ) <EOL> self . __pending_response = status_message <EOL> self . __test_state_cv . release ( ) <EOL> def advance_until ( self , final_state ) : <EOL> """<STR_LIT>""" <EOL> self . __test_state_cv . acquire ( ) <EOL> original_count = self . __test_state_changes <EOL> while self . __test_state_changes <= original_count or self . __test_state != final_state : <EOL> self . __advance_requests_cv . acquire ( ) <EOL> self . __advance_requests += <NUM_LIT:1> <EOL> self . __advance_requests_cv . notifyAll ( ) <EOL> self . __advance_requests_cv . release ( ) <EOL> self . __test_state_cv . wait ( ) <EOL> self . __test_state_cv . release ( ) <EOL> def stop_manager ( self , wait_on_join = True , join_timeout = <NUM_LIT:5> ) : <EOL> """<STR_LIT>""" <EOL> self . __test_state_cv . acquire ( ) <EOL> self . __test_stopping = True <EOL> self . __test_state_cv . release ( ) <EOL> self . __advance_requests_cv . acquire ( ) <EOL> self . __advance_requests += <NUM_LIT:1> <EOL> self . __advance_requests_cv . notifyAll ( ) <EOL> self . __advance_requests_cv . release ( ) <EOL> CopyingManager . stop_manager ( self , wait_on_join = wait_on_join , join_timeout = join_timeout ) <EOL> @ property <EOL> def test_state ( self ) : <EOL> """<STR_LIT>""" <EOL> self . __test_state_cv . acquire ( ) <EOL> try : <EOL> return self . __test_state <EOL> finally : <EOL> self . __test_state_cv . release ( ) <EOL> class TestController ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , copying_manager ) : <EOL> self . __copying_manager = copying_manager <EOL> copying_manager . start_manager ( dict ( fake_client = True ) ) <EOL> self . __copying_manager . advance_until ( '<STR_LIT>' ) <EOL> def perform_scan ( self ) : <EOL> """<STR_LIT>""" <EOL> self . __copying_manager . captured_request ( ) <EOL> self . __copying_manager . advance_until ( '<STR_LIT>' ) <EOL> def perform_pipeline_scan ( self ) : <EOL> """<STR_LIT>""" <EOL> self . __copying_manager . advance_until ( '<STR_LIT>' ) <EOL> def wait_for_rpc ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . __copying_manager . test_state != '<STR_LIT>' : <EOL> self . __copying_manager . advance_until ( '<STR_LIT>' ) <EOL> request = self . __copying_manager . captured_request ( ) <EOL> def send_response ( status_message ) : <EOL> self . __copying_manager . set_response ( status_message ) <EOL> self . __copying_manager . advance_until ( '<STR_LIT>' ) <EOL> return request , send_response <EOL> def stop ( self ) : <EOL> self . __copying_manager . stop_manager ( ) <EOL> class FakeMonitor ( object ) : <EOL> def __init__ ( self , monitor_log_config ) : <EOL> self . module_name = '<STR_LIT>' <EOL> self . log_config = monitor_log_config <EOL> def set_log_watcher ( self , log_watcher ) : <EOL> pass </s>
