<commit_before>import uuid
from django.test import SimpleTestCase, TestCase
from kafka import KeyedProducer
from kafka.common import KafkaUnavailableError, OffsetOutOfRangeError
from corehq.apps.change_feed import topics
from corehq.apps.change_feed.connection import get_kafka_client_or_none
from corehq.apps.change_feed.consumer.feed import KafkaChangeFeed, MultiTopicCheckpointEventHandler
from corehq.apps.change_feed.exceptions import UnavailableKafkaOffset
from corehq.apps.change_feed.producer import send_to_kafka
from corehq.apps.change_feed.topics import get_multi_topic_first_available_offsets
from corehq.util.test_utils import trap_extra_setup
from dimagi.utils.decorators.memoized import memoized
from pillowtop.checkpoints.manager import PillowCheckpoint
from pillowtop.feed.interface import ChangeMeta
from pillowtop.pillow.interface import ConstructedPillow
from pillowtop.processors.sample import CountingProcessor


class KafkaChangeFeedTest(SimpleTestCase):

    @trap_extra_setup(KafkaUnavailableError)
    def test_multiple_topics(self):
        feed = KafkaChangeFeed(topics=[topics.FORM, topics.CASE], group_id='test-kafka-feed')
        self.assertEqual(0, len(list(feed.iter_changes(since=None, forever=False))))
        offsets = feed.get_current_offsets()
        expected_metas = [publish_stub_change(topics.FORM), publish_stub_change(topics.CASE)]
        unexpected_metas = [publish_stub_change(topics.FORM_SQL), publish_stub_change(topics.CASE_SQL)]
        changes = list(feed.iter_changes(since=offsets, forever=False))
        self.assertEqual(2, len(changes))<commit_msg>[ci skip] unused import<commit_after>import uuid
from django.test import SimpleTestCase, TestCase
from kafka import KeyedProducer
from kafka.common import KafkaUnavailableError
from corehq.apps.change_feed import topics
from corehq.apps.change_feed.connection import get_kafka_client_or_none
from corehq.apps.change_feed.consumer.feed import KafkaChangeFeed, MultiTopicCheckpointEventHandler
from corehq.apps.change_feed.exceptions import UnavailableKafkaOffset
from corehq.apps.change_feed.producer import send_to_kafka
from corehq.apps.change_feed.topics import get_multi_topic_first_available_offsets
from corehq.util.test_utils import trap_extra_setup
from dimagi.utils.decorators.memoized import memoized
from pillowtop.checkpoints.manager import PillowCheckpoint
from pillowtop.feed.interface import ChangeMeta
from pillowtop.pillow.interface import ConstructedPillow
from pillowtop.processors.sample import CountingProcessor


class KafkaChangeFeedTest(SimpleTestCase):

    @trap_extra_setup(KafkaUnavailableError)
    def test_multiple_topics(self):
        feed = KafkaChangeFeed(topics=[topics.FORM, topics.CASE], group_id='test-kafka-feed')
        self.assertEqual(0, len(list(feed.iter_changes(since=None, forever=False))))
        offsets = feed.get_current_offsets()
        expected_metas = [publish_stub_change(topics.FORM), publish_stub_change(topics.CASE)]
        unexpected_metas = [publish_stub_change(topics.FORM_SQL), publish_stub_change(topics.CASE_SQL)]
        changes = list(feed.iter_changes(since=offsets, forever=False))
        self.assertEqual(2, len(changes))
{'fraction_non_alphanumeric': 0.07278975320212433, 'fraction_numerical': 0.0012496094970321774, 'mean_word_length': 6.678657074340528, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, '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': '4666059', 'n_tokens_mistral': 959, 'n_tokens_neox': 962, 'n_words': 181}