mirror of
https://github.com/d3vyce/taskiq-deduplication.git
synced 2026-08-04 19:14:07 +00:00
d3vyceandGitHub
b4ee3fce45
* feat: add taskiq deduplication * doc: rework class comment + update README * fix: make build_deduplication_key private * fix: raise_on_duplicate is now False by default * chore: remove pre_execute * chore: add documentation
44 lines
863 B
Python
44 lines
863 B
Python
import pytest
|
|
import fakeredis.aioredis
|
|
from taskiq import TaskiqMessage, TaskiqResult
|
|
|
|
|
|
@pytest.fixture
|
|
def anyio_backend():
|
|
return "asyncio"
|
|
|
|
|
|
@pytest.fixture
|
|
async def fake_redis():
|
|
client = fakeredis.aioredis.FakeRedis()
|
|
yield client
|
|
await client.aclose()
|
|
|
|
|
|
@pytest.fixture
|
|
def make_message():
|
|
def _make(task_name="my_task", task_id="task-1", labels=None, kwargs=None):
|
|
return TaskiqMessage(
|
|
task_id=task_id,
|
|
task_name=task_name,
|
|
labels=labels or {},
|
|
labels_types={},
|
|
args=[],
|
|
kwargs=kwargs or {},
|
|
)
|
|
|
|
return _make
|
|
|
|
|
|
@pytest.fixture
|
|
def make_result():
|
|
def _make(is_err=False):
|
|
return TaskiqResult(
|
|
is_err=is_err,
|
|
log="",
|
|
return_value=None,
|
|
execution_time=0.0,
|
|
)
|
|
|
|
return _make
|