test_json_kafka_offsets.py 文件源码

def test_as_dict(self):
        topic_1 = str(uuid.uuid4())
        partition_1 = random.randint(0, 1024)
        until_offset_1 = random.randint(0, sys.maxsize)
        from_offset_1 = random.randint(0, sys.maxsize)
        app_name_1 = str(uuid.uuid4())

        offset_spec = OffsetSpec(
            app_name=app_name_1, topic=topic_1, partition=partition_1,
            from_offset=from_offset_1, until_offset=until_offset_1)
        offset_spec_dict = JSONOffsetSpecs.as_dict(offset_spec)
        self.assertions_on_offset(
            used_value={
                "topic": topic_1, "partition": partition_1,
                "app_name": app_name_1, "from_offset": from_offset_1,
                "until_offset": until_offset_1},
            offset_value=offset_spec_dict)