Minor improvements in testing, builds and pipeline (#707)

* Doing nit fixes
- Adding flake8 description
- Adding typing info for parameters
- fixing minor flake8 issues.
- Using pytest's plugins to improve static testing, and using it to do coverage.
- Adding schedule for running tests on branches

Author: vrdmr

.ci/linux_devops_code_coverage_generate.sh

@@ -6,4 +6,4 @@ export AzureWebJobsCosmosDBConnectionString=$LINUXCOSMOSDBCONNECTIONSTRING
 export AzureWebJobsEventHubConnectionString=$LINUXEVENTHUBCONNECTIONSTRING
 export AzureWebJobsServiceBusConnectionString=$LINUXSERVICEBUSCONNECTIONSTRING
 
-coverage run -p --branch -m pytest tests/endtoend
+pytest  --instafail --cov=./azure_functions_worker --cov-report xml --cov-branch --cov-append tests/endtoend

.ci/linux_devops_e2e_tests.sh

@@ -1,4 +1,4 @@
 #!/bin/bash
 
 set -e -x
-coverage run -p --branch -m pytest tests/unittests
+pytest  --instafail --cov=./azure_functions_worker --cov-report xml --cov-branch tests/unittests

.ci/linux_devops_unit_tests.sh

@@ -1,8 +1,13 @@
 [flake8]
+# it's not a bug that we aren't using all of hacking, ignore:
+# H402: Module level import not at top of file
+# W503: Line break occurred before a binary operator
+# E731: Do not assign a lambda expression, use a def
 ignore = W503,E402,E731
-exclude =
-    .git, __pycache__, build, dist, .eggs, .github, .local, docs/,
-    Samples, azure_functions_worker/protos/,
-    azure_functions_worker/_thirdparty/typing_inspect.py,
-    tests/unittests/test_typing_inspect.py,
-    .venv*, .env*, .vscode, venv
+
+exclude = .git, __pycache__, build, dist, .eggs, .github, .local, docs/,
+          Samples, azure_functions_worker/protos/,
+          azure_functions_worker/_thirdparty/typing_inspect.py,
+          tests/unittests/test_typing_inspect.py, .venv*, .env*, .vscode, venv
+
+max-line-length = 80

.flake8

@@ -8,6 +8,17 @@ trigger:
 variables:
     DOTNET_VERSION: '2.2.207'
 
+schedules:
+- cron: "0 0 * * *"
+  displayName: Daily midnight build
+  branches:
+    include:
+      - dev
+      - master
+      - release/*
+    exclude:
+    - releases/ancient/*
+
 jobs:
 - job: Tests
   pool:
@@ -60,10 +71,6 @@ jobs:
       LINUXEVENTHUBCONNECTIONSTRING: $(linuxEventHub)
       LINUXSERVICEBUSCONNECTIONSTRING: $(linuxServiceBus)
     displayName: 'E2E Tests'
-  - bash: |
-      chmod +x .ci/linux_devops_code_coverage_generate.sh
-      .ci/linux_devops_code_coverage_generate.sh
-    displayName: 'Generate Code Coverage XML file'
   - task: PublishCodeCoverageResults@1
     inputs:
         codeCoverageTool: cobertura

azure-pipelines.yml

@@ -46,15 +46,17 @@ class Dispatcher(metaclass=DispatcherMeta):
 
     _GRPC_STOP_RESPONSE = object()
 
-    def __init__(self, loop, host, port, worker_id, request_id,
-                 grpc_connect_timeout, grpc_max_msg_len=-1):
+    def __init__(self, loop, host, port: int, worker_id: str, request_id: str,
+                 grpc_connect_timeout: float, grpc_max_msg_len: int = -1):
         self._loop = loop
         self._host = host
         self._port = port
         self._request_id = request_id
         self._worker_id = worker_id
         self._functions = functions.Registry()
 
+        self._old_task_factory = None
+
         # A thread-pool for synchronous function calls.  We limit
         # the number of threads to 1 so that one Python worker can
         # only run one synchronous function in parallel.  This is
@@ -75,7 +77,8 @@ def __init__(self, loop, host, port, worker_id, request_id,
         self._grpc_thread = threading.Thread(
             name='grpc-thread', target=self.__poll_grpc)
 
-    def load_bindings(self):
+    @staticmethod
+    def load_bindings():
         """Load out-of-tree binding implementations."""
         services = {}
 
@@ -88,18 +91,17 @@ def load_bindings(self):
     @classmethod
     async def connect(cls, host, port, worker_id, request_id,
                       connect_timeout):
-        loop = asyncio._get_running_loop()
-        disp = cls(loop, host, port, worker_id, request_id,
-                   connect_timeout)
+        loop = asyncio.events.get_event_loop()
+        disp = cls(loop, host, port, worker_id, request_id, connect_timeout)
         disp._grpc_thread.start()
         await disp._grpc_connected_fut
         logger.info('Successfully opened gRPC channel to %s:%s', host, port)
         return disp
 
     async def dispatch_forever(self):
         if DispatcherMeta.__current_dispatcher__ is not None:
-            raise RuntimeError(
-                'there can be only one running dispatcher per process')
+            raise RuntimeError('there can be only one running dispatcher per '
+                               'process')
 
         self._old_task_factory = self._loop.get_task_factory()
 
@@ -131,7 +133,7 @@ async def dispatch_forever(self):
             try:
                 await forever
             finally:
-                # Reenable console logging when there's an exception
+                # Re-enable console logging when there's an exception
                 enable_console_logging()
                 root_logger.removeHandler(logging_handler)
         finally:
@@ -152,7 +154,7 @@ def stop(self):
             self._sync_call_tp.shutdown()
             self._sync_call_tp = None
 
-    def _on_logging(self, record: logging.LogRecord, formatted_msg: str):
+    def on_logging(self, record: logging.LogRecord, formatted_msg: str):
         if record.levelno >= logging.CRITICAL:
             log_level = protos.RpcLog.Critical
         elif record.levelno >= logging.ERROR:
@@ -201,7 +203,9 @@ def request_id(self):
     def worker_id(self):
         return self._worker_id
 
-    def _serialize_exception(self, exc):
+    # noinspection PyBroadException
+    @staticmethod
+    def _serialize_exception(exc: Exception):
         try:
             message = f'{type(exc).__name__}: {exc}'
         except Exception:
@@ -222,8 +226,8 @@ async def _dispatch_grpc_request(self, request):
             # 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.
-            logger.error(
-                f'unknown StreamingMessage content type {content_type}')
+            logger.error(f'unknown StreamingMessage content type '
+                         f'{content_type}')
             return
 
         resp = await request_handler(request)
@@ -524,7 +528,7 @@ def emit(self, record):
         # Since we disable console log after gRPC channel is initiated
         # We should redirect all the messages into dispatcher
         msg = self.format(record)
-        Dispatcher.current._on_logging(record, msg)
+        Dispatcher.current.on_logging(record, msg)
 
 
 class ContextEnabledTask(asyncio.Task):

azure_functions_worker/dispatcher.py

@@ -290,7 +290,12 @@ def run(self):
             'mypy',
             'pytest',
             'requests==2.*',
-            'coverage'
+            'coverage',
+            'pytest-sugar',
+            'pytest-cov',
+            'pytest-xdist',
+            'pytest-randomly',
+            'pytest-instafail'
         ]
     },
     include_package_data=True,

setup.py

@@ -94,7 +94,7 @@ async def _runner():
 
             # Trimming off carriage return charater when testing on Windows
             stdout_lines = [
-                l.replace(b'\r', b'') for l in stdout.strip().split(b'\n')
+                line.replace(b'\r', b'') for line in stdout.strip().split(b'\n')
             ]
             self.assertEqual(stdout_lines, [b'True', b'True'])
 

tests/unittests/test_loader.py

@@ -22,7 +22,8 @@ async def test_call_sync_function_check_logs(self):
             self.assertEqual(r.response.result.status,
                              protos.StatusResult.Success)
 
-            user_logs = [l for l in r.logs if l.category == 'my function']
+            user_logs = [line for line in r.logs
+                         if line.category == 'my function']
             self.assertEqual(len(user_logs), 1)
 
             log = user_logs[0]
@@ -48,7 +49,8 @@ async def test_call_async_function_check_logs(self):
             self.assertEqual(r.response.result.status,
                              protos.StatusResult.Success)
 
-            user_logs = [l for l in r.logs if l.category == 'my function']
+            user_logs = [line for line in r.logs if
+                         line.category == 'my function']
             self.assertEqual(len(user_logs), 2)
 
             first_msg = user_logs[0]