Merge branch 'master' of github.com:celery/celery

celery/utils/functional.py

@@ -234,11 +234,25 @@ def _argsfromspec(spec, replace_defaults=True):
         optional = list(zip(spec.args[-split:], defaults))
     else:
         positional, optional = spec.args, []
+
+    if spec.kwonlydefaults:
+        split = len(spec.kwonlydefaults)
+        kwonlyargs = spec.kwonlyargs[:-split]
+        if replace_defaults:
+            kwonlyargs_optional = [(kw, i) for i, kw in enumerate(spec.kwonlyargs[-split:])]
+        else:
+            kwonlyargs_optional = list(spec.kwonlydefaults.items())
+    else:
+        kwonlyargs, kwonlyargs_optional = spec.kwonlyargs, []
+
     return ', '.join(filter(None, [
         ', '.join(positional),
         ', '.join('{0}={1}'.format(k, v) for k, v in optional),
         '*{0}'.format(spec.varargs) if spec.varargs else None,
         '**{0}'.format(spec.varkw) if spec.varkw else None,
+        '*' if kwonlyargs or kwonlyargs_optional else None,
+        ', '.join(kwonlyargs),
+        ', '.join('{0}="{1}"'.format(k, v) for k, v in kwonlyargs_optional),
     ]))
 
 

t/unit/utils/test_functional.py

@@ -1,6 +1,7 @@
 from __future__ import absolute_import, unicode_literals
 import pytest
 from kombu.utils.functional import lazy
+from case import skip
 from celery.five import range, nextfun
 from celery.utils.functional import (
     DummyContext,
@@ -178,6 +179,26 @@ class test_head_from_fun:
         g(1, 2)
         g(1, 2, kwarg=3)
 
+    @skip.unless_python3()
+    def test_from_fun_forced_kwargs(self):
+        local = {}
+        fun = ('def f_kwargs(*, a, b="b", c=None):'
+               '    return')
+        try:
+            exec(fun, {}, local)
+        except SyntaxError:
+            # Python 2.
+            return
+        f_kwargs = local['f_kwargs']
+
+        g = head_from_fun(f_kwargs)
+        with pytest.raises(TypeError):
+            g(1)
+
+        g(a=1)
+        g(a=1, b=2)
+        g(a=1, b=2, c=3)
+
 
 class test_fun_takes_argument: