from rhodecode.lib.celerylib.uniqueue_tasks import stable_payload_key class TestUniqueTask: def test_stable_payload_key_generation(self): task_name = "test_task" task_args = (1, 2, 3) task_kwargs = {"arg1": 10} expected_output = "celery:unique:test_task:463d9df63be487de5fd2073cfc26a6813713ee96c0cf9f7b5f949e12b8643511" for _ in range(10): res = stable_payload_key(task_name, task_args, task_kwargs) assert res == expected_output