test_delay.py 文件源码

def test_memory_delay_task_decorator_invalid_function(event_loop):

    import logging

    logger = logging.getLogger("aiotasks")

    class CustomLogger(logging.StreamHandler):

        def __init__(self):
            super(CustomLogger, self).__init__()
            self.content = []

        def emit(self, record):
            self.content.append(record.msg)

    custom = CustomLogger()
    logger.addHandler(custom)

    manager = build_manager(dsn="memory://", loop=event_loop)

    async def run():

        # Send an invalid task name
        task_id = uuid.uuid4().hex

        await manager._task_queue.put((manager.task_list_name,
                                      msgpack.packb(dict(task_id=task_id,
                                                         function="non_exist",
                                                         args=[],
                                                         kwargs={}),
                                                    use_bin_type=True)))

        manager.run()

        await manager.wait(timeout=0.2, exit_on_finish=False, wait_timeout=0.1)

    event_loop.run_until_complete(run())
    manager.stop()

    assert "No local task with name 'non_exist'" in custom.content