AsyncTestCase

Async test classes.

AsyncTestCase

Bases: BaseTestCase

Асинхронный кейс тестирования.

Пример

from qtasks import QueueTasks
from qtasks.tests import AsyncTestCase

app = QueueTasks()

test_case = AsyncTestCase(app=app)

__init__(app, name=None)

Асинхронный тестовый кейс.

Parameters:

Name	Type	Description	Default
app	(QueueTasks, Doc)	Основной экземпляр.	required
name	str	Имя проекта. Это имя может быть использовано для тестовых компонентов. По умолчанию: None.	None

add_task(task_name, priority=0, args=None, kwargs=None, timeout=None) async

Добавить задачу.

Parameters:

Name	Type	Description	Default
task_name	str	Имя задачи.	required
priority	int	Приоритет задачи. По умолчанию: 0.	0
args	tuple	args задачи. По умолчанию: ().	None
kwargs	dict	kwargs задачи. По умолчанию: {}	None
timeout	float	Таймаут задачи. Если указан, задача вызывается через qtasks.results.AsyncResult.	None

Returns:

Type	Description
Task | None	Task|None: Данные задачи или None.

get(uuid) async

Получить задачу.

Parameters:

Name	Type	Description	Default
uuid	UUID | str	UUID Задачи.	required

Returns:

Type	Description
Task | None	Task|None: Данные задачи или None.

start(loop=None, starter=None, num_workers=4, reset_config=True)

Запускает app.run_forever().

Parameters:

Name	Type	Description	Default
loop	AbstractEventLoop	асинхронный loop. По умолчанию: None.	None
starter	BaseStarter	Стартер. По умолчанию: qtasks.starters.AsyncStarter.	None
num_workers	int	Количество запущенных воркеров. По умолчанию: 4.	4
reset_config	bool	Обновить config у воркера и брокера. По умолчанию: True.	True

start_in_background(starter=None, num_workers=4, reset_config=True)

Запустить app.run_forever() в фоновом режиме.

Parameters:

Name	Type	Description	Default
starter	BaseStarter	Стартер. По умолчанию: qtasks.starters.AsyncStarter.	None
num_workers	int	Количество запущенных воркеров. По умолчанию: 4.	4
reset_config	bool	Обновить config у воркера и брокера. По умолчанию: True.	True

stop() async

Останавливает кейс тестирования.