producer_test.py 文件源码

def test_log_error_on_exception(self, message):
        with mock.patch.object(
            data_pipeline._clog_writer.clog,
            'log_line',
            side_effect=RandomException()
        ) as mock_log_line, mock.patch.object(
            data_pipeline._clog_writer,
            'logger'
        ) as mock_logger:
            writer = ClogWriter()
            writer.publish(message)

        call_args = "Failed to scribe message - {}".format(str(message))

        assert mock_log_line.called
        assert mock_logger.error.call_args_list[0] == mock.call(call_args)