first_attempt_at_spark_test.py 文件源码

def test_transform_to_recordstore(self):
        # simply verify that the transform method is called first, then
        # rdd to recordstore
        kafka_stream = MagicMock(name='kafka_stream')
        transformed_stream = MagicMock(name='transformed_stream')
        kafka_stream.transform.return_value = transformed_stream
        MonMetricsKafkaProcessor.transform_to_recordstore(
            kafka_stream)

        # TODO(someone) figure out why these asserts now fail
        # transformed_stream_expected = call.foreachRDD(
        #     MonMetricsKafkaProcessor.rdd_to_recordstore
        # ).call_list()
        # kafka_stream_expected = call.transform(
        #     MonMetricsKafkaProcessor.store_offset_ranges
        # ).call_list()
        # self.assertEqual(kafka_stream_expected, kafka_stream.mock_calls)
        # self.assertEqual(transformed_stream_expected,
        #                  transformed_stream.mock_calls)