<commit_before>from __future__ import absolute_import
from __future__ import unicode_literals
import uuid
from django.test import TestCase
from corehq.apps.change_feed import topics
from corehq.apps.change_feed.consumer.feed import KafkaChangeFeed
from corehq.apps.commtrack.helpers import make_product
from corehq.apps.commtrack.tests.util import get_single_balance_block
from corehq.apps.hqcase.utils import submit_case_blocks
from corehq.apps.receiverwrapper.util import submit_form_locally
from corehq.form_processor.interfaces.dbaccessors import FormAccessors, CaseAccessors
from corehq.form_processor.tests.utils import FormProcessorTestUtils, use_sql_backend
from corehq.form_processor.utils import get_simple_form_xml, should_use_sql_backend
from corehq.util.test_utils import create_and_save_a_case, create_and_save_a_form
from pillowtop.pillow.interface import ConstructedPillow
from pillowtop.processors.sample import TestProcessor
from testapps.test_pillowtop.utils import process_pillow_changes


@use_sql_backend
class KafkaPublishingSQLTest(TestCase):

    domain = 'sql-kafka-publishing-test'

    def setUp(self):
        super(KafkaPublishingSQLTest, self).setUp()
        FormProcessorTestUtils.delete_all_cases_forms_ledgers()
        self.form_accessors = FormAccessors(domain=self.domain)
        self.processor = TestProcessor()
        self.case_pillow = ConstructedPillow(
            name='test-kafka-case-feed',
            checkpoint=None,
            change_feed=KafkaChangeFeed(topics=[topics.CASE, topics.CASE_SQL], client_id='test-kafka-case-feed'),
            processor=self.processor
        )
        self.ledger_pillow = ConstructedPillow(
            name='test-kafka-ledger-feed',
            checkpoint=None,
            change_feed=KafkaChangeFeed(topics=[topics.LEDGER], client_id='test-kafka-ledger-feed'),
            processor=self.processor
        )

    def test_duplicate_case_published(self):
        # this test only runs on sql because it's handling a sql-specific edge case where duplicate
        # form submissions should cause cases to be resubmitted.
        # see: http://manage.dimagi.com/default.asp?228463 for context
        case_id = uuid.uuid4().hex
        form_xml = get_simple_form_xml(uuid.uuid4().hex, case_id)
        submit_form_locally(form_xml, domain=self.domain)
        self.assertEqual(1, len(CaseAccessors(self.domain).get_case_ids_in_domain()))

        with process_pillow_changes(self.case_pillow):
            with process_pillow_changes('DefaultChangeFeedPillow'):
                dupe_form = submit_form_locally(form_xml, domain=self.domain).xform
                self.assertTrue(dupe_form.is_duplicate)

        # check the case was republished
        self.assertEqual(1, len(self.processor.changes_seen))
        case_meta = self.processor.changes_seen[0].metadata
        self.assertEqual(case_id, case_meta.document_id)
        self.assertEqual(self.domain, case_meta.domain)

    def test_duplicate_ledger_published(self):
        # this test also only runs on the sql backend for reasons described in test_duplicate_case_published
        # setup products and case
        product_a = make_product(self.domain, 'A Product', 'prodcode_a')
        product_b = make_product(self.domain, 'B Product', 'prodcode_b')
        case_id = uuid.uuid4().hex
        form_xml = get_simple_form_xml(uuid.uuid4().hex, case_id)
        submit_form_locally(form_xml, domain=self.domain)

        # submit ledger data
        balances = (
            (product_a._id, 100),
            (product_b._id, 50),<commit_msg>remove test<commit_after>from __future__ import absolute_import
from __future__ import unicode_literals
import uuid
from django.test import TestCase
from corehq.apps.change_feed import topics
from corehq.apps.change_feed.consumer.feed import KafkaChangeFeed
from corehq.apps.commtrack.helpers import make_product
from corehq.apps.commtrack.tests.util import get_single_balance_block
from corehq.apps.hqcase.utils import submit_case_blocks
from corehq.apps.receiverwrapper.util import submit_form_locally
from corehq.form_processor.interfaces.dbaccessors import FormAccessors, CaseAccessors
from corehq.form_processor.tests.utils import FormProcessorTestUtils, use_sql_backend
from corehq.form_processor.utils import get_simple_form_xml, should_use_sql_backend
from corehq.util.test_utils import create_and_save_a_case, create_and_save_a_form
from pillowtop.pillow.interface import ConstructedPillow
from pillowtop.processors.sample import TestProcessor
from testapps.test_pillowtop.utils import process_pillow_changes


@use_sql_backend
class KafkaPublishingSQLTest(TestCase):

    domain = 'sql-kafka-publishing-test'

    def setUp(self):
        super(KafkaPublishingSQLTest, self).setUp()
        FormProcessorTestUtils.delete_all_cases_forms_ledgers()
        self.form_accessors = FormAccessors(domain=self.domain)
        self.processor = TestProcessor()
        self.case_pillow = ConstructedPillow(
            name='test-kafka-case-feed',
            checkpoint=None,
            change_feed=KafkaChangeFeed(topics=[topics.CASE, topics.CASE_SQL], client_id='test-kafka-case-feed'),
            processor=self.processor
        )
        self.ledger_pillow = ConstructedPillow(
            name='test-kafka-ledger-feed',
            checkpoint=None,
            change_feed=KafkaChangeFeed(topics=[topics.LEDGER], client_id='test-kafka-ledger-feed'),
            processor=self.processor
        )

    def test_duplicate_ledger_published(self):
        # this test also only runs on the sql backend for reasons described in test_duplicate_case_published
        # setup products and case
        product_a = make_product(self.domain, 'A Product', 'prodcode_a')
        product_b = make_product(self.domain, 'B Product', 'prodcode_b')
        case_id = uuid.uuid4().hex
        form_xml = get_simple_form_xml(uuid.uuid4().hex, case_id)
        submit_form_locally(form_xml, domain=self.domain)

        # submit ledger data
        balances = (
            (product_a._id, 100),
            (product_b._id, 50),
{'fraction_non_alphanumeric': 0.07855973813420622, 'fraction_numerical': 0.004091653027823241, 'mean_word_length': 3.8270142180094786, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 12}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5924545', 'n_tokens_mistral': 1869, 'n_tokens_neox': 1836, 'n_words': 347}