Merge pull request #25 from MagicStack/poc

prohibit "inout" bindings; ensure functions without "$return" work; don't crash on unsupported messages

Author: asavaritayal

azure/worker/dispatcher.py

@@ -29,6 +29,7 @@ class FunctionInfo(typing.NamedTuple):
  outputs: typing.Set[str]
  requires_context: bool
  is_async: bool
+ has_return: bool
 
 
 class DispatcherMeta(type):
@@ -74,6 +75,8 @@ def __init__(self, loop, host, port, worker_id, request_id,
  self._grpc_thread = threading.Thread(
  name='grpc-thread', target=self.__poll_grpc)
 
+ self._logger = logging.getLogger('python-azure-worker')
+
  @classmethod
  async def connect(cls, host, port, worker_id, request_id,
  connect_timeout):
@@ -184,14 +187,25 @@ def _register_function(self, function_id: str, func: callable,
  bindings = {}
  return_binding = None
  for name, desc in metadata.bindings.items():
+ if desc.direction == protos.BindingInfo.inout:
+ raise TypeError(
+ f'cannot load the {func_name} function: '
+ f'"inout" bindings are not supported')
+
  if name == '$return':
  # TODO:
  # * add proper gRPC->Python type reflection;
  # * convert the type from function.json to a Python type;
  # * enforce return type of a function call in Python;
  # * use the return type information to marshal the result into
  # a correct gRPC type.
- return_binding = desc # NoQA
+
+ if desc.direction != protos.BindingInfo.out:
+ raise TypeError(
+ f'cannot load the {func_name} function: '
+ f'"$return" binding must have direction set to "out"')
+
+ return_binding = desc
  else:
  bindings[name] = desc
 
@@ -251,14 +265,19 @@ def _register_function(self, function_id: str, func: callable,
  directory=metadata.directory,
  outputs=frozenset(outputs),
  requires_context=requires_context,
- is_async=inspect.iscoroutinefunction(func))
+ is_async=inspect.iscoroutinefunction(func),
+ has_return=return_binding is not None)
 
  async def _dispatch_grpc_request(self, request):
  content_type = request.WhichOneof('content')
  request_handler = getattr(self, f'_handle__{content_type}', None)
  if request_handler is None:
- raise RuntimeError(
+ # Don't crash on unknown messages. Some of them can be ignored;
+ # and if something goes really wrong the host can always just
+ # kill the worker's process.
+ self._logger.error(
  f'unknown StreamingMessage content type {content_type}')
+ return
 
  resp = await request_handler(request)
  self._grpc_resp_queue.put_nowait(resp)
@@ -349,11 +368,15 @@ async def _handle__invocation_request(self, req):
  name=name,
  data=rpc_val))
 
+ return_value = None
+ if fi.has_return:
+ return_value = rpc_types.to_outgoing_proto(call_result)
+
  return protos.StreamingMessage(
  request_id=self.request_id,
  invocation_response=protos.InvocationResponse(
  invocation_id=invocation_id,
- return_value=rpc_types.to_outgoing_proto(call_result),
+ return_value=return_value,
  result=protos.StatusResult(
  status=protos.StatusResult.Success),
  output_data=output_data))

azure/worker/protos/__init__.py

@@ -13,6 +13,7 @@
  FunctionLoadResponse,
  InvocationRequest,
  InvocationResponse,
+ WorkerHeartbeat,
  BindingInfo,
  StatusResult,
  RpcException,

azure/worker/tests/broken_functions/inout_param/function.json

@@ -0,0 +1,17 @@
+{
+ "scriptFile": "main.py",
+ "disabled": false,
+ "bindings": [
+ {
+ "authLevel": "anonymous",
+ "type": "httpTrigger",
+ "direction": "in",
+ "name": "req"
+ },
+ {
+ "type": "http",
+ "direction": "inout",
+ "name": "abc"
+ }
+ ]
+}

azure/worker/tests/broken_functions/inout_param/main.py

@@ -0,0 +1,2 @@
+def main(req, abc):
+ return 'trust me, it is OK!'

azure/worker/tests/broken_functions/return_param_in/function.json

@@ -0,0 +1,17 @@
+{
+ "scriptFile": "main.py",
+ "disabled": false,
+ "bindings": [
+ {
+ "authLevel": "anonymous",
+ "type": "httpTrigger",
+ "direction": "in",
+ "name": "req"
+ },
+ {
+ "type": "http",
+ "direction": "in",
+ "name": "$return"
+ }
+ ]
+}

azure/worker/tests/broken_functions/return_param_in/main.py

@@ -0,0 +1,2 @@
+def main(req):
+ return 'trust me, it is OK!'

azure/worker/tests/functions/no_return/function.json

@@ -0,0 +1,12 @@
+{
+ "scriptFile": "main.py",
+ "disabled": false,
+ "bindings": [
+ {
+ "authLevel": "anonymous",
+ "type": "httpTrigger",
+ "direction": "in",
+ "name": "req"
+ }
+ ]
+}

azure/worker/tests/functions/no_return/main.py

@@ -0,0 +1,8 @@
+import logging
+
+
+logger = logging.getLogger('test')
+
+
+def main(req):
+ logger.error('hi')

azure/worker/tests/test_broken_functions.py

@@ -93,3 +93,33 @@ async def test_load_broken__syntax_error(self):
  protos.StatusResult.Failure)
 
  self.assertIn('SyntaxError', r.response.result.exception.message)
+
+ async def test_load_broken__inout_param(self):
+ async with testutils.start_mockhost(
+ script_root='broken_functions') as host:
+
+ func_id, r = await host.load_function('inout_param')
+
+ self.assertEqual(r.response.function_id, func_id)
+ self.assertEqual(r.response.result.status,
+ protos.StatusResult.Failure)
+
+ self.assertRegex(
+ r.response.result.exception.message,
+ r'.*cannot load the inout_param function'
+ r'.*"inout" bindings.*')
+
+ async def test_load_broken__return_param_in(self):
+ async with testutils.start_mockhost(
+ script_root='broken_functions') as host:
+
+ func_id, r = await host.load_function('return_param_in')
+
+ self.assertEqual(r.response.function_id, func_id)
+ self.assertEqual(r.response.result.status,
+ protos.StatusResult.Failure)
+
+ self.assertRegex(
+ r.response.result.exception.message,
+ r'.*cannot load the return_param_in function'
+ r'.*"\$return" .* set to "out"')

azure/worker/tests/test_functions.py

@@ -19,6 +19,10 @@ def test_return_str(self):
  self.assertEqual(r.status_code, 200)
  self.assertEqual(r.text, 'Hello World!')
 
+ def test_no_return(self):
+ r = self.webhost.request('GET', 'no_return')
+ self.assertEqual(r.status_code, 204)
+
  def test_async_return_str(self):
  r = self.webhost.request('GET', 'async_return_str')
  self.assertEqual(r.status_code, 200)