diff --git a/.flake8 b/.flake8
index caba1c751..74e4d83b7 100644
--- a/.flake8
+++ b/.flake8
@@ -10,6 +10,6 @@ exclude = .git, __pycache__, build, dist, .eggs, .github, .local, docs/,
           azure_functions_worker/_thirdparty/typing_inspect.py,
           tests/unittests/test_typing_inspect.py,
           tests/unittests/broken_functions/syntax_error/main.py,
-          .env*, .vscode, venv*, *.venv*,
+          .env*, .vscode, venv*, *.venv*
 
 max-line-length = 80
diff --git a/azure_functions_worker/testutils.py b/azure_functions_worker/testutils.py
index e06a01386..62e14a2e3 100644
--- a/azure_functions_worker/testutils.py
+++ b/azure_functions_worker/testutils.py
@@ -745,10 +745,13 @@ def __init__(self, proc, addr):
     def request(self, meth, funcname, *args, **kwargs):
         request_method = getattr(requests, meth.lower())
         params = dict(kwargs.pop('params', {}))
+        no_prefix = kwargs.pop('no_prefix', False)
         if 'code' not in params:
             params['code'] = 'testFunctionKey'
-        return request_method(self._addr + '/api/' + funcname,
-                              *args, params=params, **kwargs)
+
+        return request_method(
+            self._addr + ('/' if no_prefix else '/api/') + funcname,
+            *args, params=params, **kwargs)
 
     def close(self):
         if self._proc.stdout:
diff --git a/setup.py b/setup.py
index 95484f0ef..0554e9852 100644
--- a/setup.py
+++ b/setup.py
@@ -112,6 +112,9 @@
     "dev": [
         "azure-eventhub~=5.7.0",  # Used for EventHub E2E tests
         "python-dateutil~=2.8.2",
+        "flask",
+        "fastapi",
+        "pydantic",
         "pycryptodome~=3.10.1",
         "flake8~=4.0.1",
         "mypy",
diff --git a/tests/endtoend/test_third_party_http_functions.py b/tests/endtoend/test_third_party_http_functions.py
new file mode 100644
index 000000000..8ca63f325
--- /dev/null
+++ b/tests/endtoend/test_third_party_http_functions.py
@@ -0,0 +1,168 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+import os
+from unittest.mock import patch
+
+import requests
+
+from azure_functions_worker import testutils as utils, testutils
+
+# from tests.stein_tests import testutils
+# from tests.stein_tests.constants import E2E_TESTS_ROOT
+from azure_functions_worker.testutils import E2E_TESTS_ROOT
+
+HOST_JSON_TEMPLATE = """\
+{
+    "version": "2.0",
+    "logging": {
+        "logLevel": {
+           "default": "Trace"
+        }
+    },
+    "extensions": {
+        "http": {
+          "routePrefix": ""
+        }
+    },
+    "functionTimeout": "00:05:00"
+}
+"""
+
+
+class ThirdPartyHttpFunctionsTestBase:
+    """Base test class containing common asgi/wsgi testcases, only testcases
+    in classes extending TestThirdPartyHttpFunctions will by run"""
+
+    class TestThirdPartyHttpFunctions(testutils.WebHostTestCase):
+        @classmethod
+        def setUpClass(cls):
+            host_json = cls.get_script_dir() / 'host.json'
+            with open(host_json, 'w+') as f:
+                f.write(HOST_JSON_TEMPLATE)
+            os_environ = os.environ.copy()
+            # Turn on feature flag
+            os_environ['AzureWebJobsFeatureFlags'] = 'EnableWorkerIndexing'
+            cls._patch_environ = patch.dict('os.environ', os_environ)
+            cls._patch_environ.start()
+            super().setUpClass()
+
+        @classmethod
+        def tearDownClass(cls):
+            super().tearDownClass()
+            cls._patch_environ.stop()
+
+        @classmethod
+        def get_script_dir(cls):
+            pass
+
+        @utils.retryable_test(3, 5)
+        def test_function_index_page_should_return_undefined(self):
+            root_url = self.webhost._addr
+            r = requests.get(root_url)
+            self.assertEqual(r.status_code, 404)
+
+        @utils.retryable_test(3, 5)
+        def test_get_endpoint_should_return_ok(self):
+            """Test if the default template of Http trigger in Python
+            Function app
+            will return OK
+            """
+            r = self.webhost.request('GET', 'get_query_param', no_prefix=True)
+            self.assertTrue(r.ok)
+            self.assertEqual(r.text, "hello world")
+
+        @utils.retryable_test(3, 5)
+        def test_get_endpoint_should_accept_query_param(self):
+            """Test if the azure.functions SDK is able to deserialize query
+            parameter from the default template
+            """
+            r = self.webhost.request('GET', 'get_query_param',
+                                     params={'name': 'dummy'}, no_prefix=True)
+            self.assertTrue(r.ok)
+            self.assertEqual(
+                r.text,
+                "hello dummy"
+            )
+
+        @utils.retryable_test(3, 5)
+        def test_post_endpoint_should_accept_body(self):
+            """Test if the azure.functions SDK is able to deserialize http body
+            and pass it to default template
+            """
+            r = self.webhost.request('POST', 'post_str',
+                                     data="dummy",
+                                     headers={'content-type': 'text/plain'},
+                                     no_prefix=True)
+            self.assertTrue(r.ok)
+            self.assertEqual(
+                r.text,
+                "hello dummy"
+            )
+
+        @utils.retryable_test(3, 5)
+        def test_worker_status_endpoint_should_return_ok(self):
+            """Test if the worker status endpoint will trigger
+            _handle__worker_status_request and sends a worker status
+            response back
+            to host
+            """
+            root_url = self.webhost._addr
+            health_check_url = f'{root_url}/admin/host/ping'
+            r = requests.post(health_check_url,
+                              params={'checkHealth': '1'})
+            self.assertTrue(r.ok)
+
+        @utils.retryable_test(3, 5)
+        def test_worker_status_endpoint_should_return_ok_when_disabled(self):
+            """Test if the worker status endpoint will trigger
+            _handle__worker_status_request and sends a worker status
+            response back
+            to host
+            """
+            os.environ['WEBSITE_PING_METRICS_SCALE_ENABLED'] = '0'
+            root_url = self.webhost._addr
+            health_check_url = f'{root_url}/admin/host/ping'
+            r = requests.post(health_check_url,
+                              params={'checkHealth': '1'})
+            self.assertTrue(r.ok)
+
+        @utils.retryable_test(3, 5)
+        def test_get_endpoint_should_accept_path_param(self):
+            r = self.webhost.request('GET', 'get_path_param/1', no_prefix=True)
+            self.assertTrue(r.ok)
+            self.assertEqual(r.text, "hello 1")
+
+        @utils.retryable_test(3, 5)
+        def test_post_json_body_and_return_json_response(self):
+            test_data = {
+                "name": "apple",
+                "description": "yummy"
+            }
+            r = self.webhost.request('POST', 'post_json_return_json_response',
+                                     json=test_data,
+                                     no_prefix=True)
+            self.assertTrue(r.ok)
+            self.assertEqual(r.json(), test_data)
+
+        @utils.retryable_test(3, 5)
+        def test_raise_exception_should_return_not_found(self):
+            r = self.webhost.request('GET', 'raise_http_exception',
+                                     no_prefix=True)
+            self.assertEqual(r.status_code, 404)
+            self.assertEqual(r.json(), {"detail": "Item not found"})
+
+
+class TestAsgiHttpFunctions(
+        ThirdPartyHttpFunctionsTestBase.TestThirdPartyHttpFunctions):
+    @classmethod
+    def get_script_dir(cls):
+        return E2E_TESTS_ROOT / 'third_party_http_functions' / 'stein' / \
+            'asgi_function'
+
+
+class TestWsgiHttpFunctions(
+        ThirdPartyHttpFunctionsTestBase.TestThirdPartyHttpFunctions):
+    @classmethod
+    def get_script_dir(cls):
+        return E2E_TESTS_ROOT / 'third_party_http_functions' / 'stein' / \
+            'wsgi_function'
diff --git a/tests/endtoend/third_party_http_functions/stein/asgi_function/function_app.py b/tests/endtoend/third_party_http_functions/stein/asgi_function/function_app.py
new file mode 100644
index 000000000..0654b349a
--- /dev/null
+++ b/tests/endtoend/third_party_http_functions/stein/asgi_function/function_app.py
@@ -0,0 +1,41 @@
+from typing import Optional
+
+import azure.functions as func
+from fastapi import FastAPI, Response, Body, HTTPException
+from pydantic import BaseModel
+
+fast_app = FastAPI()
+
+
+class Fruit(BaseModel):
+    name: str
+    description: Optional[str] = None
+
+
+@fast_app.get("/get_query_param")
+async def get_query_param(name: str = "world"):
+    return Response(content=f"hello {name}", media_type="text/plain")
+
+
+@fast_app.post("/post_str")
+async def post_str(person: str = Body(...)):
+    return Response(content=f"hello {person}", media_type="text/plain")
+
+
+@fast_app.post("/post_json_return_json_response")
+async def post_json_return_json_response(fruit: Fruit):
+    return fruit
+
+
+@fast_app.get("/get_path_param/{id}")
+async def get_path_param(id):
+    return Response(content=f"hello {id}", media_type="text/plain")
+
+
+@fast_app.get("/raise_http_exception")
+async def raise_http_exception():
+    raise HTTPException(status_code=404, detail="Item not found")
+
+
+app = func.FunctionApp(asgi_app=fast_app,
+                       http_auth_level=func.AuthLevel.ANONYMOUS)
diff --git a/tests/endtoend/third_party_http_functions/stein/wsgi_function/function_app.py b/tests/endtoend/third_party_http_functions/stein/wsgi_function/function_app.py
new file mode 100644
index 000000000..267fb85ef
--- /dev/null
+++ b/tests/endtoend/third_party_http_functions/stein/wsgi_function/function_app.py
@@ -0,0 +1,36 @@
+import azure.functions as func
+from flask import Flask, request
+
+flask_app = Flask(__name__)
+
+
+@flask_app.get("/get_query_param")
+def get_query_param():
+    name = request.args.get("name")
+    if name is None:
+        name = "world"
+    return f"hello {name}"
+
+
+@flask_app.post("/post_str")
+def post_str():
+    return f"hello {request.data.decode()}"
+
+
+@flask_app.post("/post_json_return_json_response")
+def post_json_return_json_response():
+    return request.get_json()
+
+
+@flask_app.get("/get_path_param/<id>")
+def get_path_param(id):
+    return f"hello {id}"
+
+
+@flask_app.get("/raise_http_exception")
+def raise_http_exception():
+    return {"detail": "Item not found"}, 404
+
+
+app = func.FunctionApp(wsgi_app=flask_app.wsgi_app,
+                       http_auth_level=func.AuthLevel.ANONYMOUS)
diff --git a/tests/unittests/test_third_party_http_functions.py b/tests/unittests/test_third_party_http_functions.py
new file mode 100644
index 000000000..8e2aca4e3
--- /dev/null
+++ b/tests/unittests/test_third_party_http_functions.py
@@ -0,0 +1,223 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License
+import filecmp
+import os
+import pathlib
+import re
+import typing
+from unittest.mock import patch
+
+from azure_functions_worker import testutils
+from azure_functions_worker.testutils import UNIT_TESTS_ROOT
+
+HOST_JSON_TEMPLATE = """\
+{
+    "version": "2.0",
+    "logging": {
+        "logLevel": {
+           "default": "Trace"
+        }
+    },
+    "extensions": {
+        "http": {
+          "routePrefix": ""
+        }
+    },
+    "functionTimeout": "00:05:00"
+}
+"""
+
+
+class ThirdPartyHttpFunctionsTestBase:
+    class TestThirdPartyHttpFunctions(testutils.WebHostTestCase):
+
+        @classmethod
+        def setUpClass(cls):
+            host_json = cls.get_script_dir() / 'host.json'
+            with open(host_json, 'w+') as f:
+                f.write(HOST_JSON_TEMPLATE)
+            os_environ = os.environ.copy()
+            # Turn on feature flag
+            os_environ['AzureWebJobsFeatureFlags'] = 'EnableWorkerIndexing'
+            cls._patch_environ = patch.dict('os.environ', os_environ)
+            cls._patch_environ.start()
+            super().setUpClass()
+
+        @classmethod
+        def tearDownClass(cls):
+            super().tearDownClass()
+            cls._patch_environ.stop()
+
+        @classmethod
+        def get_script_dir(cls):
+            pass
+
+        def test_debug_logging(self):
+            r = self.webhost.request('GET', 'debug_logging', no_prefix=True)
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(r.text, 'OK-debug')
+
+        def check_log_debug_logging(self, host_out: typing.List[str]):
+            self.assertIn('logging info', host_out)
+            self.assertIn('logging warning', host_out)
+            self.assertIn('logging error', host_out)
+            self.assertNotIn('logging debug', host_out)
+
+        def test_debug_with_user_logging(self):
+            r = self.webhost.request('GET', 'debug_user_logging',
+                                     no_prefix=True)
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(r.text, 'OK-user-debug')
+
+        def check_log_debug_with_user_logging(self,
+                                              host_out: typing.List[str]):
+            self.assertIn('logging info', host_out)
+            self.assertIn('logging warning', host_out)
+            self.assertIn('logging debug', host_out)
+            self.assertIn('logging error', host_out)
+
+        def test_print_logging_no_flush(self):
+            r = self.webhost.request('GET', 'print_logging?message=Secret42',
+                                     no_prefix=True)
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(r.text, 'OK-print-logging')
+
+        def check_log_print_logging_no_flush(self, host_out: typing.List[str]):
+            self.assertIn('Secret42', host_out)
+
+        def test_print_logging_with_flush(self):
+            r = self.webhost.request('GET',
+                                     'print_logging?flush=true&message'
+                                     '=Secret42',
+                                     no_prefix=True)
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(r.text, 'OK-print-logging')
+
+        def check_log_print_logging_with_flush(self,
+                                               host_out: typing.List[str]):
+            self.assertIn('Secret42', host_out)
+
+        def test_print_to_console_stdout(self):
+            r = self.webhost.request('GET',
+                                     'print_logging?console=true&message'
+                                     '=Secret42',
+                                     no_prefix=True)
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(r.text, 'OK-print-logging')
+
+        def check_log_print_to_console_stdout(self,
+                                              host_out: typing.List[str]):
+            # System logs stdout should not exist in host_out
+            self.assertNotIn('Secret42', host_out)
+
+        def test_print_to_console_stderr(self):
+            r = self.webhost.request('GET', 'print_logging?console=true'
+                                            '&message=Secret42&is_stderr=true',
+                                     no_prefix=True)
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(r.text, 'OK-print-logging')
+
+        def check_log_print_to_console_stderr(self,
+                                              host_out: typing.List[str], ):
+            # System logs stderr should not exist in host_out
+            self.assertNotIn('Secret42', host_out)
+
+        def test_raw_body_bytes(self):
+            parent_dir = pathlib.Path(__file__).parent.parent
+            image_file = parent_dir / 'unittests/resources/functions.png'
+            with open(image_file, 'rb') as image:
+                img = image.read()
+                img_len = len(img)
+                r = self.webhost.request('POST', 'raw_body_bytes', data=img,
+                                         no_prefix=True)
+
+            received_body_len = int(r.headers['body-len'])
+            self.assertEqual(received_body_len, img_len)
+
+            body = r.content
+            try:
+                received_img_file = parent_dir / 'received_img.png'
+                with open(received_img_file, 'wb') as received_img:
+                    received_img.write(body)
+                self.assertTrue(filecmp.cmp(received_img_file, image_file))
+            finally:
+                if (os.path.exists(received_img_file)):
+                    os.remove(received_img_file)
+
+        def test_return_http_no_body(self):
+            r = self.webhost.request('GET', 'return_http_no_body',
+                                     no_prefix=True)
+            self.assertEqual(r.text, '')
+            self.assertEqual(r.status_code, 200)
+
+        def test_return_http_redirect(self):
+            r = self.webhost.request('GET', 'return_http_redirect',
+                                     no_prefix=True)
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(r.text, '<h1>Hello World™</h1>')
+
+            r = self.webhost.request('GET', 'return_http_redirect',
+                                     allow_redirects=False, no_prefix=True)
+            self.assertEqual(r.status_code, 302)
+
+        def test_unhandled_error(self):
+            r = self.webhost.request('GET', 'unhandled_error', no_prefix=True)
+            self.assertEqual(r.status_code, 500)
+            # https://github.com/Azure/azure-functions-host/issues/2706
+            # self.assertIn('ZeroDivisionError', r.text)
+
+        def check_log_unhandled_error(self,
+                                      host_out: typing.List[str]):
+            r = re.compile(".*ZeroDivisionError: division by zero.*")
+            error_log = list(filter(r.match, host_out))
+            self.assertGreaterEqual(len(error_log), 1)
+
+        def test_unhandled_unserializable_error(self):
+            r = self.webhost.request(
+                'GET', 'unhandled_unserializable_error', no_prefix=True)
+            self.assertEqual(r.status_code, 500)
+
+        def test_unhandled_urllib_error(self):
+            r = self.webhost.request(
+                'GET', 'unhandled_urllib_error',
+                params={'img': 'http://example.com/nonexistent.jpg'},
+                no_prefix=True)
+            self.assertEqual(r.status_code, 500)
+
+
+class TestAsgiHttpFunctions(
+        ThirdPartyHttpFunctionsTestBase.TestThirdPartyHttpFunctions):
+    @classmethod
+    def get_script_dir(cls):
+        return UNIT_TESTS_ROOT / 'third_party_http_functions' / 'stein' / \
+            'asgi_function'
+
+    def test_hijack_current_event_loop(self):
+        r = self.webhost.request('GET', 'hijack_current_event_loop',
+                                 no_prefix=True)
+        self.assertEqual(r.status_code, 200)
+        self.assertEqual(r.text, 'OK-hijack-current-event-loop')
+
+    def check_log_hijack_current_event_loop(self,
+                                            host_out: typing.List[str]):
+        # User logs should exist in host_out
+        self.assertIn('parallelly_print', host_out)
+        self.assertIn('parallelly_log_info at root logger', host_out)
+        self.assertIn('parallelly_log_warning at root logger', host_out)
+        self.assertIn('parallelly_log_error at root logger', host_out)
+        self.assertIn('parallelly_log_exception at root logger',
+                      host_out)
+        self.assertIn('parallelly_log_custom at custom_logger', host_out)
+        self.assertIn('callsoon_log', host_out)
+
+        # System logs should not exist in host_out
+        self.assertNotIn('parallelly_log_system at disguised_logger',
+                         host_out)
+
+
+class TestWsgiHttpFunctions(
+        ThirdPartyHttpFunctionsTestBase.TestThirdPartyHttpFunctions):
+    @classmethod
+    def get_script_dir(cls):
+        return UNIT_TESTS_ROOT / 'third_party_http_functions' / 'stein' / \
+            'wsgi_function'
diff --git a/tests/unittests/third_party_http_functions/stein/asgi_function/function_app.py b/tests/unittests/third_party_http_functions/stein/asgi_function/function_app.py
new file mode 100644
index 000000000..7ac326783
--- /dev/null
+++ b/tests/unittests/third_party_http_functions/stein/asgi_function/function_app.py
@@ -0,0 +1,176 @@
+import asyncio
+import logging
+import sys
+from urllib.request import urlopen
+
+import azure.functions as func
+from fastapi import FastAPI, Request, Response
+from fastapi.responses import RedirectResponse
+
+fast_app = FastAPI()
+logger = logging.getLogger("my-function")
+# Attempt to log info into system log from customer code
+disguised_logger = logging.getLogger('azure_functions_worker')
+
+
+async def parallelly_print():
+    await asyncio.sleep(0.1)
+    print('parallelly_print')
+
+
+async def parallelly_log_info():
+    await asyncio.sleep(0.2)
+    logging.info('parallelly_log_info at root logger')
+
+
+async def parallelly_log_warning():
+    await asyncio.sleep(0.3)
+    logging.warning('parallelly_log_warning at root logger')
+
+
+async def parallelly_log_error():
+    await asyncio.sleep(0.4)
+    logging.error('parallelly_log_error at root logger')
+
+
+async def parallelly_log_exception():
+    await asyncio.sleep(0.5)
+    try:
+        raise Exception('custom exception')
+    except Exception:
+        logging.exception('parallelly_log_exception at root logger',
+                          exc_info=sys.exc_info())
+
+
+async def parallelly_log_custom():
+    await asyncio.sleep(0.6)
+    logger.info('parallelly_log_custom at custom_logger')
+
+
+async def parallelly_log_system():
+    await asyncio.sleep(0.7)
+    disguised_logger.info('parallelly_log_system at disguised_logger')
+
+
+@fast_app.get("/debug_logging")
+async def debug_logging():
+    logging.critical('logging critical', exc_info=True)
+    logging.info('logging info', exc_info=True)
+    logging.warning('logging warning', exc_info=True)
+    logging.debug('logging debug', exc_info=True)
+    logging.error('logging error', exc_info=True)
+
+    return Response(content='OK-debug', media_type="text/plain")
+
+
+@fast_app.get("/debug_user_logging")
+async def debug_user_logging():
+    logger.setLevel(logging.DEBUG)
+
+    logger.critical('logging critical', exc_info=True)
+    logger.info('logging info', exc_info=True)
+    logger.warning('logging warning', exc_info=True)
+    logger.debug('logging debug', exc_info=True)
+    logger.error('logging error', exc_info=True)
+
+    return Response(content='OK-user-debug', media_type="text/plain")
+
+
+@fast_app.get("/hijack_current_event_loop")
+async def hijack_current_event_loop():
+    loop = asyncio.get_event_loop()
+
+    # Create multiple tasks and schedule it into one asyncio.wait blocker
+    task_print: asyncio.Task = loop.create_task(parallelly_print())
+    task_info: asyncio.Task = loop.create_task(parallelly_log_info())
+    task_warning: asyncio.Task = loop.create_task(parallelly_log_warning())
+    task_error: asyncio.Task = loop.create_task(parallelly_log_error())
+    task_exception: asyncio.Task = loop.create_task(parallelly_log_exception())
+    task_custom: asyncio.Task = loop.create_task(parallelly_log_custom())
+    task_disguise: asyncio.Task = loop.create_task(parallelly_log_system())
+
+    # Create an awaitable future and occupy the current event loop resource
+    future = loop.create_future()
+    loop.call_soon_threadsafe(future.set_result, 'callsoon_log')
+
+    # WaitAll
+    await asyncio.wait([task_print, task_info, task_warning, task_error,
+                        task_exception, task_custom, task_disguise, future])
+
+    # Log asyncio low-level future result
+    logging.info(future.result())
+
+    return Response(content='OK-hijack-current-event-loop',
+                    media_type="text/plain")
+
+
+@fast_app.get("/print_logging")
+async def print_logging(message: str = "", flush: str = 'false',
+                        console: str = 'false', is_stderr: str = 'false'):
+    flush_required = False
+    is_console_log = False
+    is_stderr = False
+
+    if flush == 'true':
+        flush_required = True
+    if console == 'true':
+        is_console_log = True
+    if is_stderr == 'true':
+        is_stderr = True
+
+    # Adding LanguageWorkerConsoleLog will make function host to treat
+    # this as system log and will be propagated to kusto
+    prefix = 'LanguageWorkerConsoleLog' if is_console_log else ''
+    print(f'{prefix} {message}'.strip(),
+          file=sys.stderr if is_stderr else sys.stdout,
+          flush=flush_required)
+
+    return Response(content='OK-print-logging', media_type="text/plain")
+
+
+@fast_app.post("/raw_body_bytes")
+async def raw_body_bytes(request: Request):
+    raw_body = await request.body()
+    return Response(content=raw_body, headers={'body-len': str(len(raw_body))})
+
+
+@fast_app.get("/return_http_no_body")
+async def return_http_no_body():
+    return Response(content='', media_type="text/plain")
+
+
+@fast_app.get("/return_http")
+async def return_http(request: Request):
+    return Response('<h1>Hello World™</h1>', media_type='text/html')
+
+
+@fast_app.get("/return_http_redirect")
+async def return_http_redirect(request: Request, code: str = ''):
+    location = 'return_http?code={}'.format(code)
+    return RedirectResponse(status_code=302,
+                            url=f"http://{request.url.components[1]}/"
+                                f"{location}")
+
+
+@fast_app.get("/unhandled_error")
+async def unhandled_error():
+    1 / 0
+
+
+@fast_app.get("/unhandled_urllib_error")
+async def unhandled_urllib_error(img: str = ''):
+    urlopen(img).read()
+
+
+class UnserializableException(Exception):
+    def __str__(self):
+        raise RuntimeError('cannot serialize me')
+
+
+@fast_app.get("/unhandled_unserializable_error")
+async def unhandled_unserializable_error():
+    raise UnserializableException('foo')
+
+
+app = func.FunctionApp(asgi_app=fast_app,
+                       http_auth_level=func.AuthLevel.ANONYMOUS)
diff --git a/tests/unittests/third_party_http_functions/stein/wsgi_function/function_app.py b/tests/unittests/third_party_http_functions/stein/wsgi_function/function_app.py
new file mode 100644
index 000000000..dbd2c311a
--- /dev/null
+++ b/tests/unittests/third_party_http_functions/stein/wsgi_function/function_app.py
@@ -0,0 +1,103 @@
+import logging
+import sys
+from urllib.request import urlopen
+
+import azure.functions as func
+from flask import Flask, request, Response, redirect, url_for
+
+flask_app = Flask(__name__)
+logger = logging.getLogger("my-function")
+
+
+@flask_app.get("/debug_logging")
+def debug_logging():
+    logging.critical('logging critical', exc_info=True)
+    logging.info('logging info', exc_info=True)
+    logging.warning('logging warning', exc_info=True)
+    logging.debug('logging debug', exc_info=True)
+    logging.error('logging error', exc_info=True)
+
+    return 'OK-debug'
+
+
+@flask_app.get("/debug_user_logging")
+def debug_user_logging():
+    logger.setLevel(logging.DEBUG)
+
+    logger.critical('logging critical', exc_info=True)
+    logger.info('logging info', exc_info=True)
+    logger.warning('logging warning', exc_info=True)
+    logger.debug('logging debug', exc_info=True)
+    logger.error('logging error', exc_info=True)
+    return 'OK-user-debug'
+
+
+@flask_app.get("/print_logging")
+def print_logging():
+    flush_required = False
+    is_console_log = False
+    is_stderr = False
+
+    message = request.args.get("message", '')
+
+    if request.args.get("flush") == 'true':
+        flush_required = True
+    if request.args.get("console") == 'true':
+        is_console_log = True
+    if request.args.get("is_stderr") == 'true':
+        is_stderr = True
+
+    # Adding LanguageWorkerConsoleLog will make function host to treat
+    # this as system log and will be propagated to kusto
+    prefix = 'LanguageWorkerConsoleLog' if is_console_log else ''
+    print(f'{prefix} {message}'.strip(),
+          file=sys.stderr if is_stderr else sys.stdout,
+          flush=flush_required)
+
+    return 'OK-print-logging'
+
+
+@flask_app.post("/raw_body_bytes")
+def raw_body_bytes():
+    body = request.get_data()
+
+    return Response(body, headers={'body-len': str(len(body))})
+
+
+@flask_app.get("/return_http_no_body")
+def return_http_no_body():
+    return ''
+
+
+@flask_app.get("/return_http")
+def return_http():
+    return Response('<h1>Hello World™</h1>', mimetype='text/html')
+
+
+@flask_app.get("/return_http_redirect")
+def return_http_redirect(code: str = ''):
+    return redirect(url_for('return_http'))
+
+
+@flask_app.get("/unhandled_error")
+def unhandled_error():
+    1 / 0
+
+
+@flask_app.get("/unhandled_urllib_error")
+def unhandled_urllib_error(img: str = ''):
+    urlopen(img).read()
+
+
+class UnserializableException(Exception):
+    def __str__(self):
+        raise RuntimeError('cannot serialize me')
+
+
+@flask_app.get("/unhandled_unserializable_error")
+def unhandled_unserializable_error():
+    raise UnserializableException('foo')
+
+
+app = func.FunctionApp(wsgi_app=flask_app.wsgi_app,
+                       http_auth_level=func.AuthLevel.ANONYMOUS)