<s> from corehq . apps . change_feed import topics <EOL> from corehq . apps . change_feed . consumer . feed import change_meta_from_kafka_message <EOL> from corehq . apps . es . sms import SMSES <EOL> from corehq . apps . sms . models import MessagingEvent , MessagingSubEvent , SMS <EOL> from corehq . elastic import get_es_new <EOL> from corehq . pillows . mappings . sms_mapping import SMS_INDEX_INFO <EOL> from corehq . pillows . sms import get_sql_sms_pillow <EOL> from corehq . util . elastic import ensure_index_deleted <EOL> from datetime import datetime <EOL> from dimagi . utils . parsing import json_format_datetime <EOL> from django . test import TestCase <EOL> from mock import patch <EOL> from testapps . test_pillowtop . utils import get_test_kafka_consumer <EOL> @ patch ( '<STR_LIT>' ) <EOL> class SqlSMSPillowTest ( TestCase ) : <EOL> dependent_apps = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> domain = '<STR_LIT>' <EOL> def setUp ( self ) : <EOL> self . elasticsearch = get_es_new ( ) <EOL> ensure_index_deleted ( SMS_INDEX_INFO . index ) <EOL> def tearDown ( self ) : <EOL> ensure_index_deleted ( SMS_INDEX_INFO . index ) <EOL> SMS . objects . filter ( domain = self . domain ) . delete ( ) <EOL> MessagingSubEvent . objects . filter ( parent__domain = self . domain ) . delete ( ) <EOL> MessagingEvent . objects . filter ( domain = self . domain ) . delete ( ) <EOL> def _create_sms ( self ) : <EOL> event = MessagingEvent . objects . create ( <EOL> domain = self . domain , <EOL> date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:12> , <NUM_LIT:0> ) , <EOL> source = MessagingEvent . SOURCE_OTHER , <EOL> source_id = None , <EOL> content_type = MessagingEvent . CONTENT_SMS , <EOL> form_unique_id = None , <EOL> form_name = None , <EOL> status = MessagingEvent . STATUS_COMPLETED , <EOL> error_code = None , <EOL> additional_error_text = None , <EOL> recipient_type = None , <EOL> recipient_id = None <EOL> ) <EOL> subevent = MessagingSubEvent . objects . create ( <EOL> parent = event , <EOL> date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:12> , <NUM_LIT:0> ) , <EOL> recipient_type = MessagingEvent . RECIPIENT_CASE , <EOL> recipient_id = None , <EOL> content_type = MessagingEvent . CONTENT_SMS , <EOL> form_unique_id = None , <EOL> form_name = None , <EOL> xforms_session = None , <EOL> case_id = None , <EOL> status = MessagingEvent . STATUS_COMPLETED , <EOL> error_code = None , <EOL> additional_error_text = None <EOL> ) <EOL> self . sms_dict = dict ( <EOL> domain = self . domain , <EOL> date = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:12> , <NUM_LIT:0> ) , <EOL> couch_recipient_doc_type = '<STR_LIT>' , <EOL> couch_recipient = '<STR_LIT>' , <EOL> phone_number = '<STR_LIT>' , <EOL> direction = '<STR_LIT:O>' , <EOL> error = False , <EOL> system_error_message = '<STR_LIT>' , <EOL> system_phone_number = '<STR_LIT>' , <EOL> backend_api = '<STR_LIT>' , <EOL> backend_id = '<STR_LIT>' , <EOL> billed = False , <EOL> workflow = '<STR_LIT:default>' , <EOL> xforms_session_couch_id = '<STR_LIT>' , <EOL> reminder_id = '<STR_LIT>' , <EOL> location_id = '<STR_LIT>' , <EOL> messaging_subevent_id = subevent . pk , <EOL> text = '<STR_LIT>' , <EOL> raw_text = '<STR_LIT>' , <EOL> datetime_to_process = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:11> , <NUM_LIT> ) , <EOL> processed = True , <EOL> num_processing_attempts = <NUM_LIT:1> , <EOL> queued_timestamp = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:11> , <NUM_LIT> ) , <EOL> processed_timestamp = datetime ( <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:12> , <NUM_LIT:0> <NUM_LIT:1> ) , <EOL> domain_scope = self . domain , <EOL> ignore_opt_out = False , <EOL> backend_message_id = '<STR_LIT>' , <EOL> chat_user_id = '<STR_LIT>' , <EOL> invalid_survey_response = False , <EOL> fri_message_bank_lookup_completed = True , <EOL> fri_message_bank_message_id = '<STR_LIT>' , <EOL> fri_id = '<STR_LIT>' , <EOL> fri_risk_profile = '<STR_LIT:X>' , <EOL> ) <EOL> self . sms = SMS . objects . create ( <EOL> ** self . sms_dict <EOL> ) <EOL> def _to_json ( self , sms_dict , sms ) : <EOL> result = { '<STR_LIT>' : sms . couch_id , '<STR_LIT:id>' : sms . pk } <EOL> for k , v in sms_dict . iteritems ( ) : <EOL> value = json_format_datetime ( v ) if isinstance ( v , datetime ) else v <EOL> result [ k ] = value <EOL> return result <EOL> def test_sql_sms_pillow ( self , mock_do_publish ) : <EOL> mock_do_publish . return_value = True <EOL> consumer = get_test_kafka_consumer ( topics . SMS ) <EOL> kafka_seq = consumer . offsets ( ) [ '<STR_LIT>' ] [ ( topics . SMS , <NUM_LIT:0> ) ] <EOL> self . _create_sms ( ) <EOL> sms_json = self . _to_json ( self . sms_dict , self . sms ) <EOL> self . assertEqual ( self . sms . to_json ( ) , sms_json ) <EOL> self . sms . publish_change ( ) <EOL> message = consumer . next ( ) <EOL> change_meta = change_meta_from_kafka_message ( message . value ) <EOL> self . assertEqual ( self . sms . couch_id , change_meta . document_id ) <EOL> self . assertEqual ( self . domain , change_meta . domain ) <EOL> sms_pillow = get_sql_sms_pillow ( '<STR_LIT>' ) <EOL> sms_pillow . process_changes ( since = kafka_seq , forever = False ) <EOL> self . elasticsearch . indices . refresh ( SMS_INDEX_INFO . index ) <EOL> results = SMSES ( ) . run ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , results . total ) <EOL> sms_doc = results . hits [ <NUM_LIT:0> ] <EOL> self . assertEqual ( sms_doc , sms_json ) </s>
