Adding @typed_rpc tests

spyne/decorator.py

@@ -578,7 +578,10 @@ def typed_rpc(*args, **kwargs):
     no_args = False
 
     if len(args) > 0 and inspect.isclass(args[0]):
-        raise Exception("*params must be empty when type annotations are used")
+        raise ValueError("*params must be empty when type annotations are used")
+
+    if "_returns" in kwargs:
+        raise ValueError("_returns must be omitted when type annotations are used. Please annotate the return type")
 
     if len(args) == 1 and not kwargs and callable(args[0]):
         # Called without args
@@ -587,15 +590,22 @@ def typed_rpc(*args, **kwargs):
     def _typed_rpc(func):
         inputs = []
         definition = inspect.signature(func)
+        missing_type_annotations = []
         for param_name, param_type in definition.parameters.items():
-            if param_name == "self":
+            if param_name in ("self", "ctx"):
                 continue
+            if param_type.annotation is inspect._empty:
+                missing_type_annotations.append(param_name)
             inputs.append(param_type.annotation)
+
+        if missing_type_annotations:
+            caller = inspect.getframeinfo(inspect.stack()[2][0])
+            raise ValueError(f"{caller.filename}:{caller.lineno} - Missing type annotation for the parameters: {missing_type_annotations}")
 
-        if definition.return_annotation is inspect._empty:
-            raise Exception("_returns must be omitted when type annotations are used. Please annotate the return type")
-
-        new_func = rpc(*inputs, _returns=definition.return_annotation, **kwargs)(func)
+        if definition.return_annotation is not inspect._empty:
+            new_func = rpc(*inputs, _returns=definition.return_annotation, **kwargs)(func)
+        else:
+            new_func = rpc(*inputs, **kwargs)(func)
 
         @wraps(new_func)
         def wrapper(*args, **kwargs):

spyne/test/test_service.py

@@ -35,7 +35,7 @@
 from spyne.model.primitive import NATIVE_MAP
 
 from spyne.service import Service
-from spyne.decorator import rpc, srpc
+from spyne.decorator import rpc, srpc, typed_rpc
 from spyne.application import Application
 from spyne.auxproc.sync import SyncAuxProc
 from spyne.auxproc.thread import ThreadAuxProc
@@ -505,6 +505,71 @@ def method(ctx):
             raise Exception("Must fail with: "
                         "'SelfReference can't be used inside @rpc and its ilk'")
 
+    def test_typed_rpc_works_with_no_kwargs(self):
+        class someCallResponse(ComplexModel):
+            __namespace__ = 'tns'
+            s = String
+
+        class SomeService(Service):
+            @typed_rpc
+            def someCall(ctx, x: someCallResponse) -> Array(String):
+                return ['abc', 'def']
+
+    def test_typed_rpc_works_with_kwargs(self):
+        class someCallResponse(ComplexModel):
+            __namespace__ = 'tns'
+            s = String
+
+        class SomeService(Service):
+            @typed_rpc(_is_async=True)
+            def someCall(ctx, x: someCallResponse) -> Array(String):
+                return ['abc', 'def']
+
+
+    def test_typed_rpc_works_with_no_response_works(self):
+        class someCallResponse(ComplexModel):
+            __namespace__ = 'tns'
+            s = String
+
+        class SomeService(Service):
+            @typed_rpc(_is_async=True)
+            def someCall(ctx, x: someCallResponse):
+                return ['abc', 'def']
+
+    def test_typed_rpc_works_with__returns_kwarg_raises(self):
+        class someCallResponse(ComplexModel):
+            __namespace__ = 'tns'
+            s = String
+
+        expected_message = "_returns must be omitted when type annotations are used. Please annotate the return type"
+        try:
+            class SomeService(Service):
+                @typed_rpc(_returns=Array(String))
+                def someCall(ctx, x: someCallResponse) -> Array(String):
+                    return ['abc', 'def']
+        except ValueError as e:
+            assert str(e) == expected_message
+        else:
+            raise Exception(f"Must fail with: ValueError('{expected_message}'")
+
+    def test_typed_rpc_works_with_missing_type_annotation_raises(self):
+        class someCallResponse(ComplexModel):
+            __namespace__ = 'tns'
+            s = String
+
+
+        expected_sub_message = "Missing type annotation for the parameters: ['y']"
+        try:
+            class SomeService(Service):
+                @typed_rpc(_is_async=True)
+                def someCall(ctx, x: someCallResponse, y) -> Array(String):
+                    return ['abc', 'def']
+        except ValueError as e:
+            assert expected_sub_message in str(e)
+        else:
+            raise Exception(f"Must fail with: ValueError('{expected_sub_message}'")
+
+
 
 if __name__ == '__main__':
     unittest.main()