producer_test.py 文件源码

def test_meteorite_on_off(
        self,
        create_message,
        registered_schema,
        producer,
        enable_meteorite,
        expected_call_count
    ):
        with mock.patch.object(
            data_pipeline.tools.meteorite_wrappers.StatsCounter,
            'process',
            autospec=True
        ) as mock_stats_counter:
            producer.enable_meteorite = enable_meteorite
            m = create_message(registered_schema, timeslot=1.0)
            producer.publish(m)
            assert mock_stats_counter.call_count == expected_call_count