worker_tests.py 文件源码

def test_rabbitmq_worker_callback_method_with_invalid_serialization_method(self, getLogger, import_string):
        assert isinstance(import_string, mock.MagicMock)
        # Arrange
        queue_name = "queue_name"
        rabbitmq_configs = {}
        serialization_method = "invalid"
        result_backend = {
            "result_backend_class": "easy_job.result_backends.dummy.DummyBackend",
        }
        logger = getLogger.return_value = mock.MagicMock()
        # Act
        rbt = RabbitMQWorker(queue_name=queue_name,
                             rabbitmq_configs=rabbitmq_configs,
                             serialization_method=serialization_method,
                             result_backend=result_backend,
                             logger="log")
        rbt.callback(mock.MagicMock(), mock.MagicMock(), None, "")

        # Assert

        logger.log.assert_called_once_with(logging.ERROR, mock.ANY)