import uuid
from django.test import TestCase
from casexml.apps.case.mock import CaseBlock
from casexml.apps.case.signals import case_post_save
from casexml.apps.case.util import post_case_blocks
from corehq.apps.es import CaseES
from corehq.form_processor.tests.utils import FormProcessorTestUtils
from corehq.pillows.case import CasePillow
from corehq.util.context_managers import drop_connected_signals
from corehq.util.elastic import delete_es_index


class CasePillowTest(TestCase):

    domain = 'case-pillowtest-domain'

    def setUp(self):
        FormProcessorTestUtils.delete_all_cases()
        self.pillow = CasePillow()
        self.elasticsearch = self.pillow.get_es_new()
        delete_es_index(self.pillow.es_index)

    def tearDown(self):
        delete_es_index(self.pillow.es_index)

    def test_case_pillow_couch(self):
        # make a case
        case_id = uuid.uuid4().hex
        case_name = 'case-name-{}'.format(uuid.uuid4().hex)
        case = self._make_a_case(case_id, case_name)

        # send to elasticsearch
        self.pillow.process_changes(since=0, forever=False)
        self.elasticsearch.indices.refresh(self.pillow.es_index)

        # verify there
        results = CaseES().run()
        self.assertEqual(1, results.total)
        case_doc = results.hits[0]
        self.assertEqual(self.domain, case_doc['domain'])
        self.assertEqual(case_id, case_doc['_id'])
        self.assertEqual(case_name, case_doc['name'])

        # cleanup
        case.delete()

    def _make_a_case(self, case_id, case_name):
        # this avoids having to deal with all the reminders code bootstrap
        with drop_connected_signals(case_post_save):
            form, cases = post_case_blocks(
                [
                    CaseBlock(
                        create=True,
                        case_id=case_id,
                        case_name=case_name,
                    ).as_xml()
                ], domain=self.domain
            )
        self.assertEqual(1, len(cases))
        return cases[0]

@@ -37,17 +37,115 @@
 TestCase
+, override_settings%0Afrom kafka import KafkaConsumer%0Afrom kafka.common import KafkaUnavailableError
 %0A
-
 from cas
@@ -286,16 +286,140 @@
 _blocks%0A
+from corehq.apps.change_feed import topics%0Afrom corehq.apps.change_feed.consumer.feed import change_meta_from_kafka_message%0A
 from cor
@@ -668,16 +668,134 @@
 s_index%0A
+from corehq.util.test_utils import trap_extra_setup%0Afrom testapps.test_pillowtop.utils import get_test_kafka_consumer%0A
 %0A%0Aclass 
@@ -1821,16 +1821,16 @@
 cleanup%0A
-
         
@@ -1844,16 +1844,1306 @@
 lete()%0A%0A
+    @override_settings(TESTS_SHOULD_USE_SQL_BACKEND=True)%0A    def test_case_pillow_sql(self):%0A        consumer = get_test_kafka_consumer(topics.CASE_SQL)%0A        # have to get the seq id before the change is processed%0A        kafka_seq = consumer.offsets()%5B'fetch'%5D%5B(topics.CASE_SQL, 0)%5D%0A%0A        # make a case%0A        case_id = uuid.uuid4().hex%0A        case_name = 'case-name-%7B%7D'.format(uuid.uuid4().hex)%0A        case = self._make_a_case(case_id, case_name)%0A%0A        # confirm change made it to kafka%0A        message = consumer.next()%0A        change_meta = change_meta_from_kafka_message(message.value)%0A        self.assertEqual(case.case_id, change_meta.document_id)%0A        self.assertEqual(self.domain, change_meta.domain)%0A%0A        # todo: send to elasticsearch%0A        # sql_pillow = get_sql_xform_to_elasticsearch_pillow()%0A        # sql_pillow.process_changes(since=kafka_seq, forever=False)%0A        # self.elasticsearch.indices.refresh(self.pillow.es_index)%0A%0A        # confirm change made it to elasticserach%0A        results = CaseES().run()%0A        self.assertEqual(1, results.total)%0A        case_doc = results.hits%5B0%5D%0A        self.assertEqual(self.domain, case_doc%5B'domain'%5D)%0A        self.assertEqual(case_id, case_doc%5B'_id'%5D)%0A        self.assertEqual(case_name, case_doc%5B'name'%5D)%0A%0A
     def 

add failing test outline for sql cases to elastic