import uuid

from django.test import TestCase, override_settings

from pillowtop.es_utils import initialize_index_and_mapping

from corehq.apps.app_manager.models import Application
from corehq.apps.app_manager.tests.app_factory import AppFactory
from corehq.apps.change_feed import topics
from corehq.apps.change_feed.consumer.feed import (
    change_meta_from_kafka_message,
)
from corehq.apps.change_feed.producer import producer
from corehq.apps.change_feed.tests.utils import get_test_kafka_consumer
from corehq.apps.change_feed.topics import get_multi_topic_offset
from corehq.apps.receiverwrapper.util import submit_form_locally
from corehq.apps.userreports.tests.utils import doc_to_change
from corehq.elastic import get_es_new
from corehq.form_processor.tests.utils import FormProcessorTestUtils
from corehq.form_processor.utils import TestFormMetadata, get_simple_form_xml
from corehq.pillows.mappings.user_mapping import USER_INDEX, USER_INDEX_INFO
from corehq.pillows.xform import get_xform_pillow
from corehq.util.elastic import ensure_index_deleted


class FormPillowTest(TestCase):
    domain = 'test-form-pillow-domain'

    def setUp(self):
        super(FormPillowTest, self).setUp()
        FormProcessorTestUtils.delete_all_xforms()
        self.pillow = get_xform_pillow(skip_ucr=True)
        factory = AppFactory(domain=self.domain)
        self.app = factory.app
        self.app.save()

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        cls.es = get_es_new()
        initialize_index_and_mapping(cls.es, USER_INDEX_INFO)

    def tearDown(self):
        self.app.delete()
        super(FormPillowTest, self).tearDown()

    @classmethod
    def tearDownClass(cls):
        ensure_index_deleted(USER_INDEX)
        super().tearDownClass()

    def test_xform_pillow_couch(self):
        form = self._make_form()
        kafka_seq = self._get_kafka_seq()
        producer.send_change(topics.FORM, doc_to_change(form.to_json()).metadata)
        self.assertFalse(self.app.has_submissions)

        self.pillow.process_changes(since=kafka_seq, forever=False)
        self.assertTrue(Application.get(self.app._id).has_submissions)

    @override_settings(TESTS_SHOULD_USE_SQL_BACKEND=True)
    def test_form_pillow_sql(self):
        consumer = get_test_kafka_consumer(topics.FORM, topics.FORM_SQL)
        kafka_seq = self._get_kafka_seq()

        form = self._make_form()

        # confirm change made it to kafka
        message = next(consumer)
        change_meta = change_meta_from_kafka_message(message.value)
        self.assertEqual(form.form_id, change_meta.document_id)
        self.assertEqual(self.domain, change_meta.domain)
        self.assertFalse(self.app.has_submissions)

        self.pillow.process_changes(since=kafka_seq, forever=False)
        self.assertTrue(Application.get(self.app._id).has_submissions)

    @override_settings(TESTS_SHOULD_USE_SQL_BACKEND=True)
    def test_form_pillow_non_existant_build_id(self):
        consumer = get_test_kafka_consumer(topics.FORM, topics.FORM_SQL)
        kafka_seq = self._get_kafka_seq()

        form = self._make_form(build_id='not-here')

        # confirm change made it to kafka
        message = next(consumer)
        change_meta = change_meta_from_kafka_message(message.value)
        self.assertEqual(form.form_id, change_meta.document_id)
        self.assertEqual(self.domain, change_meta.domain)
        self.assertFalse(self.app.has_submissions)

        self.pillow.process_changes(since=kafka_seq, forever=False)
        self.assertFalse(Application.get(self.app._id).has_submissions)

    @override_settings(TESTS_SHOULD_USE_SQL_BACKEND=True)
    def test_form_pillow_mismatch_domains(self):
        consumer = get_test_kafka_consumer(topics.FORM, topics.FORM_SQL)
        kafka_seq = self._get_kafka_seq()
        self.app.domain = 'not-this-domain'
        self.app.save()

        form = self._make_form()

        # confirm change made it to kafka
        message = next(consumer)
        change_meta = change_meta_from_kafka_message(message.value)
        self.assertEqual(form.form_id, change_meta.document_id)
        self.assertEqual(self.domain, change_meta.domain)
        self.assertFalse(self.app.has_submissions)

        self.pillow.process_changes(since=kafka_seq, forever=False)
        self.assertFalse(Application.get(self.app._id).has_submissions)

    @override_settings(TESTS_SHOULD_USE_SQL_BACKEND=True)
    def test_two_forms_with_same_app(self):
        """Ensures two forms submitted to the same app does not error"""
        kafka_seq = self._get_kafka_seq()

        self._make_form()

        # confirm change made it to kafka
        self.assertFalse(self.app.has_submissions)

        self.pillow.process_changes(since=kafka_seq, forever=False)
        newly_saved_app = Application.get(self.app._id)
        self.assertTrue(newly_saved_app.has_submissions)
        # Ensure that the app has been saved
        self.assertNotEqual(self.app._rev, newly_saved_app._rev)

        self._make_form()
        self.pillow.process_changes(since=kafka_seq, forever=False)
        self.assertTrue(Application.get(self.app._id).has_submissions)
        # Ensure that the app has not been saved twice
        self.assertEqual(Application.get(self.app._id)._rev, newly_saved_app._rev)

    def _make_form(self, build_id=None):
        metadata = TestFormMetadata(domain=self.domain)
        form_xml = get_simple_form_xml(uuid.uuid4().hex, metadata=metadata)
        result = submit_form_locally(
            form_xml,
            self.domain,
            build_id=build_id or self.app._id
        )
        return result.xform

    def _get_kafka_seq(self):
        # KafkaChangeFeed listens for multiple topics (form, form-sql) in the form pillow,
        # so we need to provide a dict of seqs to kafka
        return get_multi_topic_offset([topics.FORM, topics.FORM_SQL])

@@ -377,62 +377,8 @@
 ,%0A)%0A
-from corehq.apps.change_feed.producer import producer%0A
 from
@@ -580,70 +580,8 @@
 lly%0A
-from corehq.apps.userreports.tests.utils import doc_to_change%0A
 from
@@ -1675,396 +1675,8 @@
 ()%0A%0A
-    def test_xform_pillow_couch(self):%0A        form = self._make_form()%0A        kafka_seq = self._get_kafka_seq()%0A        producer.send_change(topics.FORM, doc_to_change(form.to_json()).metadata)%0A        self.assertFalse(self.app.has_submissions)%0A%0A        self.pillow.process_changes(since=kafka_seq, forever=False)%0A        self.assertTrue(Application.get(self.app._id).has_submissions)%0A%0A
     

Remove obsolete pillowtop Couch test