Repository: quora/pyanalyze (language: python)

Issue:
Support Callable types
Currently pyanalyze basically doesn't support `typing.Callable`; the arguments and return type are simply ignored. A proper implementation of Callable type would share a lot of characteristics with the `ExtendedArgSpec` type that pyanalyze uses for type checking function calls, so #51 could also help make this issue easier.


Discussion / hints:

Fix patch:
diff --git a/pyanalyze/signature.py b/pyanalyze/signature.py
index d4649638..7e84c6cd 100644
--- a/pyanalyze/signature.py
+++ b/pyanalyze/signature.py
@@ -15,6 +15,7 @@
 )
 from .value import (
     AnnotatedValue,
+    CanAssignContext,
     HasAttrExtension,
     HasAttrGuardExtension,
     KnownValue,
@@ -27,6 +28,7 @@
     TypeVarMap,
     extract_typevars,
     stringify_object,
+    unify_typevar_maps,
 )
 
 import ast
@@ -146,6 +148,7 @@ class Signature:
     signature: inspect.Signature
     implementation: Optional[ImplementationFn] = None
     callable: Optional[object] = None
+    is_asynq: bool = False
     has_return_annotation: bool = True
     logger: Optional[Logger] = field(repr=False, default=None, compare=False)
     typevars_of_params: Dict[str, List["TypeVar"]] = field(
@@ -367,6 +370,189 @@ def check_call(
         else:
             return self._apply_annotated_constraints(return_value, bound_args)
 
+    def can_assign(
+        self, other: "Signature", ctx: CanAssignContext
+    ) -> Union[str, TypeVarMap]:
+        """Equivalent of Value.can_assign.
+
+        If the other signature is incompatible with this signature, return a string
+        explaining the discrepancy.
+
+        """
+        if self.is_asynq and not other.is_asynq:
+            return "callable is not asynq"
+        their_return = other.signature.return_annotation
+        my_return = self.signature.return_annotation
+        return_tv_map = my_return.can_assign(their_return, ctx)
+        if return_tv_map is None:
+            return (
+                f"return annotation {their_return} is not compatible with {my_return}"
+            )
+        tv_maps = [return_tv_map]
+        their_params = list(other.signature.parameters.values())
+        their_args = other.get_param_of_kind(SigParameter.VAR_POSITIONAL)
+        their_kwargs = other.get_param_of_kind(SigParameter.VAR_KEYWORD)
+        consumed_positional = set()
+        consumed_keyword = set()
+        for i, my_param in enumerate(self.signature.parameters.values()):
+            if my_param.kind is SigParameter.POSITIONAL_ONLY:
+                if i < len(their_params) and their_params[i].kind in (
+                    SigParameter.POSITIONAL_ONLY,
+                    SigParameter.POSITIONAL_OR_KEYWORD,
+                ):
+                    tv_map = their_params[i].annotation.can_assign(
+                        my_param.annotation, ctx
+                    )
+                    if tv_map is None:
+                        return (
+                            f"type of positional-only parameter {my_param.name!r} is incompatible: "
+                            f"{their_params[i].annotation} is incompatible with {my_param.annotation}"
+                        )
+                    tv_maps.append(tv_map)
+                    consumed_positional.add(their_params[i].name)
+                elif their_args is not None:
+                    tv_map = their_args.annotation.can_assign(my_param.annotation, ctx)
+                    if tv_map is None:
+                        return (
+                            f"type of positional-only parameter {my_param.name!r} is incompatible: "
+                            f"*args type {their_args.annotation} is incompatible with {my_param.annotation}"
+                        )
+                    tv_maps.append(tv_map)
+                else:
+                    return f"positional-only parameter {i} is not accepted"
+            elif my_param.kind is SigParameter.POSITIONAL_OR_KEYWORD:
+                if (
+                    i < len(their_params)
+                    and their_params[i].kind is SigParameter.POSITIONAL_OR_KEYWORD
+                ):
+                    if my_param.name != their_params[i].name:
+                        return f"param name {their_params[i].name!r} does not match {my_param.name!r}"
+                    tv_map = their_params[i].annotation.can_assign(
+                        my_param.annotation, ctx
+                    )
+                    if tv_map is None:
+                        return (
+                            f"type of parameter {my_param.name!r} is incompatible: "
+                            f"{their_params[i].annotation} is incompatible with {my_param.annotation}"
+                        )
+                    tv_maps.append(tv_map)
+                    consumed_positional.add(their_params[i])
+                    consumed_keyword.add(their_params[i])
+                elif (
+                    i < len(their_params)
+                    and their_params[i].kind is SigParameter.POSITIONAL_ONLY
+                ):
+                    return f"parameter {my_param.name!r} is not accepted as a keyword argument"
+                elif their_args is not None and their_kwargs is not None:
+                    tv_map = their_args.annotation.can_assign(my_param.annotation, ctx)
+                    if tv_map is None:
+                        return (
+                            f"type of parameter {my_param.name!r} is incompatible: "
+                            f"*args type {their_args.annotation} is incompatible with {my_param.annotation}"
+                        )
+                    tv_maps.append(tv_map)
+                    tv_map = their_kwargs.annotation.can_assign(
+                        my_param.annotation, ctx
+                    )
+                    if tv_map is None:
+                        return (
+                            f"type of parameter {my_param.name!r} is incompatible: "
+                            f"**kwargs type {their_kwargs.annotation} is incompatible with {my_param.annotation}"
+                        )
+                    tv_maps.append(tv_map)
+                else:
+                    return f"parameter {my_param.name!r} is not accepted"
+            elif my_param.kind is SigParameter.KEYWORD_ONLY:
+                their_param = other.signature.parameters.get(my_param.name)
+                if their_param is not None and their_param.kind in (
+                    SigParameter.POSITIONAL_OR_KEYWORD,
+                    SigParameter.KEYWORD_ONLY,
+                ):
+                    tv_map = their_param.annotation.can_assign(my_param.annotation, ctx)
+                    if tv_map is None:
+                        return (
+                            f"type of parameter {my_param.name!r} is incompatible: "
+                            f"{their_param.annotation} is incompatible with {my_param.annotation}"
+                        )
+                    tv_maps.append(tv_map)
+                    consumed_keyword.add(their_param.name)
+                elif their_kwargs is not None:
+                    tv_map = their_kwargs.annotation.can_assign(
+                        my_param.annotation, ctx
+                    )
+                    if tv_map is None:
+                        return (
+                            f"type of parameter {my_param.name!r} is incompatible: "
+                            f"**kwargs type {their_kwargs.annotation} is incompatible with {my_param.annotation}"
+                        )
+                    tv_maps.append(tv_map)
+                else:
+                    return f"parameter {my_param.name!r} is not accepted"
+            elif my_param.kind is SigParameter.VAR_POSITIONAL:
+                if their_args is None:
+                    return "*args are not accepted"
+                tv_map = their_args.annotation.can_assign(my_param.annotation, ctx)
+                if tv_map is None:
+                    return (
+                        f"type of *args is incompatible: "
+                        f"{their_args.annotation} is incompatible with {my_param.annotation}"
+                    )
+                tv_maps.append(tv_map)
+                extra_positional = [
+                    param
+                    for param in their_params
+                    if param.name not in consumed_positional
+                    and param.kind
+                    in (
+                        SigParameter.POSITIONAL_ONLY,
+                        SigParameter.POSITIONAL_OR_KEYWORD,
+                    )
+                ]
+                for extra_param in extra_positional:
+                    tv_map = extra_param.annotation.can_assign(my_param.annotation, ctx)
+                    if tv_map is None:
+                        return (
+                            f"type of param {extra_param.name!r} is incompatible with "
+                            f"*args type {their_args.annotation}"
+                        )
+                    tv_maps.append(tv_map)
+            elif my_param.kind is SigParameter.VAR_KEYWORD:
+                if their_kwargs is None:
+                    return "**kwargs are not accepted"
+                tv_map = their_kwargs.annotation.can_assign(my_param.annotation, ctx)
+                if tv_map is None:
+                    return (
+                        f"type of **kwargs is incompatible: "
+                        f"{their_kwargs.annotation} is incompatible with {my_param.annotation}"
+                    )
+                tv_maps.append(tv_map)
+                extra_keyword = [
+                    param
+                    for param in their_params
+                    if param.name not in consumed_keyword
+                    and param.kind
+                    in (SigParameter.KEYWORD_ONLY, SigParameter.POSITIONAL_OR_KEYWORD)
+                ]
+                for extra_param in extra_keyword:
+                    tv_map = extra_param.annotation.can_assign(my_param.annotation, ctx)
+                    if tv_map is None:
+                        return (
+                            f"type of param {extra_param.name!r} is incompatible with "
+                            f"**kwargs type {their_kwargs.annotation}"
+                        )
+                    tv_maps.append(tv_map)
+
+        final_tv_map = unify_typevar_maps(tv_maps)
+        if final_tv_map is None:
+            return "callables are incompatible"
+        return final_tv_map
+
+    def get_param_of_kind(self, kind: inspect._ParameterKind) -> Optional[SigParameter]:
+        for param in self.signature.parameters.values():
+            if param.kind is kind:
+                return param
+        return None
+
     @classmethod
     def make(
         cls,


Test patch:
diff --git a/pyanalyze/test_signature.py b/pyanalyze/test_signature.py
index 70657b0c..ca103910 100644
--- a/pyanalyze/test_signature.py
+++ b/pyanalyze/test_signature.py
@@ -1,7 +1,186 @@
 # static analysis: ignore
+from .value import TypedValue, UNRESOLVED_VALUE
 from .test_name_check_visitor import TestNameCheckVisitorBase
 from .test_node_visitor import assert_fails, assert_passes, skip_before
 from .error_code import ErrorCode
+from .signature import Signature, SigParameter as P
+from .test_value import CTX
+
+
+class TestCanAssign:
+    def can(self, left: Signature, right: Signature) -> None:
+        tv_map = left.can_assign(right, CTX)
+        assert isinstance(
+            tv_map, dict
+        ), f"cannot assign {right} to {left} due to {tv_map}"
+
+    def cannot(self, left: Signature, right: Signature) -> None:
+        tv_map = left.can_assign(right, CTX)
+        assert isinstance(tv_map, str), f"can assign {right} to {left}"
+
+    def test_return_value(self) -> None:
+        self.can(
+            Signature.make([], UNRESOLVED_VALUE), Signature.make([], TypedValue(int))
+        )
+        self.can(
+            Signature.make([], TypedValue(int)), Signature.make([], TypedValue(int))
+        )
+        self.can(
+            Signature.make([], TypedValue(int)), Signature.make([], TypedValue(bool))
+        )
+        self.cannot(
+            Signature.make([], TypedValue(bool)), Signature.make([], TypedValue(int))
+        )
+
+    def test_pos_only(self):
+        pos_only_int = P("x", annotation=TypedValue(int), kind=P.POSITIONAL_ONLY)
+        pos_only_object = P("y", annotation=TypedValue(object), kind=P.POSITIONAL_ONLY)
+        pos_only_bool = P("z", annotation=TypedValue(bool), kind=P.POSITIONAL_ONLY)
+        pos_kw_int = P("a", annotation=TypedValue(int), kind=P.POSITIONAL_OR_KEYWORD)
+        pos_only_sig = Signature.make([pos_only_int])
+        self.can(pos_only_sig, pos_only_sig)
+        self.can(pos_only_sig, Signature.make([pos_kw_int]))
+        self.can(pos_only_sig, Signature.make([pos_only_object]))
+        self.cannot(pos_only_sig, Signature.make([pos_only_bool]))
+        self.cannot(
+            pos_only_sig,
+            Signature.make([P("x", annotation=TypedValue(int), kind=P.KEYWORD_ONLY)]),
+        )
+
+        # *args interaction
+        self.can(
+            pos_only_sig,
+            Signature.make(
+                [P("whatever", annotation=TypedValue(int), kind=P.VAR_POSITIONAL)]
+            ),
+        )
+        self.cannot(
+            pos_only_sig,
+            Signature.make(
+                [P("x", annotation=TypedValue(bool), kind=P.VAR_POSITIONAL)]
+            ),
+        )
+
+    def test_pos_or_keyword(self) -> None:
+        pos_kw_int = P("a", annotation=TypedValue(int), kind=P.POSITIONAL_OR_KEYWORD)
+        pos_kw_int_b = P("b", annotation=TypedValue(int), kind=P.POSITIONAL_OR_KEYWORD)
+        pos_kw_object = P(
+            "a", annotation=TypedValue(object), kind=P.POSITIONAL_OR_KEYWORD
+        )
+        pos_kw_bool = P("a", annotation=TypedValue(bool), kind=P.POSITIONAL_OR_KEYWORD)
+        pos_only_int = P("a", annotation=TypedValue(int), kind=P.POSITIONAL_ONLY)
+        pos_kw_sig = Signature.make([pos_kw_int])
+        self.can(pos_kw_sig, pos_kw_sig)
+        self.can(pos_kw_sig, Signature.make([pos_kw_object]))
+        self.cannot(pos_kw_sig, Signature.make([pos_kw_bool]))
+        self.cannot(pos_kw_sig, Signature.make([pos_only_int]))
+        self.cannot(pos_kw_sig, Signature.make([pos_kw_int_b]))
+        self.cannot(
+            pos_kw_sig,
+            Signature.make([P("x", annotation=TypedValue(int), kind=P.VAR_POSITIONAL)]),
+        )
+        self.cannot(
+            pos_kw_sig,
+            Signature.make([P("x", annotation=TypedValue(int), kind=P.VAR_KEYWORD)]),
+        )
+        self.can(
+            pos_kw_sig,
+            Signature.make(
+                [
+                    P("x", annotation=TypedValue(int), kind=P.VAR_POSITIONAL),
+                    P("y", annotation=TypedValue(int), kind=P.VAR_KEYWORD),
+                ]
+            ),
+        )
+        self.can(
+            pos_kw_sig,
+            Signature.make(
+                [
+                    P("x", annotation=TypedValue(object), kind=P.VAR_POSITIONAL),
+                    P("y", annotation=TypedValue(int), kind=P.VAR_KEYWORD),
+                ]
+            ),
+        )
+        self.cannot(
+            pos_kw_sig,
+            Signature.make(
+                [
+                    P("x", annotation=TypedValue(bool), kind=P.VAR_POSITIONAL),
+                    P("y", annotation=TypedValue(int), kind=P.VAR_KEYWORD),
+                ]
+            ),
+        )
+
+    def test_kw_only(self) -> None:
+        kw_only_int = P("a", annotation=TypedValue(int), kind=P.KEYWORD_ONLY)
+        kw_only_int_b = P("b", annotation=TypedValue(int), kind=P.KEYWORD_ONLY)
+        kw_only_bool = P("a", annotation=TypedValue(bool), kind=P.KEYWORD_ONLY)
+        kw_only_object = P("a", annotation=TypedValue(object), kind=P.KEYWORD_ONLY)
+        pos_only_int = P("a", annotation=TypedValue(int), kind=P.POSITIONAL_ONLY)
+        pos_kw_int = P("a", annotation=TypedValue(int), kind=P.POSITIONAL_OR_KEYWORD)
+        kw_only_sig = Signature.make([kw_only_int])
+        self.can(kw_only_sig, kw_only_sig)
+        self.can(kw_only_sig, Signature.make([kw_only_object]))
+        self.cannot(kw_only_sig, Signature.make([kw_only_bool]))
+        self.cannot(kw_only_sig, Signature.make([kw_only_int_b]))
+        self.can(kw_only_sig, Signature.make([pos_kw_int]))
+        self.cannot(kw_only_sig, Signature.make([pos_only_int]))
+        self.can(
+            kw_only_sig,
+            Signature.make([P("x", annotation=TypedValue(int), kind=P.VAR_KEYWORD)]),
+        )
+        self.cannot(
+            kw_only_sig,
+            Signature.make([P("x", annotation=TypedValue(bool), kind=P.VAR_KEYWORD)]),
+        )
+
+    def test_var_positional(self) -> None:
+        var_pos_int = P("a", annotation=TypedValue(int), kind=P.VAR_POSITIONAL)
+        var_pos_object = P("b", annotation=TypedValue(object), kind=P.VAR_POSITIONAL)
+        var_pos_bool = P("c", annotation=TypedValue(bool), kind=P.VAR_POSITIONAL)
+        var_pos_sig = Signature.make([var_pos_int])
+        self.can(var_pos_sig, var_pos_sig)
+        self.can(var_pos_sig, Signature.make([var_pos_object]))
+        self.cannot(var_pos_sig, Signature.make([var_pos_bool]))
+        self.can(
+            var_pos_sig,
+            Signature.make(
+                [
+                    P("d", annotation=TypedValue(object), kind=P.POSITIONAL_ONLY),
+                    var_pos_int,
+                ]
+            ),
+        )
+        self.cannot(
+            var_pos_sig,
+            Signature.make(
+                [
+                    P("d", annotation=TypedValue(bool), kind=P.POSITIONAL_ONLY),
+                    var_pos_int,
+                ]
+            ),
+        )
+
+    def test_var_keyword(self) -> None:
+        var_kw_int = P("a", annotation=TypedValue(int), kind=P.VAR_KEYWORD)
+        var_kw_object = P("b", annotation=TypedValue(object), kind=P.VAR_KEYWORD)
+        var_kw_bool = P("c", annotation=TypedValue(bool), kind=P.VAR_KEYWORD)
+        var_kw_sig = Signature.make([var_kw_int])
+        self.can(var_kw_sig, var_kw_sig)
+        self.can(var_kw_sig, Signature.make([var_kw_object]))
+        self.cannot(var_kw_sig, Signature.make([var_kw_bool]))
+        self.can(
+            var_kw_sig,
+            Signature.make(
+                [P("d", annotation=TypedValue(object), kind=P.KEYWORD_ONLY), var_kw_int]
+            ),
+        )
+        self.cannot(
+            var_kw_sig,
+            Signature.make(
+                [P("d", annotation=TypedValue(bool), kind=P.KEYWORD_ONLY), var_kw_int]
+            ),
+        )
 
 
 class TestProperty(TestNameCheckVisitorBase):
diff --git a/pyanalyze/test_value.py b/pyanalyze/test_value.py
index 6177e56f..69d8648a 100644
--- a/pyanalyze/test_value.py
+++ b/pyanalyze/test_value.py
@@ -38,15 +38,15 @@ def get_generic_bases(
         return self.arg_spec_cache.get_generic_bases(typ, generic_args)
 
 
-_CTX = Context()
+CTX = Context()
 
 
 def assert_cannot_assign(left: Value, right: Value) -> None:
-    assert_is(None, left.can_assign(right, _CTX))
+    assert_is(None, left.can_assign(right, CTX))
 
 
 def assert_can_assign(left: Value, right: Value, typevar_map: TypeVarMap = {}) -> None:
-    assert_eq({}, left.can_assign(right, _CTX))
+    assert_eq({}, left.can_assign(right, CTX))
 
 
 def test_UNRESOLVED_VALUE() -> None:
@@ -393,21 +393,21 @@ def test_io() -> None:
 
 
 def test_concrete_values_from_iterable() -> None:
-    assert_is(None, concrete_values_from_iterable(KnownValue(1), _CTX))
-    assert_eq((), concrete_values_from_iterable(KnownValue([]), _CTX))
+    assert_is(None, concrete_values_from_iterable(KnownValue(1), CTX))
+    assert_eq((), concrete_values_from_iterable(KnownValue([]), CTX))
     assert_eq(
         (KnownValue(1), KnownValue(2)),
-        concrete_values_from_iterable(KnownValue((1, 2)), _CTX),
+        concrete_values_from_iterable(KnownValue((1, 2)), CTX),
     )
     assert_eq(
         (KnownValue(1), KnownValue(2)),
         concrete_values_from_iterable(
-            SequenceIncompleteValue(list, [KnownValue(1), KnownValue(2)]), _CTX
+            SequenceIncompleteValue(list, [KnownValue(1), KnownValue(2)]), CTX
         ),
     )
     assert_eq(
         TypedValue(int),
-        concrete_values_from_iterable(GenericValue(list, [TypedValue(int)]), _CTX),
+        concrete_values_from_iterable(GenericValue(list, [TypedValue(int)]), CTX),
     )
     assert_eq(
         [
@@ -421,7 +421,7 @@ def test_concrete_values_from_iterable() -> None:
                     KnownValue((3, 4)),
                 ]
             ),
-            _CTX,
+            CTX,
         ),
     )
     assert_eq(
@@ -433,7 +433,7 @@ def test_concrete_values_from_iterable() -> None:
                     GenericValue(list, [TypedValue(int)]),
                 ]
             ),
-            _CTX,
+            CTX,
         ),
     )
     assert_eq(
@@ -445,7 +445,7 @@ def test_concrete_values_from_iterable() -> None:
                     KnownValue((3,)),
                 ]
             ),
-            _CTX,
+            CTX,
         ),
     )