<s> """<STR_LIT>""" <EOL> import datetime <EOL> import time <EOL> from core import jobs_registry <EOL> from core . domain import event_services <EOL> from core . domain import exp_services <EOL> from core . domain import stats_jobs_continuous <EOL> from core . platform import models <EOL> from core . tests import test_utils <EOL> import feconf <EOL> ( exp_models , stats_models ) = models . Registry . import_models ( [ <EOL> models . NAMES . exploration , models . NAMES . statistics ] ) <EOL> EMPTY_STATE_HIT_COUNTS_DICT = { <EOL> feconf . DEFAULT_INIT_STATE_NAME : { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> } <EOL> } <EOL> class ModifiedStatisticsAggregator ( stats_jobs_continuous . StatisticsAggregator ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> def _get_batch_job_manager_class ( cls ) : <EOL> return ModifiedStatisticsMRJobManager <EOL> @ classmethod <EOL> def _kickoff_batch_job_after_previous_one_ends ( cls ) : <EOL> pass <EOL> class ModifiedStatisticsMRJobManager ( <EOL> stats_jobs_continuous . StatisticsMRJobManager ) : <EOL> @ classmethod <EOL> def _get_continuous_computation_class ( cls ) : <EOL> return ModifiedStatisticsAggregator <EOL> class StatsAggregatorUnitTests ( test_utils . GenericTestBase ) : <EOL> """<STR_LIT>""" <EOL> ALL_CC_MANAGERS_FOR_TESTS = [ ModifiedStatisticsAggregator ] <EOL> def _get_swap_context ( self ) : <EOL> return self . swap ( <EOL> jobs_registry , '<STR_LIT>' , <EOL> self . ALL_CC_MANAGERS_FOR_TESTS ) <EOL> def _record_start ( self , exp_id , exp_version , state , session_id ) : <EOL> event_services . StartExplorationEventHandler . record ( <EOL> exp_id , exp_version , state , session_id , { } , <EOL> feconf . PLAY_TYPE_NORMAL ) <EOL> def _record_leave ( self , exp_id , exp_version , state , session_id ) : <EOL> event_services . MaybeLeaveExplorationEventHandler . record ( <EOL> exp_id , exp_version , state , session_id , <NUM_LIT> , { } , <EOL> feconf . PLAY_TYPE_NORMAL ) <EOL> def _record_complete ( self , exp_id , exp_version , state , session_id ) : <EOL> event_services . CompleteExplorationEventHandler . record ( <EOL> exp_id , exp_version , state , session_id , <NUM_LIT> , { } , <EOL> feconf . PLAY_TYPE_NORMAL ) <EOL> def _record_state_hit ( self , exp_id , exp_version , state , session_id ) : <EOL> event_services . StateHitEventHandler . record ( <EOL> exp_id , exp_version , state , session_id , { } , <EOL> feconf . PLAY_TYPE_NORMAL ) <EOL> def _create_state_counter ( self , exp_id , state , first_entry_count ) : <EOL> counter = stats_models . StateCounterModel . get_or_create ( exp_id , state ) <EOL> counter . first_entry_count = first_entry_count <EOL> counter . put ( ) <EOL> def test_state_hit ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> exp_id = '<STR_LIT>' <EOL> exploration = self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> stats_jobs_continuous . _STATE_COUNTER_CUTOFF_DATE = ( <EOL> datetime . datetime . utcnow ( ) ) <EOL> original_init_state = exploration . init_state_name <EOL> new_init_state_name = '<STR_LIT>' <EOL> exploration . rename_state ( original_init_state , new_init_state_name ) <EOL> exp_services . _save_exploration ( '<STR_LIT>' , exploration , '<STR_LIT>' , [ ] ) <EOL> exp_version = <NUM_LIT:2> <EOL> state2_name = '<STR_LIT>' <EOL> self . _record_state_hit ( <EOL> exp_id , exp_version , new_init_state_name , '<STR_LIT>' ) <EOL> self . _record_state_hit ( <EOL> exp_id , exp_version , new_init_state_name , '<STR_LIT>' ) <EOL> self . _create_state_counter ( exp_id , original_init_state , <NUM_LIT> ) <EOL> self . _record_state_hit ( <EOL> exp_id , exp_version , state2_name , '<STR_LIT>' ) <EOL> self . _create_state_counter ( exp_id , state2_name , <NUM_LIT:9> ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> ModifiedStatisticsAggregator . start_computation ( ) <EOL> self . assertEqual ( self . count_jobs_in_taskqueue ( ) , <NUM_LIT:1> ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> state_hit_counts = ( <EOL> stats_jobs_continuous . StatisticsAggregator . get_statistics ( <EOL> exp_id , exp_version ) [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <EOL> state_hit_counts [ new_init_state_name ] [ '<STR_LIT>' ] , <NUM_LIT:2> ) <EOL> self . assertEqual ( <EOL> state_hit_counts [ state2_name ] [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> output_model = ( <EOL> stats_jobs_continuous . StatisticsAggregator . get_statistics ( <EOL> exp_id , stats_jobs_continuous . VERSION_NONE ) ) <EOL> state_hit_counts = output_model [ '<STR_LIT>' ] <EOL> self . assertEqual ( <EOL> state_hit_counts [ original_init_state ] [ '<STR_LIT>' ] , <NUM_LIT> ) <EOL> self . assertEqual ( <EOL> state_hit_counts [ state2_name ] [ '<STR_LIT>' ] , <NUM_LIT:9> ) <EOL> self . assertEqual ( output_model [ '<STR_LIT>' ] , <NUM_LIT> ) <EOL> state_hit_counts = ( <EOL> stats_jobs_continuous . StatisticsAggregator . get_statistics ( <EOL> exp_id , <EOL> stats_jobs_continuous . VERSION_ALL ) [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <EOL> state_hit_counts [ original_init_state ] [ '<STR_LIT>' ] , <NUM_LIT> ) <EOL> self . assertEqual ( <EOL> state_hit_counts [ new_init_state_name ] [ '<STR_LIT>' ] , <NUM_LIT:2> ) <EOL> self . assertEqual ( <EOL> state_hit_counts [ state2_name ] [ '<STR_LIT>' ] , <NUM_LIT:10> ) <EOL> def test_no_completion ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> exp_id = '<STR_LIT>' <EOL> exp_version = <NUM_LIT:1> <EOL> exploration = self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> state = exploration . init_state_name <EOL> self . _record_start ( exp_id , exp_version , state , '<STR_LIT>' ) <EOL> self . _record_start ( exp_id , exp_version , state , '<STR_LIT>' ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> ModifiedStatisticsAggregator . start_computation ( ) <EOL> self . assertEqual ( self . count_jobs_in_taskqueue ( ) , <NUM_LIT:1> ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> model_id = '<STR_LIT>' % ( exp_id , exp_version ) <EOL> output_model = stats_models . ExplorationAnnotationsModel . get ( <EOL> model_id ) <EOL> self . assertEqual ( output_model . num_starts , <NUM_LIT:2> ) <EOL> self . assertEqual ( output_model . num_completions , <NUM_LIT:0> ) <EOL> def test_all_complete ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> exp_id = '<STR_LIT>' <EOL> exp_version = <NUM_LIT:1> <EOL> exploration = self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> state = exploration . init_state_name <EOL> self . _record_start ( exp_id , exp_version , state , '<STR_LIT>' ) <EOL> self . _record_complete ( <EOL> exp_id , exp_version , stats_jobs_continuous . OLD_END_DEST , <EOL> '<STR_LIT>' ) <EOL> self . _record_start ( exp_id , exp_version , state , '<STR_LIT>' ) <EOL> self . _record_complete ( <EOL> exp_id , exp_version , stats_jobs_continuous . OLD_END_DEST , <EOL> '<STR_LIT>' ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> ModifiedStatisticsAggregator . start_computation ( ) <EOL> self . assertEqual ( self . count_jobs_in_taskqueue ( ) , <NUM_LIT:1> ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> model_id = '<STR_LIT>' % ( exp_id , exp_version ) <EOL> output_model = stats_models . ExplorationAnnotationsModel . get ( <EOL> model_id ) <EOL> self . assertEqual ( output_model . num_starts , <NUM_LIT:2> ) <EOL> self . assertEqual ( output_model . num_completions , <NUM_LIT:2> ) <EOL> def test_one_leave_and_one_complete ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> exp_id = '<STR_LIT>' <EOL> exp_version = <NUM_LIT:1> <EOL> exploration = self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> state = exploration . init_state_name <EOL> self . _record_start ( exp_id , exp_version , state , '<STR_LIT>' ) <EOL> self . _record_leave ( exp_id , exp_version , state , '<STR_LIT>' ) <EOL> self . _record_start ( exp_id , exp_version , state , '<STR_LIT>' ) <EOL> self . _record_complete ( <EOL> exp_id , exp_version , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> ModifiedStatisticsAggregator . start_computation ( ) <EOL> self . assertEqual ( self . count_jobs_in_taskqueue ( ) , <NUM_LIT:1> ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> model_id = '<STR_LIT>' % ( exp_id , exp_version ) <EOL> output_model = stats_models . ExplorationAnnotationsModel . get ( <EOL> model_id ) <EOL> self . assertEqual ( output_model . num_starts , <NUM_LIT:2> ) <EOL> self . assertEqual ( output_model . num_completions , <NUM_LIT:1> ) <EOL> def test_one_leave_and_one_complete_same_session ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> exp_id = '<STR_LIT>' <EOL> exp_version = <NUM_LIT:1> <EOL> exploration = self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> init_state = exploration . init_state_name <EOL> self . _record_start ( exp_id , exp_version , init_state , '<STR_LIT>' ) <EOL> self . _record_state_hit ( exp_id , exp_version , init_state , '<STR_LIT>' ) <EOL> self . _record_leave ( exp_id , exp_version , init_state , '<STR_LIT>' ) <EOL> self . _record_state_hit ( exp_id , exp_version , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . _record_complete ( exp_id , exp_version , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> ModifiedStatisticsAggregator . start_computation ( ) <EOL> self . assertEqual ( self . count_jobs_in_taskqueue ( ) , <NUM_LIT:1> ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> model_id = '<STR_LIT>' % ( exp_id , exp_version ) <EOL> output_model = stats_models . ExplorationAnnotationsModel . get ( <EOL> model_id ) <EOL> self . assertEqual ( output_model . num_starts , <NUM_LIT:1> ) <EOL> self . assertEqual ( output_model . num_completions , <NUM_LIT:1> ) <EOL> stats_dict = ModifiedStatisticsAggregator . get_statistics ( exp_id , <NUM_LIT:1> ) <EOL> self . assertEqual ( stats_dict [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> self . assertEqual ( stats_dict [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> self . assertEqual ( stats_dict [ '<STR_LIT>' ] , { <EOL> exploration . init_state_name : { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } <EOL> } ) <EOL> def test_multiple_maybe_leaves_same_session ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> exp_id = '<STR_LIT>' <EOL> exp_version = <NUM_LIT:1> <EOL> exploration = self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> state = exploration . init_state_name <EOL> self . _record_start ( exp_id , exp_version , state , '<STR_LIT>' ) <EOL> self . _record_leave ( exp_id , exp_version , state , '<STR_LIT>' ) <EOL> self . _record_leave ( exp_id , exp_version , state , '<STR_LIT>' ) <EOL> self . _record_complete ( <EOL> exp_id , exp_version , stats_jobs_continuous . OLD_END_DEST , <EOL> '<STR_LIT>' ) <EOL> self . _record_start ( exp_id , exp_version , state , '<STR_LIT>' ) <EOL> self . _record_leave ( exp_id , exp_version , state , '<STR_LIT>' ) <EOL> self . _record_leave ( exp_id , exp_version , state , '<STR_LIT>' ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> ModifiedStatisticsAggregator . start_computation ( ) <EOL> self . assertEqual ( self . count_jobs_in_taskqueue ( ) , <NUM_LIT:1> ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> model_id = '<STR_LIT>' % ( exp_id , exp_version ) <EOL> output_model = stats_models . ExplorationAnnotationsModel . get ( <EOL> model_id ) <EOL> self . assertEqual ( output_model . num_starts , <NUM_LIT:2> ) <EOL> self . assertEqual ( output_model . num_completions , <NUM_LIT:1> ) <EOL> def test_multiple_explorations ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> exp_version = <NUM_LIT:1> <EOL> exp_id_1 = '<STR_LIT>' <EOL> exploration = self . save_new_valid_exploration ( exp_id_1 , '<STR_LIT>' ) <EOL> state_1_1 = exploration . init_state_name <EOL> exp_id_2 = '<STR_LIT>' <EOL> exploration = self . save_new_valid_exploration ( exp_id_2 , '<STR_LIT>' ) <EOL> state_2_1 = exploration . init_state_name <EOL> self . _record_start ( exp_id_1 , exp_version , state_1_1 , '<STR_LIT>' ) <EOL> self . _record_start ( exp_id_1 , exp_version , state_1_1 , '<STR_LIT>' ) <EOL> self . _record_start ( exp_id_2 , exp_version , state_2_1 , '<STR_LIT>' ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> ModifiedStatisticsAggregator . start_computation ( ) <EOL> self . assertEqual ( self . count_jobs_in_taskqueue ( ) , <NUM_LIT:1> ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> results = ModifiedStatisticsAggregator . get_statistics ( <EOL> exp_id_1 , stats_jobs_continuous . VERSION_ALL ) <EOL> self . assertDictContainsSubset ( { <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : EMPTY_STATE_HIT_COUNTS_DICT , <EOL> } , results ) <EOL> results = ModifiedStatisticsAggregator . get_statistics ( <EOL> exp_id_2 , stats_jobs_continuous . VERSION_ALL ) <EOL> self . assertDictContainsSubset ( { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : EMPTY_STATE_HIT_COUNTS_DICT , <EOL> } , results ) <EOL> self . _record_start ( exp_id_1 , exp_version , state_1_1 , '<STR_LIT>' ) <EOL> self . _record_start ( exp_id_2 , exp_version , state_2_1 , '<STR_LIT>' ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> results = ModifiedStatisticsAggregator . get_statistics ( <EOL> exp_id_1 , stats_jobs_continuous . VERSION_ALL ) <EOL> self . assertDictContainsSubset ( { <EOL> '<STR_LIT>' : <NUM_LIT:3> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : EMPTY_STATE_HIT_COUNTS_DICT , <EOL> } , results ) <EOL> results = ModifiedStatisticsAggregator . get_statistics ( <EOL> exp_id_2 , stats_jobs_continuous . VERSION_ALL ) <EOL> self . assertDictContainsSubset ( { <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : EMPTY_STATE_HIT_COUNTS_DICT , <EOL> } , results ) <EOL> views_for_all_exps = ModifiedStatisticsAggregator . get_views_multi ( [ <EOL> exp_id_1 , exp_id_2 ] ) <EOL> self . assertEqual ( views_for_all_exps , [ <NUM_LIT:3> , <NUM_LIT:2> ] ) </s>
