<s> """<STR_LIT>""" <EOL> from core import jobs_registry <EOL> from core . domain import feedback_jobs_continuous <EOL> from core . domain import feedback_services <EOL> from core . platform import models <EOL> from core . tests import test_utils <EOL> ( feedback_models , ) = models . Registry . import_models ( [ models . NAMES . feedback ] ) <EOL> class ModifiedFeedbackAnalyticsAggregator ( <EOL> feedback_jobs_continuous . FeedbackAnalyticsAggregator ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> def _get_batch_job_manager_class ( cls ) : <EOL> return ModifiedFeedbackAnalyticsMRJobManager <EOL> @ classmethod <EOL> def _kickoff_batch_job_after_previous_one_ends ( cls ) : <EOL> pass <EOL> class ModifiedFeedbackAnalyticsMRJobManager ( <EOL> feedback_jobs_continuous . FeedbackAnalyticsMRJobManager ) : <EOL> @ classmethod <EOL> def _get_continuous_computation_class ( cls ) : <EOL> return ModifiedFeedbackAnalyticsAggregator <EOL> class FeedbackAnalyticsAggregatorUnitTests ( test_utils . GenericTestBase ) : <EOL> """<STR_LIT>""" <EOL> ALL_CC_MANAGERS_FOR_TESTS = [ ModifiedFeedbackAnalyticsAggregator ] <EOL> def _get_swap_context ( self ) : <EOL> return self . swap ( <EOL> jobs_registry , '<STR_LIT>' , <EOL> self . ALL_CC_MANAGERS_FOR_TESTS ) <EOL> def _run_job_and_check_results ( self , exp_id , <EOL> expected_thread_analytics_dict ) : <EOL> self . process_and_flush_pending_tasks ( ) <EOL> ModifiedFeedbackAnalyticsAggregator . start_computation ( ) <EOL> self . assertEqual ( self . count_jobs_in_taskqueue ( ) , <NUM_LIT:1> ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> self . assertEqual ( <EOL> ModifiedFeedbackAnalyticsAggregator . get_thread_analytics ( <EOL> exp_id ) . to_dict ( ) , <EOL> expected_thread_analytics_dict ) <EOL> def test_no_threads ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> exp_id = '<STR_LIT>' <EOL> self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> self . _run_job_and_check_results ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> } ) <EOL> def test_single_thread_single_exp ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> exp_id = '<STR_LIT>' <EOL> thread_id = '<STR_LIT>' <EOL> self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> thread = feedback_models . FeedbackThreadModel . create ( <EOL> exp_id , thread_id ) <EOL> thread . exploration_id = exp_id <EOL> thread . put ( ) <EOL> self . _run_job_and_check_results ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> def test_multiple_threads_single_exp ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> exp_id = '<STR_LIT>' <EOL> thread_id_1 = '<STR_LIT>' <EOL> thread_id_2 = '<STR_LIT>' <EOL> self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> thread_1 = feedback_models . FeedbackThreadModel . create ( <EOL> exp_id , thread_id_1 ) <EOL> thread_1 . exploration_id = exp_id <EOL> thread_1 . put ( ) <EOL> thread_2 = feedback_models . FeedbackThreadModel . create ( <EOL> exp_id , thread_id_2 ) <EOL> thread_2 . exploration_id = exp_id <EOL> thread_2 . put ( ) <EOL> self . _run_job_and_check_results ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> } ) <EOL> def test_multiple_threads_multiple_exp ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> exp_id_1 = '<STR_LIT>' <EOL> exp_id_2 = '<STR_LIT>' <EOL> thread_id_1 = '<STR_LIT>' <EOL> thread_id_2 = '<STR_LIT>' <EOL> thread_id_3 = '<STR_LIT>' <EOL> thread_id_4 = '<STR_LIT>' <EOL> self . save_new_valid_exploration ( exp_id_1 , '<STR_LIT>' ) <EOL> self . save_new_valid_exploration ( exp_id_2 , '<STR_LIT>' ) <EOL> thread_1 = feedback_models . FeedbackThreadModel . create ( <EOL> exp_id_1 , thread_id_1 ) <EOL> thread_1 . exploration_id = exp_id_1 <EOL> thread_1 . put ( ) <EOL> thread_2 = feedback_models . FeedbackThreadModel . create ( <EOL> exp_id_1 , thread_id_2 ) <EOL> thread_2 . exploration_id = exp_id_1 <EOL> thread_2 . put ( ) <EOL> thread_3 = feedback_models . FeedbackThreadModel . create ( <EOL> exp_id_2 , thread_id_3 ) <EOL> thread_3 . exploration_id = exp_id_2 <EOL> thread_3 . put ( ) <EOL> thread_4 = feedback_models . FeedbackThreadModel . create ( <EOL> exp_id_2 , thread_id_4 ) <EOL> thread_4 . exploration_id = exp_id_2 <EOL> thread_4 . put ( ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> ModifiedFeedbackAnalyticsAggregator . start_computation ( ) <EOL> self . assertEqual ( self . count_jobs_in_taskqueue ( ) , <NUM_LIT:1> ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> self . assertEqual ( <EOL> ModifiedFeedbackAnalyticsAggregator . get_thread_analytics ( <EOL> exp_id_1 ) . to_dict ( ) , <EOL> { <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> } ) <EOL> self . assertEqual ( <EOL> ModifiedFeedbackAnalyticsAggregator . get_thread_analytics ( <EOL> exp_id_2 ) . to_dict ( ) , <EOL> { <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> } ) <EOL> def test_thread_closed_job_running ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> user_id = '<STR_LIT>' <EOL> exp_id = '<STR_LIT>' <EOL> thread_id_1 = '<STR_LIT>' <EOL> self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> thread_1 = feedback_models . FeedbackThreadModel . create ( <EOL> exp_id , thread_id_1 ) <EOL> thread_1 . exploration_id = exp_id <EOL> thread_1 . put ( ) <EOL> self . _run_job_and_check_results ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> ModifiedFeedbackAnalyticsAggregator . stop_computation ( user_id ) <EOL> self . assertEqual ( self . count_jobs_in_taskqueue ( ) , <NUM_LIT:0> ) <EOL> thread = ( feedback_models . FeedbackThreadModel . <EOL> get_by_exp_and_thread_id ( exp_id , thread_id_1 ) ) <EOL> thread . status = feedback_models . STATUS_CHOICES_FIXED <EOL> thread . put ( ) <EOL> self . _run_job_and_check_results ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> def test_thread_closed_reopened_again ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> user_id = '<STR_LIT>' <EOL> exp_id = '<STR_LIT>' <EOL> thread_id_1 = '<STR_LIT>' <EOL> self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> thread_1 = feedback_models . FeedbackThreadModel . create ( <EOL> exp_id , thread_id_1 ) <EOL> thread_1 . exploration_id = exp_id <EOL> thread_1 . put ( ) <EOL> self . _run_job_and_check_results ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> ModifiedFeedbackAnalyticsAggregator . stop_computation ( user_id ) <EOL> self . assertEqual ( self . count_jobs_in_taskqueue ( ) , <NUM_LIT:0> ) <EOL> thread = ( feedback_models . FeedbackThreadModel . <EOL> get_by_exp_and_thread_id ( exp_id , thread_id_1 ) ) <EOL> thread . status = feedback_models . STATUS_CHOICES_FIXED <EOL> thread . put ( ) <EOL> self . _run_job_and_check_results ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> ModifiedFeedbackAnalyticsAggregator . stop_computation ( user_id ) <EOL> self . assertEqual ( self . count_jobs_in_taskqueue ( ) , <NUM_LIT:0> ) <EOL> thread = ( feedback_models . FeedbackThreadModel . <EOL> get_by_exp_and_thread_id ( exp_id , thread_id_1 ) ) <EOL> thread . status = feedback_models . STATUS_CHOICES_OPEN <EOL> thread . put ( ) <EOL> self . _run_job_and_check_results ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> def test_thread_closed_status_changed ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> user_id = '<STR_LIT>' <EOL> exp_id = '<STR_LIT>' <EOL> thread_id_1 = '<STR_LIT>' <EOL> self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> thread_1 = feedback_models . FeedbackThreadModel . create ( <EOL> exp_id , thread_id_1 ) <EOL> thread_1 . exploration_id = exp_id <EOL> thread_1 . put ( ) <EOL> self . _run_job_and_check_results ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> ModifiedFeedbackAnalyticsAggregator . stop_computation ( user_id ) <EOL> self . assertEqual ( self . count_jobs_in_taskqueue ( ) , <NUM_LIT:0> ) <EOL> thread = ( feedback_models . FeedbackThreadModel . <EOL> get_by_exp_and_thread_id ( exp_id , thread_id_1 ) ) <EOL> thread . status = feedback_models . STATUS_CHOICES_FIXED <EOL> thread . put ( ) <EOL> self . _run_job_and_check_results ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> ModifiedFeedbackAnalyticsAggregator . stop_computation ( user_id ) <EOL> self . assertEqual ( self . count_jobs_in_taskqueue ( ) , <NUM_LIT:0> ) <EOL> thread = ( feedback_models . FeedbackThreadModel . <EOL> get_by_exp_and_thread_id ( exp_id , thread_id_1 ) ) <EOL> thread . status = feedback_models . STATUS_CHOICES_IGNORED <EOL> thread . put ( ) <EOL> self . _run_job_and_check_results ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> class RealtimeFeedbackAnalyticsUnitTests ( test_utils . GenericTestBase ) : <EOL> """<STR_LIT>""" <EOL> ALL_CC_MANAGERS_FOR_TESTS = [ ModifiedFeedbackAnalyticsAggregator ] <EOL> def _get_swap_context ( self ) : <EOL> return self . swap ( <EOL> jobs_registry , '<STR_LIT>' , <EOL> self . ALL_CC_MANAGERS_FOR_TESTS ) <EOL> def _flush_tasks_and_check_analytics ( <EOL> self , exp_id , expected_thread_analytics_dict ) : <EOL> self . process_and_flush_pending_tasks ( ) <EOL> self . assertEqual ( <EOL> ModifiedFeedbackAnalyticsAggregator . get_thread_analytics ( <EOL> exp_id ) . to_dict ( ) , expected_thread_analytics_dict ) <EOL> def test_no_threads ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> exp_id = '<STR_LIT>' <EOL> self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> self . _flush_tasks_and_check_analytics ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> } ) <EOL> def test_single_thread_single_exp ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> exp_id = '<STR_LIT>' <EOL> self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> feedback_services . create_thread ( <EOL> exp_id , '<STR_LIT>' , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . _flush_tasks_and_check_analytics ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> def test_multiple_threads_single_exp ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> exp_id = '<STR_LIT>' <EOL> self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> feedback_services . create_thread ( <EOL> exp_id , '<STR_LIT>' , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> feedback_services . create_thread ( <EOL> exp_id , '<STR_LIT>' , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . _flush_tasks_and_check_analytics ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> } ) <EOL> def test_multiple_threads_multiple_exp ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> exp_id_1 = '<STR_LIT>' <EOL> exp_id_2 = '<STR_LIT>' <EOL> self . save_new_valid_exploration ( exp_id_1 , '<STR_LIT>' ) <EOL> self . save_new_valid_exploration ( exp_id_2 , '<STR_LIT>' ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> feedback_services . create_thread ( <EOL> exp_id_1 , '<STR_LIT>' , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> feedback_services . create_thread ( <EOL> exp_id_1 , '<STR_LIT>' , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> feedback_services . create_thread ( <EOL> exp_id_2 , '<STR_LIT>' , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> feedback_services . create_thread ( <EOL> exp_id_2 , '<STR_LIT>' , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . _flush_tasks_and_check_analytics ( exp_id_1 , { <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> } ) <EOL> self . _flush_tasks_and_check_analytics ( exp_id_2 , { <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> } ) <EOL> def test_thread_closed_job_running ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> exp_id = '<STR_LIT>' <EOL> self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> feedback_services . create_thread ( <EOL> exp_id , '<STR_LIT>' , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . _flush_tasks_and_check_analytics ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> threadlist = feedback_services . get_all_threads ( exp_id , False ) <EOL> thread_id = threadlist [ <NUM_LIT:0> ] . get_thread_id ( ) <EOL> feedback_services . create_message ( <EOL> exp_id , thread_id , '<STR_LIT>' , <EOL> feedback_models . STATUS_CHOICES_FIXED , None , '<STR_LIT>' ) <EOL> self . _flush_tasks_and_check_analytics ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> def test_thread_closed_reopened_again ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> exp_id = '<STR_LIT>' <EOL> self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> feedback_services . create_thread ( <EOL> exp_id , '<STR_LIT>' , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . _flush_tasks_and_check_analytics ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> threadlist = feedback_services . get_all_threads ( exp_id , False ) <EOL> thread_id = threadlist [ <NUM_LIT:0> ] . get_thread_id ( ) <EOL> feedback_services . create_message ( <EOL> exp_id , thread_id , '<STR_LIT>' , <EOL> feedback_models . STATUS_CHOICES_FIXED , None , '<STR_LIT>' ) <EOL> self . _flush_tasks_and_check_analytics ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> threadlist = feedback_services . get_all_threads ( exp_id , False ) <EOL> thread_id = threadlist [ <NUM_LIT:0> ] . get_thread_id ( ) <EOL> feedback_services . create_message ( <EOL> exp_id , thread_id , '<STR_LIT>' , <EOL> feedback_models . STATUS_CHOICES_OPEN , None , '<STR_LIT>' ) <EOL> self . _flush_tasks_and_check_analytics ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> def test_thread_closed_status_changed ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> exp_id = '<STR_LIT>' <EOL> self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> feedback_services . create_thread ( <EOL> exp_id , '<STR_LIT>' , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . _flush_tasks_and_check_analytics ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> threadlist = feedback_services . get_all_threads ( exp_id , False ) <EOL> thread_id = threadlist [ <NUM_LIT:0> ] . get_thread_id ( ) <EOL> feedback_services . create_message ( <EOL> exp_id , thread_id , '<STR_LIT>' , <EOL> feedback_models . STATUS_CHOICES_FIXED , None , '<STR_LIT>' ) <EOL> self . _flush_tasks_and_check_analytics ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> threadlist = feedback_services . get_all_threads ( exp_id , False ) <EOL> thread_id = threadlist [ <NUM_LIT:0> ] . get_thread_id ( ) <EOL> feedback_services . create_message ( <EOL> exp_id , thread_id , '<STR_LIT>' , <EOL> feedback_models . STATUS_CHOICES_IGNORED , None , '<STR_LIT>' ) <EOL> self . _flush_tasks_and_check_analytics ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> def test_realtime_with_batch_computation ( self ) : <EOL> with self . _get_swap_context ( ) : <EOL> user_id = '<STR_LIT>' <EOL> exp_id = '<STR_LIT>' <EOL> self . save_new_valid_exploration ( exp_id , '<STR_LIT>' ) <EOL> feedback_services . create_thread ( <EOL> exp_id , '<STR_LIT>' , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> ModifiedFeedbackAnalyticsAggregator . start_computation ( ) <EOL> self . assertEqual ( self . count_jobs_in_taskqueue ( ) , <NUM_LIT:1> ) <EOL> self . process_and_flush_pending_tasks ( ) <EOL> ModifiedFeedbackAnalyticsAggregator . stop_computation ( user_id ) <EOL> self . assertEqual ( self . count_jobs_in_taskqueue ( ) , <NUM_LIT:0> ) <EOL> self . _flush_tasks_and_check_analytics ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> } ) <EOL> feedback_services . create_thread ( <EOL> exp_id , '<STR_LIT>' , None , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . _flush_tasks_and_check_analytics ( exp_id , { <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> } ) </s>
