| 852 | assert task.__annotations__['args'] == 'Sequence[str]' |
| 853 | |
| 854 | def test_annotate_decorator(self): |
| 855 | from celery.app.task import Task |
| 856 | |
| 857 | class adX(Task): |
| 858 | |
| 859 | def run(self, y, z, x): |
| 860 | return y, z, x |
| 861 | |
| 862 | check = Mock() |
| 863 | |
| 864 | def deco(fun): |
| 865 | |
| 866 | def _inner(*args, **kwargs): |
| 867 | check(*args, **kwargs) |
| 868 | return fun(*args, **kwargs) |
| 869 | return _inner |
| 870 | |
| 871 | self.app.conf.task_annotations = { |
| 872 | adX.name: {'@__call__': deco} |
| 873 | } |
| 874 | adX.bind(self.app) |
| 875 | assert adX.app is self.app |
| 876 | |
| 877 | i = adX() |
| 878 | i(2, 4, x=3) |
| 879 | check.assert_called_with(i, 2, 4, x=3) |
| 880 | |
| 881 | i.annotate() |
| 882 | i.annotate() |
| 883 | |
| 884 | def test_apply_async_adds_children(self): |
| 885 | from celery._state import _task_stack |