Repository: falconry/falcon
Pull request #1766 (2020-08-27 02:03:43)

feat(testing.ASGIConductor): Implement an async test conductor for ASGI apps
# Summary of Changes

This patch provides support for asynchronous interaction with the testing
client, including streaming updates to the response body.

This is mostly relevant for testing ASGI SSE and WebSocket (when implemented).

# Related Issues

#1634

# Pull Request Checklist

This is just a reminder about the most common mistakes.  Please make sure that you tick all *appropriate* boxes.  But please read our [contribution guide](../CONTRIBUTING.md) at least once; it will save you a few review cycles!

If an item doesn't apply to your pull request, **check it anyway** to make it apparent that there's nothing to do.

- [x] Applied changes to both WSGI and ASGI code paths and interfaces (where applicable).
- [x] Added **tests** for changed code.
- [x] Prefixed code comments with GitHub nick and an appropriate prefix.
- [x] Coding style is consistent with the rest of the framework.
- [x] Updated **documentation** for changed code.
    - [x] Added docstrings for any new classes, functions, or modules.
    - [x] Updated docstrings for any modifications to existing code.
    - [x] Updated both WSGI and ASGI docs (where applicable).
    - [x] Added references to new classes, functions, or modules to the relevant RST file under `docs/`.
    - [x] Updated all relevant supporting documentation files under `docs/`.
    - [x] A copyright notice is included at the top of any new modules (using your own name or the name of your organization).
    - [x] Changed/added classes/methods/functions have appropriate `versionadded`, `versionchanged`, or `deprecated` [directives](http://www.sphinx-doc.org/en/stable/usage/restructuredtext/directives.html?highlight=versionadded#directive-versionadded).
- [x] Changes (and possible deprecations) have [towncrier](https://towncrier.readthedocs.io/en/actual-freaking-docs/index.html) news fragments under `docs/_newsfragments/`, with the file name format `{issue_number}.{fragment_type}.rst`. (Run `towncrier --draft` to ensure it renders correctly.)

If you have *any* questions to *any* of the points above, just **submit and ask**! This checklist is here to *help* you, not to deter you from contributing!

*PR template inspired by the attrs project.*


codecov[bot]: # [Codecov](https://codecov.io/gh/falconry/falcon/pull/1766?src=pr&el=h1) Report
> Merging [#1766](https://codecov.io/gh/falconry/falcon/pull/1766?src=pr&el=desc) into [master](https://codecov.io/gh/falconry/falcon/commit/9b53d2bfda0d1d5b55200b052ab109948892f43e?el=desc) will **decrease** coverage by `0.15%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/falconry/falcon/pull/1766/graphs/tree.svg?width=650&height=150&src=pr&token=zOtkfNQ3h6)](https://codecov.io/gh/falconry/falcon/pull/1766?src=pr&el=tree)

```diff
@@             Coverage Diff             @@
##            master    #1766      +/-   ##
===========================================
- Coverage   100.00%   99.84%   -0.16%     
===========================================
  Files           52       53       +1     
  Lines         4295     4436     +141     
  Branches       690      699       +9     
===========================================
+ Hits          4295     4429     +134     
- Misses           0        6       +6     
- Partials         0        1       +1     
```


| [Impacted Files](https://codecov.io/gh/falconry/falcon/pull/1766?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [falcon/errors.py](https://codecov.io/gh/falconry/falcon/pull/1766/diff?src=pr&el=tree#diff-ZmFsY29uL2Vycm9ycy5weQ==) | `100.00% <ø> (ø)` | |
| [falcon/asgi/app.py](https://codecov.io/gh/falconry/falcon/pull/1766/diff?src=pr&el=tree#diff-ZmFsY29uL2FzZ2kvYXBwLnB5) | `100.00% <100.00%> (ø)` | |
| [falcon/asgi\_spec.py](https://codecov.io/gh/falconry/falcon/pull/1766/diff?src=pr&el=tree#diff-ZmFsY29uL2FzZ2lfc3BlYy5weQ==) | `100.00% <100.00%> (ø)` | |
| [falcon/testing/client.py](https://codecov.io/gh/falconry/falcon/pull/1766/diff?src=pr&el=tree#diff-ZmFsY29uL3Rlc3RpbmcvY2xpZW50LnB5) | `100.00% <100.00%> (ø)` | |
| [falcon/testing/helpers.py](https://codecov.io/gh/falconry/falcon/pull/1766/diff?src=pr&el=tree#diff-ZmFsY29uL3Rlc3RpbmcvaGVscGVycy5weQ==) | `100.00% <100.00%> (ø)` | |
| [falcon/util/misc.py](https://codecov.io/gh/falconry/falcon/pull/1766/diff?src=pr&el=tree#diff-ZmFsY29uL3V0aWwvbWlzYy5weQ==) | `96.26% <100.00%> (-3.74%)` | :arrow_down: |
| [falcon/media/multipart.py](https://codecov.io/gh/falconry/falcon/pull/1766/diff?src=pr&el=tree#diff-ZmFsY29uL21lZGlhL211bHRpcGFydC5weQ==) | `98.81% <0.00%> (-1.19%)` | :arrow_down: |
| [falcon/util/reader.py](https://codecov.io/gh/falconry/falcon/pull/1766/diff?src=pr&el=tree#diff-ZmFsY29uL3V0aWwvcmVhZGVyLnB5) | `100.00% <0.00%> (ø)` | |
| [falcon/request\_helpers.py](https://codecov.io/gh/falconry/falcon/pull/1766/diff?src=pr&el=tree#diff-ZmFsY29uL3JlcXVlc3RfaGVscGVycy5weQ==) | `100.00% <0.00%> (ø)` | |
| ... and [1 more](https://codecov.io/gh/falconry/falcon/pull/1766/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/falconry/falcon/pull/1766?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/falconry/falcon/pull/1766?src=pr&el=footer). Last update [9b53d2b...69873e7](https://codecov.io/gh/falconry/falcon/pull/1766?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

vytas7: It seems that `typing.ContextManager` was introduced into the stdlib only in 3.7+, effectively making all py35, py36 and pypy environments fail.

Patch:
diff --git a/docs/_static/custom.css b/docs/_static/custom.css
index c79b4b8..e674443 100644
--- a/docs/_static/custom.css
+++ b/docs/_static/custom.css
@@ -176,11 +176,6 @@ pre {
     float: right;
 }
 
-
-div.admonition {
-    padding-bottom: 0;
-}
-
 div input[type="text"] {
     padding: 5px;
     margin: 0 0 0.5em 0;
@@ -268,6 +263,10 @@ dl.attribute > dd > p {
     margin-bottom: 1em;
 }
 
+dl.function, dl.method, dl.attribute, dl.class {
+    padding-top: 1em;
+}
+
 div.body div.section > ul.simple > li > dl.simple > dd > ul {
     margin: 0;
 }
@@ -345,3 +344,4 @@ div.highlight > pre {
 .sphinx-tab pre {
     border: none !important;
 }
+
diff --git a/docs/api/util.rst b/docs/api/util.rst
index 3eda5a0..6930c08 100644
--- a/docs/api/util.rst
+++ b/docs/api/util.rst
@@ -35,7 +35,21 @@ HTTP Status
 
 Async
 -----
-.. autofunction:: falcon.get_loop
+
+Aliases
+~~~~~~~
+
+These functions provide simple aliases for those implemented in :any:`asyncio`, with
+fallbacks for older versions of Python.
+
+.. autofunction:: falcon.get_running_loop
+.. autofunction:: falcon.create_task
+
+Adapters
+~~~~~~~~
+
+These functions help traverse the barrier between sync and async code.
+
 .. autofunction:: falcon.sync_to_async
 .. autofunction:: falcon.wrap_sync_to_async
 .. autofunction:: falcon.wrap_sync_to_async_unsafe
diff --git a/falcon/asgi/app.py b/falcon/asgi/app.py
index d6fa7b5..6f82496 100644
--- a/falcon/asgi/app.py
+++ b/falcon/asgi/app.py
@@ -14,18 +14,20 @@
 
 """ASGI application class."""
 
+import asyncio
 from inspect import isasyncgenfunction, iscoroutinefunction
 import traceback
 
 import falcon.app
 from falcon.app_helpers import prepare_middleware
+from falcon.asgi_spec import EventType
 from falcon.errors import CompatibilityError, UnsupportedError, UnsupportedScopeError
 from falcon.http_error import HTTPError
 from falcon.http_status import HTTPStatus
 from falcon.media.multipart import MultipartFormHandler
 import falcon.routing
 from falcon.util.misc import http_status_to_code, is_python_func
-from falcon.util.sync import _wrap_non_coroutine_unsafe, get_loop
+from falcon.util.sync import _wrap_non_coroutine_unsafe, get_running_loop
 from .multipart import MultipartForm
 from .request import Request
 from .response import Response
@@ -37,7 +39,7 @@ MultipartFormHandler._ASGI_MULTIPART_FORM = MultipartForm  # type: ignore
 __all__ = ['App']
 
 
-_EVT_RESP_EOF = {'type': 'http.response.body'}
+_EVT_RESP_EOF = {'type': EventType.HTTP_RESPONSE_BODY}
 
 _BODILESS_STATUS_CODES = frozenset([
     100,
@@ -442,7 +444,7 @@ class App(falcon.app.App):
                 resp._headers['content-length'] = str(len(data)) if data else '0'
 
             await send({
-                'type': 'http.response.start',
+                'type': EventType.HTTP_RESPONSE_START,
                 'status': resp_status,
                 'headers': resp._asgi_headers(default_media_type)
             })
@@ -460,8 +462,19 @@ class App(falcon.app.App):
                     'the result to Response.sse, e.g.: resp.sse = some_asyncgen_function()'
                 )
 
+            # NOTE(kgriffs): This must be done in a separate task because
+            #   receive() can block for some time (until the connection is
+            #   actually closed).
+            async def watch_disconnect():
+                while True:
+                    received_event = await receive()
+                    if received_event['type'] == EventType.HTTP_DISCONNECT:
+                        break
+
+            watcher = falcon.create_task(watch_disconnect())
+
             await send({
-                'type': 'http.response.start',
+                'type': EventType.HTTP_RESPONSE_START,
                 'status': resp_status,
                 'headers': resp._asgi_headers('text/event-stream')
             })
@@ -474,13 +487,25 @@ class App(falcon.app.App):
                 if not event:
                     event = SSEvent()
 
+                # NOTE(kgriffs): According to the ASGI spec, once the client
+                #   disconnects, send() acts as a no-op. We have to check
+                #   the connection state using watch_disconnect() above.
                 await send({
-                    'type': 'http.response.body',
+                    'type': EventType.HTTP_RESPONSE_BODY,
                     'body': event.serialize(),
                     'more_body': True
                 })
 
-            await send({'type': 'http.response.body'})
+                if watcher.done():
+                    break
+
+            watcher.cancel()
+            try:
+                await watcher
+            except asyncio.CancelledError:
+                pass
+
+            await send({'type': EventType.HTTP_RESPONSE_BODY})
             return
 
         if data is not None:
@@ -494,13 +519,13 @@ class App(falcon.app.App):
             resp._headers['content-length'] = str(len(data))
 
             await send({
-                'type': 'http.response.start',
+                'type': EventType.HTTP_RESPONSE_START,
                 'status': resp_status,
                 'headers': resp._asgi_headers(default_media_type)
             })
 
             await send({
-                'type': 'http.response.body',
+                'type': EventType.HTTP_RESPONSE_BODY,
                 'body': data
             })
 
@@ -512,7 +537,7 @@ class App(falcon.app.App):
             resp._headers['content-length'] = '0'
 
         await send({
-            'type': 'http.response.start',
+            'type': EventType.HTTP_RESPONSE_START,
             'status': resp_status,
             'headers': resp._asgi_headers(default_media_type)
         })
@@ -532,7 +557,7 @@ class App(falcon.app.App):
                         break
                     else:
                         await send({
-                            'type': 'http.response.body',
+                            'type': EventType.HTTP_RESPONSE_BODY,
 
                             # NOTE(kgriffs): Handle the case in which data == None
                             'body': data or b'',
@@ -552,7 +577,7 @@ class App(falcon.app.App):
                             break
 
                         await send({
-                            'type': 'http.response.body',
+                            'type': EventType.HTTP_RESPONSE_BODY,
                             'body': data,
                             'more_body': True
                         })
@@ -702,7 +727,7 @@ class App(falcon.app.App):
         if not callbacks:
             return
 
-        loop = get_loop()
+        loop = get_running_loop()
 
         for cb, is_async in callbacks:
             if is_async:
@@ -720,12 +745,12 @@ class App(falcon.app.App):
                             await handler.process_startup(scope, event)
                         except Exception:
                             await send({
-                                'type': 'lifespan.startup.failed',
+                                'type': EventType.LIFESPAN_STARTUP_FAILED,
                                 'message': traceback.format_exc(),
                             })
                             return
 
-                await send({'type': 'lifespan.startup.complete'})
+                await send({'type': EventType.LIFESPAN_STARTUP_COMPLETE})
 
             elif event['type'] == 'lifespan.shutdown':
                 for handler in reversed(self._unprepared_middleware):
@@ -734,12 +759,12 @@ class App(falcon.app.App):
                             await handler.process_shutdown(scope, event)
                         except Exception:
                             await send({
-                                'type': 'lifespan.shutdown.failed',
+                                'type': EventType.LIFESPAN_SHUTDOWN_FAILED,
                                 'message': traceback.format_exc(),
                             })
                             return
 
-                await send({'type': 'lifespan.shutdown.complete'})
+                await send({'type': EventType.LIFESPAN_SHUTDOWN_COMPLETE})
                 return
 
     def _prepare_middleware(self, middleware=None, independent_middleware=False):
diff --git a/falcon/asgi_spec.py b/falcon/asgi_spec.py
new file mode 100644
index 0000000..2ff075b
--- /dev/null
+++ b/falcon/asgi_spec.py
@@ -0,0 +1,35 @@
+# Copyright 2020 by Kurt Griffiths
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Constants, etc. defined by the ASGI specification."""
+
+
+class EventType:
+    HTTP_REQUEST = 'http.request'
+    HTTP_RESPONSE_START = 'http.response.start'
+    HTTP_RESPONSE_BODY = 'http.response.body'
+    HTTP_DISCONNECT = 'http.disconnect'
+
+    LIFESPAN_STARTUP = 'lifespan.startup'
+    LIFESPAN_STARTUP_COMPLETE = 'lifespan.startup.complete'
+    LIFESPAN_STARTUP_FAILED = 'lifespan.startup.failed'
+    LIFESPAN_SHUTDOWN = 'lifespan.shutdown'
+    LIFESPAN_SHUTDOWN_COMPLETE = 'lifespan.shutdown.complete'
+    LIFESPAN_SHUTDOWN_FAILED = 'lifespan.shutdown.failed'
+
+
+class ScopeType:
+    HTTP = 'http'
+    WS = 'websocket'
+    LIFESPAN = 'lifespan'
diff --git a/falcon/errors.py b/falcon/errors.py
index abc26ae..df03740 100644
--- a/falcon/errors.py
+++ b/falcon/errors.py
@@ -48,7 +48,7 @@ class HeaderNotSupported(ValueError):
 
 
 class CompatibilityError(ValueError):
-    """The given method or value is not compatibile."""
+    """The given method, value, or type is not compatible."""
 
 
 class UnsupportedScopeError(RuntimeError):
diff --git a/falcon/routing/static.py b/falcon/routing/static.py
index b5fa12e..3917a26 100644
--- a/falcon/routing/static.py
+++ b/falcon/routing/static.py
@@ -4,7 +4,7 @@ import os
 import re
 
 import falcon
-from falcon.util.sync import get_loop
+from falcon.util.sync import get_running_loop
 
 
 class StaticRoute:
@@ -143,7 +143,7 @@ class _AsyncFileReader:
 
     def __init__(self, file):
         self._file = file
-        self._loop = get_loop()
+        self._loop = get_running_loop()
 
     async def read(self, size=-1):
         return await self._loop.run_in_executor(None, partial(self._file.read, size))
diff --git a/falcon/util/misc.py b/falcon/util/misc.py
index 0f1b585..e5d8e32 100644
--- a/falcon/util/misc.py
+++ b/falcon/util/misc.py
@@ -450,9 +450,9 @@ def code_to_http_status(status):
     try:
         code = int(status)
         if not 100 <= code <= 999:
-            raise ValueError('{} is not a valid status code'.format(code))
+            raise ValueError('{} is not a valid status code'.format(status))
     except (ValueError, TypeError):
-        raise ValueError('{!r} is not a valid status code'.format(code))
+        raise ValueError('{!r} is not a valid status code'.format(status))
 
     try:
         # NOTE(kgriffs): We do this instead of using http.HTTPStatus since
diff --git a/falcon/util/sync.py b/falcon/util/sync.py
index d63bc5b..059f4a1 100644
--- a/falcon/util/sync.py
+++ b/falcon/util/sync.py
@@ -7,7 +7,8 @@ from typing import Callable
 
 
 __all__ = [
-    'get_loop',
+    'create_task',
+    'get_running_loop',
     'invoke_coroutine_sync',
     'runs_sync',
     'sync_to_async',
@@ -20,15 +21,24 @@ _one_thread_to_rule_them_all = ThreadPoolExecutor(max_workers=1)
 
 
 try:
-    get_loop = asyncio.get_running_loop
-    """Gets the running asyncio event loop."""
+    get_running_loop = asyncio.get_running_loop
 except AttributeError:  # pragma: nocover
     # NOTE(kgriffs): This branch is definitely covered under py35 and py36
     #   but for some reason the codecov gate doesn't pick this up, hence
     #   the pragma above.
 
-    get_loop = asyncio.get_event_loop
-    """Gets the running asyncio event loop."""
+    get_running_loop = asyncio.get_event_loop
+
+
+try:
+    create_task = asyncio.create_task
+except AttributeError:  # pragma: nocover
+    # NOTE(kgriffs): This branch is definitely covered under py35 and py36
+    #   but for some reason the codecov gate doesn't pick this up, hence
+    #   the pragma above.
+
+    def create_task(coro, name=None):
+        return asyncio.ensure_future(coro)
 
 
 def wrap_sync_to_async_unsafe(func) -> Callable:
@@ -102,7 +112,7 @@ def wrap_sync_to_async(func, threadsafe=None) -> Callable:
 
     @wraps(func)
     async def wrapper(*args, **kwargs):
-        return await get_loop().run_in_executor(executor, partial(func, *args, **kwargs))
+        return await get_running_loop().run_in_executor(executor, partial(func, *args, **kwargs))
 
     return wrapper
 
@@ -145,7 +155,7 @@ async def sync_to_async(func, *args, **kwargs):
         synchronous callable.
     """
 
-    return await get_loop().run_in_executor(None, partial(func, *args, **kwargs))
+    return await get_running_loop().run_in_executor(None, partial(func, *args, **kwargs))
 
 
 def _should_wrap_non_coroutines() -> bool:


Test patch:
diff --git a/docs/api/testing.rst b/docs/api/testing.rst
index d2b7ccb..7619a96 100644
--- a/docs/api/testing.rst
+++ b/docs/api/testing.rst
@@ -8,25 +8,28 @@ Testing Helpers
 .. automodule:: falcon.testing
     :noindex:
 
-.. TODO: Add TOC here, and to other large RSTs?
-
-General Helpers
----------------
-
-.. autoclass:: TestCase
-.. autoclass:: SimpleTestResource
-.. autofunction:: rand_string
-.. autofunction:: get_unused_port
-.. autofunction:: redirected
-.. autofunction:: get_encoding_from_headers
-
-
 Simulating Requests
 -------------------
 
 Main Interface
 ~~~~~~~~~~~~~~
 
+.. autoclass:: TestClient
+   :members:
+.. autoclass:: ASGIConductor
+   :members:
+.. autoclass:: Result
+   :members:
+.. autoclass:: StreamedResult
+   :members:
+.. autoclass:: ResultBodyStream
+   :members:
+.. autoclass:: Cookie
+   :members:
+
+Standalone Methods
+~~~~~~~~~~~~~~~~~~
+
 .. autofunction:: simulate_get
 .. autofunction:: simulate_head
 .. autofunction:: simulate_post
@@ -36,9 +39,6 @@ Main Interface
 .. autofunction:: simulate_delete
 .. autofunction:: simulate_request
 
-.. autoclass:: Result
-.. autoclass:: Cookie
-.. autoclass:: TestClient
 .. autofunction:: capture_responder_args
 .. autofunction:: capture_responder_args_async
 .. autofunction:: set_resp_defaults
@@ -48,11 +48,34 @@ Low-Level Utils
 ~~~~~~~~~~~~~~~
 
 .. autoclass:: StartResponseMock
+   :members:
 .. autoclass:: ASGIRequestEventEmitter
+   :members:
 .. autoclass:: ASGILifespanEventEmitter
+   :members:
 .. autoclass:: ASGIResponseEventCollector
+   :members:
 .. autofunction:: create_environ
 .. autofunction:: create_scope
 .. autofunction:: create_req
 .. autofunction:: create_asgi_req
 .. autofunction:: closed_wsgi_iterable
+
+Other Helpers
+-------------
+
+Test Cases
+~~~~~~~~~~
+
+.. autoclass:: TestCase
+   :members:
+.. autoclass:: SimpleTestResource
+   :members:
+
+Functions
+~~~~~~~~~
+
+.. autofunction:: rand_string
+.. autofunction:: get_unused_port
+.. autofunction:: redirected
+.. autofunction:: get_encoding_from_headers
diff --git a/falcon/testing/client.py b/falcon/testing/client.py
index c4a2f6b..1c0059f 100644
--- a/falcon/testing/client.py
+++ b/falcon/testing/client.py
@@ -22,17 +22,20 @@ import asyncio
 import datetime as dt
 import inspect
 import time
-from typing import Dict, Optional, Union
+from typing import Dict, Optional, Sequence, Union
 import warnings
 import wsgiref.validate
 
+from falcon.asgi_spec import ScopeType
 from falcon.constants import COMBINED_METHODS, MEDIA_JSON
+from falcon.errors import CompatibilityError
 from falcon.testing import helpers
 from falcon.testing.srmock import StartResponseMock
 from falcon.util import (
     CaseInsensitiveDict,
     code_to_http_status,
-    get_loop,
+    create_task,
+    get_running_loop,
     http_cookies,
     http_date_to_dt,
     invoke_coroutine_sync,
@@ -134,12 +137,10 @@ class Cookie:
         return self._samesite if self._samesite else None  # type: ignore[attr-defined]
 
 
-class Result:
-    """Encapsulates the result of a simulated request.
+class _ResultBase:
+    """Base class for the result of a simulated request.
 
     Args:
-        iterable (iterable): An iterable that yields zero or more
-            bytestrings, per PEP-3333
         status (str): An HTTP status string, including status code and
             reason string
         headers (list): A list of (header_name, header_value) tuples,
@@ -171,25 +172,12 @@ class Result:
 
         encoding (str): Text encoding of the response body, or ``None``
             if the encoding can not be determined.
-        content (bytes): Raw response body, or ``bytes`` if the
-            response body was empty.
-        text (str): Decoded response body of type ``str``.
-            If the content type does not specify an encoding, UTF-8 is
-            assumed.
-        json (JSON serializable): Deserialized JSON body. Will be ``None`` if
-            the body has no content to deserialize. Otherwise, raises an error
-            if the response is not valid JSON.
     """
 
-    def __init__(self, iterable, status, headers, events=None):
-        self._text = None
-
-        self._content = b''.join(iterable)
-
+    def __init__(self, status, headers):
         self._status = status
         self._status_code = int(status[:3])
         self._headers = CaseInsensitiveDict(headers)
-        self._events = events or []
 
         cookies = http_cookies.SimpleCookie()
         for name, value in headers:
@@ -230,6 +218,90 @@ class Result:
     def encoding(self) -> str:
         return self._encoding
 
+
+class ResultBodyStream:
+    """Simple forward-only reader for a streamed test result body.
+
+    Args:
+        chunks (list): Reference to a list of body chunks that may
+            continue to be appended to as more body events are
+            collected.
+    """
+
+    def __init__(self, chunks: Sequence[bytes]):
+        self._chunks = chunks
+        self._chunk_pos = 0
+
+    async def read(self) -> bytes:
+        """Read any data that has been collected since the last call.
+
+        Returns:
+            bytes: data that has been collected since the last call,
+            or an empty byte string if no additional data is available.
+        """
+
+        # NOTE(kgriffs): Yield to other tasks to give them a chance to
+        #   send us more body chunks if any are available.
+        #
+        #   https://bugs.python.org/issue34476
+        #
+        await asyncio.sleep(0)
+
+        if self._chunk_pos >= len(self._chunks):
+            return b''
+
+        data = b''.join(self._chunks[self._chunk_pos:])
+        self._chunk_pos = len(self._chunks)
+
+        return data
+
+
+class Result(_ResultBase):
+    """Encapsulates the result of a simulated request.
+
+    Args:
+        iterable (iterable): An iterable that yields zero or more
+            bytestrings, per PEP-3333
+        status (str): An HTTP status string, including status code and
+            reason string
+        headers (list): A list of (header_name, header_value) tuples,
+            per PEP-3333
+
+    Attributes:
+        status (str): HTTP status string given in the response
+        status_code (int): The code portion of the HTTP status string
+        headers (CaseInsensitiveDict): A case-insensitive dictionary
+            containing all the headers in the response, except for
+            cookies, which may be accessed via the `cookies`
+            attribute.
+
+            Note:
+
+                Multiple instances of a header in the response are
+                currently not supported; it is unspecified which value
+                will "win" and be represented in `headers`.
+
+        cookies (dict): A dictionary of
+            :py:class:`falcon.testing.Cookie` values parsed from the
+            response, by name.
+        encoding (str): Text encoding of the response body, or ``None``
+            if the encoding can not be determined.
+        content (bytes): Raw response body, or ``bytes`` if the
+            response body was empty.
+        text (str): Decoded response body of type ``str``.
+            If the content type does not specify an encoding, UTF-8 is
+            assumed.
+        json (JSON serializable): Deserialized JSON body. Will be ``None`` if
+            the body has no content to deserialize. Otherwise, raises an error
+            if the response is not valid JSON.
+    """
+
+    def __init__(self, iterable, status, headers):
+        super().__init__(status, headers)
+
+        self._text = None
+        self._content = b''.join(iterable)
+
     @property
     def content(self) -> bytes:
         return self._content
@@ -257,6 +329,69 @@ class Result:
         return util_json.loads(self.text)
 
 
+class StreamedResult(_ResultBase):
+    """Encapsulates the streamed result of an ASGI request.
+
+    Args:
+        body_chunks (list): A list of body chunks. This list may be
+            appended to after a result object has been instantiated.
+        status (str): An HTTP status string, including status code and
+            reason string
+        headers (list): A list of (header_name, header_value) tuples,
+            per PEP-3333
+        task (asyncio.Task): The scheduled simulated request which may or
+            may not have already finished. :meth:`~.finalize`
+            will await the task before returning.
+        req_event_emitter (~falcon.testing.ASGIRequestEventEmitter): A reference
+            to the event emitter used to simulate events sent to the ASGI
+            application via its receive() method.
+            :meth:`~.finalize` will cause the event emitter to
+            simulate an 'http.disconnect' event before returning.
+
+    Attributes:
+        status (str): HTTP status string given in the response
+        status_code (int): The code portion of the HTTP status string
+        headers (CaseInsensitiveDict): A case-insensitive dictionary
+            containing all the headers in the response, except for
+            cookies, which may be accessed via the `cookies`
+            attribute.
+
+            Note:
+
+                Multiple instances of a header in the response are
+                currently not supported; it is unspecified which value
+                will "win" and be represented in `headers`.
+
+        cookies (dict): A dictionary of
+            :py:class:`falcon.testing.Cookie` values parsed from the
+            response, by name.
+        encoding (str): Text encoding of the response body, or ``None``
+            if the encoding can not be determined.
+        stream (ResultStream): Raw response body, as a byte stream.
+    """
+
+    def __init__(self, body_chunks, status, headers, task, req_event_emitter):
+        super().__init__(status, headers)
+
+        self._task = task
+        self._stream = ResultBodyStream(body_chunks)
+        self._req_event_emitter = req_event_emitter
+
+    @property
+    def stream(self) -> ResultBodyStream:
+        return self._stream
+
+    async def finalize(self):
+        """Finalize the encapsulated simulated request.
+
+        This method causes the request event emitter to begin emitting
+        'http.disconnect' events and then awaits the completion of the
+        asyncio task that is running the simulated ASGI request.
+        """
+        self._req_event_emitter.disconnect()
+        await self._task
+
+
 # NOTE(kgriffs): The default of asgi_disconnect_ttl was chosen to be
 #   relatively long (5 minutes) to help testers notice when something
 #   appears to be "hanging", which might indicates that the app is
@@ -267,13 +402,21 @@ def simulate_request(app, method='GET', path='/', query_string=None,
                      params_csv=False, protocol='http', host=helpers.DEFAULT_HOST,
                      remote_addr=None, extras=None, http_version='1.1',
                      port=None, root_path=None, cookies=None, asgi_chunk_size=4096,
-                     asgi_disconnect_ttl=300) -> Result:
+                     asgi_disconnect_ttl=300) -> _ResultBase:
 
     """Simulates a request to a WSGI or ASGI application.
 
     Performs a request against a WSGI or ASGI application. In the case of
     WSGI, uses :any:`wsgiref.validate` to ensure the response is valid.
 
+    Note:
+        In the case of an ASGI request, this method will simulate the entire
+        app lifecycle in a single shot, including lifespan and client
+        disconnect events. In order to simulate multiple interleaved
+        requests, or to test a streaming endpoint (such as one that emits
+        server-sent events), :class:`~falcon.testing.ASGIConductor` can be
+        used to more precisely control the app lifecycle.
+
     Keyword Args:
         app (callable): The WSGI or ASGI application to call
         method (str): An HTTP method to use in the request
@@ -366,96 +509,171 @@ def simulate_request(app, method='GET', path='/', query_string=None,
         :py:class:`~.Result`: The result of the request
     """
 
-    if not path.startswith('/'):
-        raise ValueError("path must start with '/'")
+    if _is_asgi_app(app):
+        return invoke_coroutine_sync(
+            _simulate_request_asgi,
 
-    if '?' in path:
-        if query_string or params:
-            raise ValueError(
-                'path may not contain a query string in combination with '
-                'the query_string or params parameters. Please use only one '
-                'way of specifying the query string.'
-            )
-        path, query_string = path.split('?', 1)
-    elif query_string and query_string.startswith('?'):
-        raise ValueError("query_string should not start with '?'")
-
-    extras = extras or {}
+            app,
 
-    if query_string is None:
-        query_string = to_query_str(
-            params,
-            comma_delimited_lists=params_csv,
-            prefix=False,
+            method=method, path=path, query_string=query_string,
+            headers=headers, content_type=content_type, body=body, json=json,
+            params=params, params_csv=params_csv, protocol=protocol, host=host,
+            remote_addr=remote_addr, extras=extras, http_version=http_version,
+            port=port, root_path=root_path, asgi_chunk_size=asgi_chunk_size,
+            asgi_disconnect_ttl=asgi_disconnect_ttl,
         )
 
-    if content_type is not None:
-        headers = headers or {}
-        headers['Content-Type'] = content_type
+    path, query_string, headers, body, extras = _prepare_sim_args(
+        path, query_string, params, params_csv, content_type, headers, body, json, extras)
 
-    if json is not None:
-        body = util_json.dumps(json, ensure_ascii=False)
-        headers = headers or {}
-        headers['Content-Type'] = MEDIA_JSON
+    env = helpers.create_environ(
+        method=method,
+        scheme=protocol,
+        path=path,
+        query_string=(query_string or ''),
+        headers=headers,
+        body=body,
+        file_wrapper=file_wrapper,
+        host=host,
+        remote_addr=remote_addr,
+        wsgierrors=wsgierrors,
+        http_version=http_version,
+        port=port,
+        root_path=root_path,
+        cookies=cookies,
+    )
 
-    if not _is_asgi_app(app):
-        env = helpers.create_environ(
-            method=method,
-            scheme=protocol,
-            path=path,
-            query_string=(query_string or ''),
-            headers=headers,
-            body=body,
-            file_wrapper=file_wrapper,
-            host=host,
-            remote_addr=remote_addr,
-            wsgierrors=wsgierrors,
-            http_version=http_version,
-            port=port,
-            root_path=root_path,
-            cookies=cookies,
+    if 'REQUEST_METHOD' in extras and extras['REQUEST_METHOD'] != method:
+        # NOTE(vytas): Even given the duct tape nature of overriding
+        # arbitrary environ variables, changing the method can potentially
+        # be very confusing, particularly when using specialized
+        # simulate_get/post/patch etc methods.
+        raise ValueError(
+            'WSGI environ extras may not override the request method. '
+            'Please use the method parameter.'
         )
 
-        if 'REQUEST_METHOD' in extras and extras['REQUEST_METHOD'] != method:
-            # NOTE(vytas): Even given the duct tape nature of overriding
-            # arbitrary environ variables, changing the method can potentially
-            # be very confusing, particularly when using specialized
-            # simulate_get/post/patch etc methods.
-            raise ValueError(
-                'WSGI environ extras may not override the request method. '
-                'Please use the method parameter.'
-            )
+    env.update(extras)
 
-        env.update(extras)
+    srmock = StartResponseMock()
+    validator = wsgiref.validate.validator(app)
 
-        srmock = StartResponseMock()
-        validator = wsgiref.validate.validator(app)
+    iterable = validator(env, srmock)
 
-        iterable = validator(env, srmock)
+    return Result(helpers.closed_wsgi_iterable(iterable),
+                  srmock.status, srmock.headers)
 
-        return Result(helpers.closed_wsgi_iterable(iterable),
-                      srmock.status, srmock.headers)
 
-    # ---------------------------------------------------------------------
-    # NOTE(kgriffs): 'lifespan' scope
-    # ---------------------------------------------------------------------
+# NOTE(kgriffs): The default of asgi_disconnect_ttl was chosen to be
+#   relatively long (5 minutes) to help testers notice when something
+#   appears to be "hanging", which might indicates that the app is
+#   not handling the reception of events correctly.
+async def _simulate_request_asgi(
+    app, method='GET', path='/', query_string=None,
+    headers=None, content_type=None, body=None, json=None, params=None,
+    params_csv=True, protocol='http', host=helpers.DEFAULT_HOST,
+    remote_addr=None, extras=None, http_version='1.1',
+    port=None, root_path=None, asgi_chunk_size=4096,
+    asgi_disconnect_ttl=300,
 
-    lifespan_scope = {
-        'type': 'lifespan',
-        'asgi': {
-            'version': '3.0',
-            'spec_version': '2.0',
-        },
-    }
+    # NOTE(kgriffs): These are undocumented because they are only
+    #   meant to be used internally by the framework (i.e., they are
+    #   not part of the public interface.) In case we ever expose
+    #   simulate_request_asgi() as part of the public interface, we
+    #   don't want these kwargs to be documented.
+    _one_shot=True, _stream_result=False,
 
-    shutting_down = asyncio.Condition()
-    lifespan_event_emitter = helpers.ASGILifespanEventEmitter(shutting_down)
-    lifespan_event_collector = helpers.ASGIResponseEventCollector()
+) -> _ResultBase:
+
+    """Simulates a request to an ASGI application.
+
+    Keyword Args:
+        app (callable): The WSGI or ASGI application to call
+        method (str): An HTTP method to use in the request
+            (default: 'GET')
+        path (str): The URL path to request (default: '/').
+
+            Note:
+                The path may contain a query string. However, neither
+                `query_string` nor `params` may be specified in this case.
+
+        root_path (str): The initial portion of the request URL's "path" that
+            corresponds to the application object, so that the application
+            knows its virtual "location". This defaults to the empty string,
+            indicating that the application corresponds to the "root" of the
+            server.
+        protocol: The protocol to use for the URL scheme
+            (default: 'http')
+        port (int): The TCP port to simulate. Defaults to
+            the standard port used by the given scheme (i.e., 80 for 'http'
+            and 443 for 'https'). A string may also be passed, as long as
+            it can be parsed as an int.
+        params (dict): A dictionary of query string parameters,
+            where each key is a parameter name, and each value is
+            either a ``str`` or something that can be converted
+            into a ``str``, or a list of such values. If a ``list``,
+            the value will be converted to a comma-delimited string
+            of values (e.g., 'thing=1,2,3').
+        params_csv (bool): Set to ``False`` to encode list values
+            in query string params by specifying multiple instances
+            of the parameter (e.g., 'thing=1&thing=2&thing=3').
+            Otherwise, parameters will be encoded as comma-separated
+            values (e.g., 'thing=1,2,3'). Defaults to ``True``.
+        query_string (str): A raw query string to include in the
+            request (default: ``None``). If specified, overrides
+            `params`.
+        content_type (str): The value to use for the Content-Type header in
+            the request. If specified, this value will take precedence over
+            any value set for the Content-Type header in the
+            `headers` keyword argument. The ``falcon`` module provides a number
+            of :ref:`constants for common media types <media_type_constants>`.
+        headers (dict): Extra headers as a dict-like (Mapping) object, or an
+            iterable yielding a series of two-member (*name*, *value*)
+            iterables. Each pair of strings provides the name and value
+            for an HTTP header. If desired, multiple header values may be
+            combined into a single (*name*, *value*) pair by joining the values
+            with a comma when the header in question supports the list
+            format (see also RFC 7230 and RFC 7231). Header names are not
+            case-sensitive.
+        body (str): The body of the request (default ''). The value will be
+            encoded as UTF-8 in the WSGI environ. Alternatively, a byte string
+            may be passed, in which case it will be used as-is.
+        json(JSON serializable): A JSON document to serialize as the
+            body of the request (default: ``None``). If specified,
+            overrides `body` and sets the Content-Type header to
+            ``'application/json'``, overriding any value specified by either
+            the `content_type` or `headers` arguments.
+        host(str): A string to use for the hostname part of the fully
+            qualified request URL (default: 'falconframework.org')
+        remote_addr (str): A string to use as the remote IP address for the
+            request (default: '127.0.0.1'). For WSGI, this corresponds to
+            the 'REMOTE_ADDR' environ variable. For ASGI, this corresponds
+            to the IP address used for the 'client' field in the connection
+            scope.
+        http_version (str): The HTTP version to simulate. Must be either
+            '2', '2.0', 1.1', '1.0', or '1' (default '1.1'). If set to '1.0',
+            the Host header will not be added to the scope.
+        asgi_chunk_size (int): The maximum number of bytes that will be
+            sent to the ASGI app in a single 'http.request' event (default
+            4096).
+        asgi_disconnect_ttl (int): The maximum number of seconds to wait
+            since the request was initiated, before emitting an
+            'http.disconnect' event when the app calls the
+            receive() function (default 300).
+        extras (dict): Additional values to add to the WSGI
+            ``environ`` dictionary or the ASGI scope for the request
+            (default: ``None``)
+
+    Returns:
+        :py:class:`~.Result`: The result of the request
+    """
+
+    path, query_string, headers, body, extras = _prepare_sim_args(
+        path, query_string, params, params_csv, content_type, headers, body, json, extras)
 
     # ---------------------------------------------------------------------
     # NOTE(kgriffs): 'http' scope
     # ---------------------------------------------------------------------
-
     content_length = None
 
     if body is not None:
@@ -485,27 +703,68 @@ def simulate_request(app, method='GET', path='/', query_string=None,
         )
 
     http_scope.update(extras)
+    # ---------------------------------------------------------------------
 
     disconnect_at = time.time() + max(0, asgi_disconnect_ttl)
     req_event_emitter = helpers.ASGIRequestEventEmitter(
         (body or b''),
-        disconnect_at,
-        chunk_size=asgi_chunk_size
+        chunk_size=asgi_chunk_size,
+        disconnect_at=disconnect_at,
     )
 
     resp_event_collector = helpers.ASGIResponseEventCollector()
 
+    if not _one_shot:
+        task_req = create_task(app(http_scope, req_event_emitter, resp_event_collector))
+
+        if _stream_result:
+            # NOTE(kgriffs): Wait until the response has been started and give
+            #   the task a chance to progress. Otherwise, we won't have a
+            #   status or headers to pass to StreamedResult.
+            while not resp_event_collector.status:
+                await asyncio.sleep(0)
+
+            return StreamedResult(resp_event_collector.body_chunks,
+                                  code_to_http_status(resp_event_collector.status),
+                                  resp_event_collector.headers,
+                                  task_req,
+                                  req_event_emitter)
+
+        req_event_emitter.disconnect()
+        await task_req
+        return Result(resp_event_collector.body_chunks,
+                      code_to_http_status(resp_event_collector.status),
+                      resp_event_collector.headers)
+
+    # ---------------------------------------------------------------------
+    # NOTE(kgriffs): 'lifespan' scope
+    # ---------------------------------------------------------------------
+    lifespan_scope = {
+        'type': ScopeType.LIFESPAN,
+        'asgi': {
+            'version': '3.0',
+            'spec_version': '2.0',
+        },
+    }
+
+    shutting_down = asyncio.Condition()
+    lifespan_event_emitter = helpers.ASGILifespanEventEmitter(shutting_down)
+    lifespan_event_collector = helpers.ASGIResponseEventCollector()
+    # ---------------------------------------------------------------------
+
     async def conductor():
         # NOTE(kgriffs): We assume this is a Falcon ASGI app, which supports
         #   the lifespan protocol and thus we do not need to catch
         #   exceptions that would signify no lifespan protocol support.
-        t = get_loop().create_task(
+        task_lifespan = get_running_loop().create_task(
             app(lifespan_scope, lifespan_event_emitter, lifespan_event_collector)
         )
 
         await _wait_for_startup(lifespan_event_collector.events)
 
-        await app(http_scope, req_event_emitter, resp_event_collector)
+        task_req = create_task(app(http_scope, req_event_emitter, resp_event_collector))
+        req_event_emitter.disconnect()
+        await task_req
 
         # NOTE(kgriffs): Notify lifespan_event_emitter that it is OK
         #   to proceed.
@@ -513,23 +772,258 @@ def simulate_request(app, method='GET', path='/', query_string=None,
             shutting_down.notify()
 
         await _wait_for_shutdown(lifespan_event_collector.events)
-        await t
+        await task_lifespan
 
-    invoke_coroutine_sync(conductor)
+    await conductor()
 
     return Result(resp_event_collector.body_chunks,
                   code_to_http_status(resp_event_collector.status),
-                  resp_event_collector.headers,
-                  events=resp_event_collector.events)
+                  resp_event_collector.headers)
+
+
+class ASGIConductor:
+    """Test conductor for ASGI apps.
+
+    This class provides more control over the lifecycle of a simulated
+    request as compared to :class:`~.TestClient`. In addition, the conductor's
+    asynchronous interface affords interleaved requests and the testing of
+    streaming protocols such as server-sent events (SSE) and WebSocket.
+
+    :class:`~.ASGIConductor` is implemented as a context manager. Upon
+    entering and exiting the context, the appropriate ASGI lifespan events
+    will be simulated.
+
+    Within the context, HTTP requests can be simulated using an interface
+    that is similar to :class:`~.TestClient`, except that all ``simulate_*()``
+    methods are coroutines::
+
+        async with testing.ASGIConductor(some_app) as conductor:
+            async def post_events():
+                for i in range(100):
+                    await conductor.simulate_post('/events', json={'id': i}):
+                    await asyncio.sleep(0.01)
+
+            async def get_events_sse():
+                # Here, we will get only some of the single server-sent events
+                # because the non-streaming method is "single-shot". In other
+                # words, simulate_get() will emit a client disconnect event
+                # into the app before returning.
+                result = await conductor.simulate_get('/events')
+
+                # Alternatively, we can use simulate_get_stream() as a context
+                # manager to perform a series of reads on the result body that
+                # are interleaved with the execution of the post_events()
+                # coroutine.
+                async with conductor.simulate_get_stream('/events') as sr:
+                    while some_condition:
+                        # Read next body chunk that was received (if any).
+                        chunk = await sr.stream.read()
+
+                        if chunk:
+                            # TODO: Do something with the chunk
+                            pass
+
+                    # Exiting the context causes the request event emitter to
+                    # begin emitting 'http.disconnect' events and then awaits
+                    # the completion of the asyncio task that is running the
+                    # simulated ASGI request.
+
+            asyncio.gather(post_events(), get_events_sse())
+
+    Note:
+        Because the :class:`~.ASGIConductor` interface uses coroutines,
+        it cannot be used directly with synchronous testing frameworks such as
+        pytest.
+
+        As a workaround, the test can be adapted by wrapping it in
+        an inline async function and then invoking it via
+        :meth:`falcon.invoke_coroutine_sync` or decorating the test function
+        with :meth:`falcon.runs_sync`.
+
+        Alternatively, you can try searching PyPI to see if an async plugin is
+        available for your testing framework of choice. For example, the
+        ``pytest-asyncio`` plugin is available for ``pytest`` users.
+
+    Args:
+        app (callable): An ASGI application to target when simulating
+            requests.
+
+    Keyword Arguments:
+        headers (dict): Default headers to set on every request (default
+            ``None``). These defaults may be overridden by passing values
+            for the same headers to one of the ``simulate_*()`` methods.
+
+    Attributes:
+        app: The app that this client instance was configured to use.
+
+    """
+    def __init__(self, app, headers=None):
+        if not _is_asgi_app(app):
+            raise CompatibilityError('ASGIConductor may only be used with an ASGI app')
+
+        self.app = app
+        self._default_headers = headers
+
+        self._shutting_down = asyncio.Condition()
+        self._lifespan_event_collector = helpers.ASGIResponseEventCollector()
+        self._lifespan_task = None
+
+    async def __aenter__(self):
+        lifespan_scope = {
+            'type': ScopeType.LIFESPAN,
+            'asgi': {
+                'version': '3.0',
+                'spec_version': '2.0',
+            },
+        }
+
+        lifespan_event_emitter = helpers.ASGILifespanEventEmitter(self._shutting_down)
+
+        # NOTE(kgriffs): We assume this is a Falcon ASGI app, which supports
+        #   the lifespan protocol and thus we do not need to catch
+        #   exceptions that would signify no lifespan protocol support.
+        self._lifespan_task = get_running_loop().create_task(
+            self.app(lifespan_scope, lifespan_event_emitter, self._lifespan_event_collector)
+        )
+
+        await _wait_for_startup(self._lifespan_event_collector.events)
+
+        return self
+
+    async def __aexit__(self, ex_type, ex, tb):
+        if ex_type:
+            return False
+
+        # NOTE(kgriffs): Notify lifespan_event_emitter that it is OK
+        #   to proceed.
+        async with self._shutting_down:
+            self._shutting_down.notify()
+
+        await _wait_for_shutdown(self._lifespan_event_collector.events)
+        await self._lifespan_task
+
+        return True
+
+    async def simulate_get(self, path='/', **kwargs) -> _ResultBase:
+        """Simulates a GET request to an ASGI application.
+
+        (See also: :py:meth:`falcon.testing.simulate_get`)
+        """
+        return await self.simulate_request('GET', path, **kwargs)
+
+    def simulate_get_stream(self, path='/', **kwargs):
+        """Simulates a GET request to an ASGI application with a streamed response.
+
+        This method returns a context manager that can be used to obtain
+        a managed :class:`~.StreamedResult` instance. Exiting the context
+        will automatically finalize the result object, causing the request
+        event emitter to begin emitting 'http.disconnect' events and then
+        await the completion of the task that is running the simulated ASGI
+        request.
+
+        In the following example, a series of streamed body chunks are read
+        from the response::
+
+            async with conductor.simulate_get_stream('/events') as sr:
+                while some_condition:
+                    # Read next body chunk that was received (if any).
+                    chunk = await sr.stream.read()
+
+                    if chunk:
+                        # TODO: Do something with the chunk. For example,
+                        #   a series of server-sent events could be validated
+                        #   by concatenating the chunks and splitting on
+                        #   double-newlines to obtain individual events.
+                        pass
+
+        """
+
+        kwargs['_stream_result'] = True
+
+        return _AsyncContextManager(self.simulate_request('GET', path, **kwargs))
+
+    async def simulate_head(self, path='/', **kwargs) -> _ResultBase:
+        """Simulates a HEAD request to an ASGI application.
+
+        (See also: :py:meth:`falcon.testing.simulate_head`)
+        """
+        return await self.simulate_request('HEAD', path, **kwargs)
 
+    async def simulate_post(self, path='/', **kwargs) -> _ResultBase:
+        """Simulates a POST request to an ASGI application.
 
-def simulate_get(app, path, **kwargs) -> Result:
+        (See also: :py:meth:`falcon.testing.simulate_post`)
+        """
+        return await self.simulate_request('POST', path, **kwargs)
+
+    async def simulate_put(self, path='/', **kwargs) -> _ResultBase:
+        """Simulates a PUT request to an ASGI application.
+
+        (See also: :py:meth:`falcon.testing.simulate_put`)
+        """
+        return await self.simulate_request('PUT', path, **kwargs)
+
+    async def simulate_options(self, path='/', **kwargs) -> _ResultBase:
+        """Simulates an OPTIONS request to an ASGI application.
+
+        (See also: :py:meth:`falcon.testing.simulate_options`)
+        """
+        return await self.simulate_request('OPTIONS', path, **kwargs)
+
+    async def simulate_patch(self, path='/', **kwargs) -> _ResultBase:
+        """Simulates a PATCH request to an ASGI application.
+
+        (See also: :py:meth:`falcon.testing.simulate_patch`)
+        """
+        return await self.simulate_request('PATCH', path, **kwargs)
+
+    async def simulate_delete(self, path='/', **kwargs) -> _ResultBase:
+        """Simulates a DELETE request to an ASGI application.
+
+        (See also: :py:meth:`falcon.testing.simulate_delete`)
+        """
+        return await self.simulate_request('DELETE', path, **kwargs)
+
+    async def simulate_request(self, *args, **kwargs) -> _ResultBase:
+        """Simulates a request to an ASGI application.
+
+        Wraps :py:meth:`falcon.testing.simulate_request` to perform a
+        WSGI request directly against ``self.app``. Equivalent to::
+
+            falcon.testing.simulate_request(self.app, *args, **kwargs)
+        """
+
+        if self._default_headers:
+            # NOTE(kgriffs): Handle the case in which headers is explicitly
+            # set to None.
+            additional_headers = kwargs.get('headers', {}) or {}
+
+            merged_headers = self._default_headers.copy()
+            merged_headers.update(additional_headers)
+
+            kwargs['headers'] = merged_headers
+
+        # NOTE(kgriffs): The conductor takes care of startup/shutdown
+        kwargs['_one_shot'] = False
+
+        return await _simulate_request_asgi(self.app, *args, **kwargs)
+
+
+def simulate_get(app, path, **kwargs) -> _ResultBase:
     """Simulates a GET request to a WSGI or ASGI application.
 
     Equivalent to::
 
          simulate_request(app, 'GET', path, **kwargs)
 
+    Note:
+        In the case of an ASGI request, this method will simulate the entire
+        app lifecycle in a single shot, including lifespan and client
+        disconnect events. In order to simulate multiple interleaved
+        requests, or to test a streaming endpoint (such as one that emits
+        server-sent events), :class:`~falcon.testing.ASGIConductor` can be
+        used to more precisely control the app lifecycle.
+
     Args:
         app (callable): The application to call
         path (str): The URL path to request
@@ -611,13 +1105,21 @@ def simulate_get(app, path, **kwargs) -> Result:
     return simulate_request(app, 'GET', path, **kwargs)
 
 
-def simulate_head(app, path, **kwargs) -> Result:
+def simulate_head(app, path, **kwargs) -> _ResultBase:
     """Simulates a HEAD request to a WSGI or ASGI application.
 
     Equivalent to::
 
          simulate_request(app, 'HEAD', path, **kwargs)
 
+    Note:
+        In the case of an ASGI request, this method will simulate the entire
+        app lifecycle in a single shot, including lifespan and client
+        disconnect events. In order to simulate multiple interleaved
+        requests, or to test a streaming endpoint (such as one that emits
+        server-sent events), :class:`~falcon.testing.ASGIConductor` can be
+        used to more precisely control the app lifecycle.
+
     Args:
         app (callable): The application to call
         path (str): The URL path to request
@@ -693,13 +1195,21 @@ def simulate_head(app, path, **kwargs) -> Result:
     return simulate_request(app, 'HEAD', path, **kwargs)
 
 
-def simulate_post(app, path, **kwargs) -> Result:
+def simulate_post(app, path, **kwargs) -> _ResultBase:
     """Simulates a POST request to a WSGI or ASGI application.
 
     Equivalent to::
 
          simulate_request(app, 'POST', path, **kwargs)
 
+    Note:
+        In the case of an ASGI request, this method will simulate the entire
+        app lifecycle in a single shot, including lifespan and client
+        disconnect events. In order to simulate multiple interleaved
+        requests, or to test a streaming endpoint (such as one that emits
+        server-sent events), :class:`~falcon.testing.ASGIConductor` can be
+        used to more precisely control the app lifecycle.
+
     Args:
         app (callable): The application to call
         path (str): The URL path to request
@@ -789,13 +1299,21 @@ def simulate_post(app, path, **kwargs) -> Result:
     return simulate_request(app, 'POST', path, **kwargs)
 
 
-def simulate_put(app, path, **kwargs) -> Result:
+def simulate_put(app, path, **kwargs) -> _ResultBase:
     """Simulates a PUT request to a WSGI or ASGI application.
 
     Equivalent to::
 
          simulate_request(app, 'PUT', path, **kwargs)
 
+    Note:
+        In the case of an ASGI request, this method will simulate the entire
+        app lifecycle in a single shot, including lifespan and client
+        disconnect events. In order to simulate multiple interleaved
+        requests, or to test a streaming endpoint (such as one that emits
+        server-sent events), :class:`~falcon.testing.ASGIConductor` can be
+        used to more precisely control the app lifecycle.
+
     Args:
         app (callable): The application to call
         path (str): The URL path to request
@@ -885,13 +1403,21 @@ def simulate_put(app, path, **kwargs) -> Result:
     return simulate_request(app, 'PUT', path, **kwargs)
 
 
-def simulate_options(app, path, **kwargs) -> Result:
+def simulate_options(app, path, **kwargs) -> _ResultBase:
     """Simulates an OPTIONS request to a WSGI or ASGI application.
 
     Equivalent to::
 
          simulate_request(app, 'OPTIONS', path, **kwargs)
 
+    Note:
+        In the case of an ASGI request, this method will simulate the entire
+        app lifecycle in a single shot, including lifespan and client
+        disconnect events. In order to simulate multiple interleaved
+        requests, or to test a streaming endpoint (such as one that emits
+        server-sent events), :class:`~falcon.testing.ASGIConductor` can be
+        used to more precisely control the app lifecycle.
+
     Args:
         app (callable): The application to call
         path (str): The URL path to request
@@ -959,13 +1485,21 @@ def simulate_options(app, path, **kwargs) -> Result:
     return simulate_request(app, 'OPTIONS', path, **kwargs)
 
 
-def simulate_patch(app, path, **kwargs) -> Result:
+def simulate_patch(app, path, **kwargs) -> _ResultBase:
     """Simulates a PATCH request to a WSGI or ASGI application.
 
     Equivalent to::
 
          simulate_request(app, 'PATCH', path, **kwargs)
 
+    Note:
+        In the case of an ASGI request, this method will simulate the entire
+        app lifecycle in a single shot, including lifespan and client
+        disconnect events. In order to simulate multiple interleaved
+        requests, or to test a streaming endpoint (such as one that emits
+        server-sent events), :class:`~falcon.testing.ASGIConductor` can be
+        used to more precisely control the app lifecycle.
+
     Args:
         app (callable): The application to call
         path (str): The URL path to request
@@ -1050,13 +1584,21 @@ def simulate_patch(app, path, **kwargs) -> Result:
     return simulate_request(app, 'PATCH', path, **kwargs)
 
 
-def simulate_delete(app, path, **kwargs) -> Result:
+def simulate_delete(app, path, **kwargs) -> _ResultBase:
     """Simulates a DELETE request to a WSGI or ASGI application.
 
     Equivalent to::
 
          simulate_request(app, 'DELETE', path, **kwargs)
 
+    Note:
+        In the case of an ASGI request, this method will simulate the entire
+        app lifecycle in a single shot, including lifespan and client
+        disconnect events. In order to simulate multiple interleaved
+        requests, or to test a streaming endpoint (such as one that emits
+        server-sent events), :class:`~falcon.testing.ASGIConductor` can be
+        used to more precisely control the app lifecycle.
+
     Args:
         app (callable): The application to call
         path (str): The URL path to request
@@ -1144,7 +1686,7 @@ def simulate_delete(app, path, **kwargs) -> Result:
 class TestClient:
     """Simulates requests to a WSGI or ASGI application.
 
-    This class provides a contextual wrapper for Falcon's `simulate_*`
+    This class provides a contextual wrapper for Falcon's ``simulate_*()``
     test functions. It lets you replace this::
 
         simulate_get(app, '/messages')
@@ -1160,6 +1702,27 @@ class TestClient:
         The methods all call ``self.simulate_request()`` for convenient
         overriding of request preparation by child classes.
 
+    Note:
+        In the case of an ASGI request, this class will simulate the entire
+        app lifecycle in a single shot, including lifespan and client
+        disconnect events. In order to simulate multiple interleaved
+        requests, or to test a streaming endpoint (such as one that emits
+        server-sent events), :class:`~falcon.testing.ASGIConductor` can be
+        used to more precisely control the app lifecycle.
+
+        An instance of :class:`~falcon.testing.ASGIConductor` may be
+        instantiated directly, or obtained from an instance of
+        :class:`~falcon.testing.TestClient` using the
+        context manager pattern, as per the following example::
+
+            client = falcon.testing.TestClient(app)
+
+            # -- snip --
+
+            async with client as conductor:
+                async with conductor.simulate_get_stream('/events') as result:
+                    pass
+
     Args:
         app (callable): A WSGI or ASGI application to target when simulating
             requests
@@ -1167,7 +1730,7 @@ class TestClient:
     Keyword Arguments:
         headers (dict): Default headers to set on every request (default
             ``None``). These defaults may be overridden by passing values
-            for the same headers to one of the `simulate_*()` methods.
+            for the same headers to one of the ``simulate_*()`` methods.
 
     Attributes:
         app: The app that this client instance was configured to use.
@@ -1177,58 +1740,83 @@ class TestClient:
     def __init__(self, app, headers=None):
         self.app = app
         self._default_headers = headers
+        self._conductor = None
+
+    async def __aenter__(self):
+        if not _is_asgi_app(self.app):
+            raise CompatibilityError(
+                'a conductor context manager may only be used with a Falcon ASGI app'
+            )
+
+        # NOTE(kgriffs): We normally do not expect someone to try to nest
+        #   contexts, so this is just a sanity-check.
+        assert not self._conductor
+
+        self._conductor = ASGIConductor(self.app, headers=self._default_headers)
+        await self._conductor.__aenter__()
+
+        return self._conductor
+
+    async def __aexit__(self, ex_type, ex, tb):
+        result = await self._conductor.__aexit__(ex_type, ex, tb)
+
+        # NOTE(kgriffs): Reset to allow this instance of TestClient to be
+        #   reused in another context.
+        self._conductor = None
 
-    def simulate_get(self, path='/', **kwargs) -> Result:
-        """Simulates a GET request to a WSGI application.
+        return result
+
+    def simulate_get(self, path='/', **kwargs) -> _ResultBase:
+        """Simulates a GET request to a WSGI or ASGI application.
 
         (See also: :py:meth:`falcon.testing.simulate_get`)
         """
         return self.simulate_request('GET', path, **kwargs)
 
-    def simulate_head(self, path='/', **kwargs) -> Result:
-        """Simulates a HEAD request to a WSGI application.
+    def simulate_head(self, path='/', **kwargs) -> _ResultBase:
+        """Simulates a HEAD request to a WSGI or ASGI application.
 
         (See also: :py:meth:`falcon.testing.simulate_head`)
         """
         return self.simulate_request('HEAD', path, **kwargs)
 
-    def simulate_post(self, path='/', **kwargs) -> Result:
-        """Simulates a POST request to a WSGI application.
+    def simulate_post(self, path='/', **kwargs) -> _ResultBase:
+        """Simulates a POST request to a WSGI or ASGI application.
 
         (See also: :py:meth:`falcon.testing.simulate_post`)
         """
         return self.simulate_request('POST', path, **kwargs)
 
-    def simulate_put(self, path='/', **kwargs) -> Result:
-        """Simulates a PUT request to a WSGI application.
+    def simulate_put(self, path='/', **kwargs) -> _ResultBase:
+        """Simulates a PUT request to a WSGI or ASGI application.
 
         (See also: :py:meth:`falcon.testing.simulate_put`)
         """
         return self.simulate_request('PUT', path, **kwargs)
 
-    def simulate_options(self, path='/', **kwargs) -> Result:
-        """Simulates an OPTIONS request to a WSGI application.
+    def simulate_options(self, path='/', **kwargs) -> _ResultBase:
+        """Simulates an OPTIONS request to a WSGI or ASGI application.
 
         (See also: :py:meth:`falcon.testing.simulate_options`)
         """
         return self.simulate_request('OPTIONS', path, **kwargs)
 
-    def simulate_patch(self, path='/', **kwargs) -> Result:
-        """Simulates a PATCH request to a WSGI application.
+    def simulate_patch(self, path='/', **kwargs) -> _ResultBase:
+        """Simulates a PATCH request to a WSGI or ASGI application.
 
         (See also: :py:meth:`falcon.testing.simulate_patch`)
         """
         return self.simulate_request('PATCH', path, **kwargs)
 
-    def simulate_delete(self, path='/', **kwargs) -> Result:
-        """Simulates a DELETE request to a WSGI application.
+    def simulate_delete(self, path='/', **kwargs) -> _ResultBase:
+        """Simulates a DELETE request to a WSGI or ASGI application.
 
         (See also: :py:meth:`falcon.testing.simulate_delete`)
         """
         return self.simulate_request('DELETE', path, **kwargs)
 
-    def simulate_request(self, *args, **kwargs) -> Result:
-        """Simulates a request to a WSGI application.
+    def simulate_request(self, *args, **kwargs) -> _ResultBase:
+        """Simulates a request to a WSGI or ASGI application.
 
         Wraps :py:meth:`falcon.testing.simulate_request` to perform a
         WSGI request directly against ``self.app``. Equivalent to::
@@ -1254,6 +1842,66 @@ class TestClient:
 # -----------------------------------------------------------------------------
 
 
+class _AsyncContextManager:
+    def __init__(self, coro):
+        self._coro = coro
+        self._obj = None
+
+    async def __aenter__(self):
+        self._obj = await self._coro
+        return self._obj
+
+    async def __aexit__(self, exc_type, exc, tb):
+        await self._obj.finalize()
+        self._obj = None
+
+
+def _prepare_sim_args(
+    path,
+    query_string,
+    params,
+    params_csv,
+    content_type,
+    headers,
+    body,
+    json,
+    extras
+):
+    if not path.startswith('/'):
+        raise ValueError("path must start with '/'")
+
+    if '?' in path:
+        if query_string or params:
+            raise ValueError(
+                'path may not contain a query string in combination with '
+                'the query_string or params parameters. Please use only one '
+                'way of specifying the query string.'
+            )
+        path, query_string = path.split('?', 1)
+    elif query_string and query_string.startswith('?'):
+        raise ValueError("query_string should not start with '?'")
+
+    extras = extras or {}
+
+    if query_string is None:
+        query_string = to_query_str(
+            params,
+            comma_delimited_lists=params_csv,
+            prefix=False,
+        )
+
+    if content_type is not None:
+        headers = headers or {}
+        headers['Content-Type'] = content_type
+
+    if json is not None:
+        body = util_json.dumps(json, ensure_ascii=False)
+        headers = headers or {}
+        headers['Content-Type'] = MEDIA_JSON
+
+    return path, query_string, headers, body, extras
+
+
 def _is_asgi_app(app):
     app_args = inspect.getfullargspec(app).args
     num_app_args = len(app_args)
@@ -1281,7 +1929,7 @@ async def _wait_for_startup(events):
             break
 
         # NOTE(kgriffs): Yield to the concurrent lifespan task
-        await asyncio.sleep(0.001)
+        await asyncio.sleep(0)
 
 
 async def _wait_for_shutdown(events):
@@ -1296,4 +1944,4 @@ async def _wait_for_shutdown(events):
             break
 
         # NOTE(kgriffs): Yield to the concurrent lifespan task
-        await asyncio.sleep(0.001)
+        await asyncio.sleep(0)
diff --git a/falcon/testing/helpers.py b/falcon/testing/helpers.py
index 9cda9eb..e5c1b95 100644
--- a/falcon/testing/helpers.py
+++ b/falcon/testing/helpers.py
@@ -33,8 +33,9 @@ import random
 import socket
 import sys
 import time
-from typing import Any, Dict
+from typing import Any, Dict, Union
 
+from falcon.asgi_spec import EventType, ScopeType
 from falcon.constants import SINGLETON_HEADERS
 import falcon.request
 from falcon.util import http_now, uri
@@ -72,18 +73,18 @@ class ASGILifespanEventEmitter:
     async def emit(self):
         if self._state == 0:
             self._state += 1
-            return {'type': 'lifespan.startup'}
+            return {'type': EventType.LIFESPAN_STARTUP}
 
         if self._state == 1:
             self._state += 1
-            # NOTE(kgriffs): This ensures the app ignores events it does
+            # NOTE(kgriffs): This verifies the app ignores events it does
             #   not recognize.
             return {'type': 'lifespan._nonstandard_event'}
 
         async with self._shutting_down:
             await self._shutting_down.wait()
 
-        return {'type': 'lifespan.shutdown'}
+        return {'type': EventType.LIFESPAN_SHUTDOWN}
 
     __call__ = emit
 
@@ -107,17 +108,28 @@ class ASGIRequestEventEmitter:
             events. May be an empty string. If a byte string, it will
             be used as-is; otherwise it will be encoded as UTF-8
             (default b'').
-        disconnect_at (int): The Unix timestamp after which to begin
-            returning http.disconnect events (default now + 30s).
         chunk_size (int): The maximum number of bytes to include in
             a single http.request event (default 4096).
+        disconnect_at (float): The Unix timestamp after which to begin
+            emitting http.disconnect events (default now + 30s). The
+            value may be either an ``int`` or a ``float``, depending
+            on the precision required.
+
+    Attributes:
+        disconnected (bool): Returns ``True`` if the simulated client
+            connection is in a "disconnected" state.
     """
 
     # TODO(kgriffs): If this pattern later becomes useful elsewhere,
     #   factor out into a standalone helper class.
     _branch_decider = defaultdict(bool)  # type: defaultdict
 
-    def __init__(self, body=None, disconnect_at=None, chunk_size=4096):
+    def __init__(
+        self,
+        body: Union[str, bytes] = None,
+        chunk_size: int = None,
+        disconnect_at: Union[int, float] = None
+    ):
         if body is None:
             body = b''
         elif not isinstance(body, bytes):
@@ -126,25 +138,56 @@ class ASGIRequestEventEmitter:
         if disconnect_at is None:
             disconnect_at = time.time() + 30
 
+        if chunk_size is None:
+            chunk_size = 4096
+
         self._body = body
         self._chunk_size = chunk_size
         self._disconnect_at = disconnect_at
+        self._disconnected = False
+        self._exhaust_body = True
 
         self._emitted_empty_chunk_a = False
         self._emitted_empty_chunk_b = False
 
+    @property
+    def disconnected(self):
+        return self._disconnected or (self._disconnect_at <= time.time())
+
+    def disconnect(self, exhaust_body: bool = None):
+        """Set the client connection state to disconnected.
+
+        Call this method to simulate an immediate client disconnect and
+        begin emitting 'http.disconnect' events.
+
+        Arguments:
+            exhaust_body (bool): Set to ``False`` in order to
+                begin emitting 'http.disconnect' events without first
+                emitting at least one 'http.request' event.
+        """
+
+        if exhaust_body is not None:
+            self._exhaust_body = exhaust_body
+
+        self._disconnected = True
+
     async def emit(self):
-        if self._body is None:
+        #
+        # NOTE(kgriffs): Based on my reading of the ASGI spec, at least one
+        #   'http.request' event should be emitted before 'http.disconnect'.
+        #
+        #   See also: https://asgi.readthedocs.io/en/latest/specs/main.html#events
+        #
+        if self._body is None or not self._exhaust_body:
             # NOTE(kgriffs): When there are no more events, an ASGI
             #   server will hang until the client connection
             #   disconnects.
-            while time.time() < self._disconnect_at:
-                await asyncio.sleep(1)
+            while not self.disconnected:
+                await asyncio.sleep(0.1)
 
-        if self._disconnect_at <= time.time():
-            return {'type': 'http.disconnect'}
+            return {'type': EventType.HTTP_DISCONNECT}
 
-        event = {'type': 'http.request'}
+        event = {'type': EventType.HTTP_REQUEST}
 
         # NOTE(kgriffs): Return a couple variations on empty chunks
         #   every time, to ensure test coverage.
@@ -199,7 +242,7 @@ class ASGIRequestEventEmitter:
 
     __call__ = emit
 
-    def _toggle_branch(self, name):
+    def _toggle_branch(self, name: str):
         self._branch_decider[name] = not self._branch_decider[name]
         return self._branch_decider[name]
 
@@ -240,7 +283,7 @@ class ASGIResponseEventCollector:
         self.body_chunks = []
         self.more_body = None
 
-    async def collect(self, event):
+    async def collect(self, event: Dict[str, Any]):
         if self.more_body is False:
             # NOTE(kgriffs): According to the ASGI spec, once we get a
             #   message setting more_body to False, any further messages
@@ -407,7 +450,7 @@ def create_scope(path='/', query_string='', method='GET', headers=None,
         raise ValueError("query_string should not start with '?'")
 
     scope = {
-        'type': 'http',
+        'type': ScopeType.HTTP,
         'asgi': {
             'version': '3.0',
             'spec_version': '2.1',
@@ -667,7 +710,7 @@ def create_asgi_req(body=None, req_type=None, options=None, **kwargs) -> falcon.
     body = body or b''
     disconnect_at = time.time() + 300
 
-    req_event_emitter = ASGIRequestEventEmitter(body, disconnect_at)
+    req_event_emitter = ASGIRequestEventEmitter(body, disconnect_at=disconnect_at)
 
     # NOTE(kgriffs): Import here in case the app is running under
     #   Python 3.5 (in which case as long as it does not call the
diff --git a/requirements/tests b/requirements/tests
index 440dc43..c0b32f1 100644
--- a/requirements/tests
+++ b/requirements/tests
@@ -5,6 +5,7 @@ requests
 testtools
 
 # ASGI Specific
+pytest-asyncio
 httpx; python_version >= '3.6'
 uvicorn; python_version >= '3.6'
 aiofiles; python_version >= '3.6'
diff --git a/tests/asgi/_asgi_test_app.py b/tests/asgi/_asgi_test_app.py
index 60b2ebd..b68fe09 100644
--- a/tests/asgi/_asgi_test_app.py
+++ b/tests/asgi/_asgi_test_app.py
@@ -8,6 +8,7 @@ import falcon
 import falcon.asgi
 import falcon.util
 
+SSE_TEST_MAX_DELAY_SEC = 1
 _WIN32 = sys.platform.startswith('win')
 
 
@@ -69,7 +70,7 @@ class Things:
             safely_values.append((a, b, c))
 
         cms = falcon.util.wrap_sync_to_async(callmesafely, threadsafe=False)
-        loop = falcon.util.get_loop()
+        loop = falcon.util.get_running_loop()
 
         # NOTE(caselit): on windows it takes more time so create less tasks
         num_cms_tasks = 100 if _WIN32 else 1000
@@ -104,10 +105,11 @@ class Bucket:
 class Events:
     async def on_get(self, req, resp):
         async def emit():
-            start = time.time()
-            while time.time() - start < 1:
+            s = 0
+            while s <= SSE_TEST_MAX_DELAY_SEC:
                 yield falcon.asgi.SSEvent(text='hello world')
-                await asyncio.sleep(0.2)
+                await asyncio.sleep(s)
+                s += (SSE_TEST_MAX_DELAY_SEC / 4)
 
         resp.sse = emit()
 
diff --git a/tests/asgi/test_asgi_conductor.py b/tests/asgi/test_asgi_conductor.py
new file mode 100644
index 0000000..656794b
--- /dev/null
+++ b/tests/asgi/test_asgi_conductor.py
@@ -0,0 +1,67 @@
+import pytest
+
+import falcon
+from falcon import testing
+from falcon.asgi import App
+
+
+@pytest.mark.asyncio
+async def test_default_headers():
+    class Resource:
+        async def on_get(self, req, resp):
+            resp.set_header('the-answer', req.get_header('the-answer'))
+
+    app = App()
+    app.add_route('/', Resource())
+    client = testing.TestClient(app, headers={'the-answer': '42'})
+
+    async with client as conductor:
+        result = await conductor.simulate_get()
+        assert result.headers['the-answer'] == '42'
+
+
+@pytest.mark.asyncio
+async def test_wsgi_not_supported():
+    with pytest.raises(falcon.CompatibilityError):
+        async with testing.TestClient(falcon.App()):
+            pass
+
+    with pytest.raises(falcon.CompatibilityError):
+        async with testing.ASGIConductor(falcon.App()):
+            pass
+
+
+@pytest.mark.parametrize('method', ['get', 'head', 'post', 'put', 'options', 'patch', 'delete'])
+@pytest.mark.asyncio
+async def test_responders(method):
+    class Resource:
+        async def on_get(self, req, resp):
+            resp.set_header('method', 'get')
+
+        async def on_head(self, req, resp):
+            resp.set_header('method', 'head')
+
+        async def on_post(self, req, resp):
+            resp.set_header('method', 'post')
+
+        async def on_put(self, req, resp):
+            resp.set_header('method', 'put')
+
+        async def on_options(self, req, resp):
+            resp.set_header('method', 'options')
+
+        async def on_patch(self, req, resp):
+            resp.set_header('method', 'patch')
+
+        async def on_delete(self, req, resp):
+            resp.set_header('method', 'delete')
+
+    resource = Resource()
+
+    app = App()
+    app.add_route('/', resource)
+
+    async with testing.ASGIConductor(app) as conductor:
+        simulate = getattr(conductor, 'simulate_' + method)
+        result = await simulate('/')
+        assert result.headers['method'] == method
diff --git a/tests/asgi/test_asgi_servers.py b/tests/asgi/test_asgi_servers.py
index 4b6ed16..cab6df2 100644
--- a/tests/asgi/test_asgi_servers.py
+++ b/tests/asgi/test_asgi_servers.py
@@ -12,6 +12,7 @@ import requests
 import requests.exceptions
 
 from falcon import testing
+from . import _asgi_test_app
 
 
 _MODULE_DIR = os.path.abspath(os.path.dirname(__file__))
@@ -118,6 +119,21 @@ class TestASGIServer:
 
         assert not events[-1]
 
+    def test_sse_client_disconnects_early(self, server_base_url):
+        """Test that when the client connection is lost, the server task does not hang.
+
+        In the case of SSE, Falcon should detect when the client connection is
+        lost and immediately bail out. Currently this is observable by watching
+        the output of the uvicorn and daphne server processes. Also, the
+        _run_server_isolated() method will fail the test if the server process
+        takes too long to shut down.
+        """
+        with pytest.raises(requests.exceptions.ConnectionError):
+            requests.get(
+                server_base_url + 'events',
+                timeout=(_asgi_test_app.SSE_TEST_MAX_DELAY_SEC / 2)
+            )
+
 
 @contextmanager
 def _run_server_isolated(process_factory, host, port):
@@ -149,8 +165,11 @@ def _run_server_isolated(process_factory, host, port):
             pass
         except subprocess.TimeoutExpired:
             print('\n[Killing stubborn server process...]')
+
             server.kill()
             server.communicate()
+
+            pytest.fail('Server process did not exit in a timely manner and had to be killed.')
     else:
         print('\n[Sending SIGTERM to server process...]')
         server.terminate()
@@ -159,9 +178,12 @@ def _run_server_isolated(process_factory, host, port):
             server.communicate(timeout=10)
         except subprocess.TimeoutExpired:
             print('\n[Killing stubborn server process...]')
+
             server.kill()
             server.communicate()
 
+            pytest.fail('Server process did not exit in a timely manner and had to be killed.')
+
 
 def _uvicorn_factory(host, port):
     if _WIN32:
diff --git a/tests/asgi/test_boundedstream_asgi.py b/tests/asgi/test_boundedstream_asgi.py
index 7d38607..73fe07e 100644
--- a/tests/asgi/test_boundedstream_asgi.py
+++ b/tests/asgi/test_boundedstream_asgi.py
@@ -1,6 +1,3 @@
-import asyncio
-import time
-
 import pytest
 
 import falcon
@@ -212,14 +209,16 @@ def test_read_chunks(body, chunk_size):
 def test_exhaust_with_disconnect():
     async def t():
         emitter = testing.ASGIRequestEventEmitter(
-            b'123456798' * 1024,
-            disconnect_at=(time.time() + 0.5)
+            b'123456789' * 2,
+
+            # NOTE(kgriffs): This must be small enough to create several events
+            chunk_size=3,
         )
         s = asgi.BoundedStream(emitter)
 
         assert await s.read(1) == b'1'
         assert await s.read(2) == b'23'
-        await asyncio.sleep(0.5)
+        emitter.disconnect(exhaust_body=False)
         await s.exhaust()
         assert await s.read(1) == b''
         assert await s.read(100) == b''
diff --git a/tests/asgi/test_lifespan_handlers.py b/tests/asgi/test_lifespan_handlers.py
index 2d7bcfd..fbd2934 100644
--- a/tests/asgi/test_lifespan_handlers.py
+++ b/tests/asgi/test_lifespan_handlers.py
@@ -1,5 +1,6 @@
 import pytest
 
+import falcon
 from falcon import testing
 from falcon.asgi import App
 
@@ -203,3 +204,33 @@ def test_multiple_handlers():
     assert b._called_shutdown == 6
 
     assert e._called_request
+
+
+def test_asgi_conductor_raised_error_skips_shutdown():
+    class SomeException(Exception):
+        pass
+
+    class Foo:
+        def __init__(self):
+            self.called_startup = False
+            self.called_shutdown = False
+
+        async def process_startup(self, scope, event):
+            self.called_startup = True
+
+        async def process_shutdown(self, scope, event):
+            self.called_shutdown = True
+
+    foo = Foo()
+
+    app = App()
+    app.add_middleware(foo)
+
+    async def t():
+        with pytest.raises(SomeException):
+            async with testing.ASGIConductor(app):
+                raise SomeException()
+
+    falcon.invoke_coroutine_sync(t)
+    assert foo.called_startup
+    assert not foo.called_shutdown
diff --git a/tests/asgi/test_request_body_asgi.py b/tests/asgi/test_request_body_asgi.py
index 3acbed3..19967b0 100644
--- a/tests/asgi/test_request_body_asgi.py
+++ b/tests/asgi/test_request_body_asgi.py
@@ -72,14 +72,14 @@ class TestRequestBody:
 
     def test_bounded_stream_alias(self):
         scope = testing.create_scope()
-        req_event_emitter = testing.ASGIRequestEventEmitter(b'', 0)
+        req_event_emitter = testing.ASGIRequestEventEmitter(b'', disconnect_at=0)
         req = falcon.asgi.Request(scope, req_event_emitter)
 
         assert req.bounded_stream is req.stream
 
     def test_request_repr(self):
         scope = testing.create_scope()
-        req_event_emitter = testing.ASGIRequestEventEmitter(b'', 0)
+        req_event_emitter = testing.ASGIRequestEventEmitter(b'', disconnect_at=0)
         req = falcon.asgi.Request(scope, req_event_emitter)
 
         _repr = '<%s: %s %r>' % (req.__class__.__name__, req.method, req.url)
diff --git a/tests/asgi/test_sse.py b/tests/asgi/test_sse.py
index ac55061..97fd467 100644
--- a/tests/asgi/test_sse.py
+++ b/tests/asgi/test_sse.py
@@ -1,5 +1,8 @@
+import asyncio
+
 import pytest
 
+import falcon
 from falcon import testing
 from falcon.asgi import App, SSEvent
 
@@ -58,27 +61,7 @@ def test_single_event():
 
 
 def test_multiple_events():
-    class SomeResource:
-        async def on_get(self, req, resp):
-            async def emitter():
-                yield SSEvent(data=b'ketchup')
-                yield SSEvent(data=b'mustard', event='condiment')
-                yield SSEvent(data=b'mayo', event='condiment', event_id='1234')
-                yield SSEvent(data=b'onions', event='topping', event_id='5678', retry=100)
-                yield SSEvent(text='guacamole \u1F951', retry=100, comment='Serve with chips.')
-                yield SSEvent(json={'condiment': 'salsa'}, retry=100)
-
-            resp.sse = emitter()
-
-    resource = SomeResource()
-
-    app = App()
-    app.add_route('/', resource)
-
-    client = testing.TestClient(app)
-
-    result = client.simulate_get()
-    assert result.text == (
+    expected_result_text = (
         'data: ketchup\n'
         '\n'
         'event: condiment\n'
@@ -102,6 +85,102 @@ def test_multiple_events():
         '\n'
     )
 
+    class SomeResource:
+        async def on_get(self, req, resp):
+            async def emitter():
+                for event in [
+                    SSEvent(data=b'ketchup'),
+                    SSEvent(data=b'mustard', event='condiment'),
+                    SSEvent(data=b'mayo', event='condiment', event_id='1234'),
+                    SSEvent(data=b'onions', event='topping', event_id='5678', retry=100),
+                    SSEvent(text='guacamole \u1F951', retry=100, comment='Serve with chips.'),
+                    SSEvent(json={'condiment': 'salsa'}, retry=100),
+                ]:
+                    yield event
+                    await asyncio.sleep(0.001)
+
+            resp.sse = emitter()
+
+    resource = SomeResource()
+
+    app = App()
+    app.add_route('/', resource)
+
+    client = testing.TestClient(app)
+
+    async def _test():
+        async with client as conductor:
+            # NOTE(kgriffs): Single-shot test will only allow the first
+            #   one or two events since a client disconnect will be emitted
+            #   into the app immediately.
+            result = await conductor.simulate_get()
+            assert expected_result_text.startswith(result.text)
+
+            async with conductor.simulate_get_stream() as sr:
+                event_count = 0
+
+                result_text = ''
+
+                while True:
+                    chunk = (await sr.stream.read()).decode()
+
+                    if not chunk:
+                        continue
+
+                    result_text += chunk
+                    event_count += len(chunk.strip().split('\n\n'))
+
+                    if 'salsa' in chunk:
+                        break
+
+                assert not (await sr.stream.read())
+
+                assert event_count == 6
+                assert result_text == expected_result_text
+
+    falcon.invoke_coroutine_sync(_test)
+
+
+def test_multiple_events_early_disconnect():
+    class SomeResource:
+        async def on_get(self, req, resp):
+            async def emitter():
+                while True:
+                    yield SSEvent(data=b'whassup')
+                    await asyncio.sleep(0.01)
+
+            resp.sse = emitter()
+
+    resource = SomeResource()
+
+    app = App()
+    app.add_route('/', resource)
+
+    async def _test():
+        conductor = testing.ASGIConductor(app)
+        result = await conductor.simulate_get()
+        assert 'data: whassup' in result.text
+
+        async with testing.ASGIConductor(app) as conductor:
+            async with conductor.simulate_get_stream() as sr:
+
+                event_count = 0
+
+                result_text = ''
+
+                while event_count < 5:
+                    chunk = (await sr.stream.read()).decode()
+                    if not chunk:
+                        continue
+
+                    result_text += chunk
+                    event_count += len(chunk.strip().split('\n\n'))
+
+                assert result_text.startswith('data: whassup\n\n' * 5)
+                assert event_count == 5
+
+    falcon.invoke_coroutine_sync(_test)
+
 
 def test_invalid_event_values():
     with pytest.raises(TypeError):
diff --git a/tests/asgi/test_sync.py b/tests/asgi/test_sync.py
index e4b58b0..80f827b 100644
--- a/tests/asgi/test_sync.py
+++ b/tests/asgi/test_sync.py
@@ -43,7 +43,7 @@ def test_sync_helpers():
             cmus = falcon.util.wrap_sync_to_async(callme_unsafely, threadsafe=True)
             cms = falcon.util.wrap_sync_to_async(callme_safely, threadsafe=False)
 
-            loop = falcon.util.get_loop()
+            loop = falcon.util.get_running_loop()
 
             # NOTE(kgriffs): create_task() is used here, so that the coroutines
             #   are scheduled immediately in the order created; under Python
diff --git a/tests/asgi/test_testing_asgi.py b/tests/asgi/test_testing_asgi.py
new file mode 100644
index 0000000..336df2f
--- /dev/null
+++ b/tests/asgi/test_testing_asgi.py
@@ -0,0 +1,104 @@
+import time
+
+import pytest
+
+from falcon import testing
+
+
+@pytest.mark.asyncio
+async def test_asgi_request_event_emitter_hang():
+    # NOTE(kgriffs): This tests the ASGI server behavior that
+    #   ASGIRequestEventEmitter simulates when emit() is called
+    #   again after there are no more events available.
+
+    expected_elapsed_min = 1
+    disconnect_at = time.time() + expected_elapsed_min
+
+    emit = testing.ASGIRequestEventEmitter(disconnect_at=disconnect_at)
+
+    start = time.time()
+    while True:
+        event = await emit()
+        if not event.get('more_body', False):
+            break
+
+    elapsed = time.time() - start
+    assert elapsed < 0.1
+
+    start = time.time()
+    await emit()
+    elapsed = time.time() - start
+
+    assert (elapsed + 0.1) > expected_elapsed_min
+
+
+@pytest.mark.asyncio
+async def test_ignore_extra_asgi_events():
+    collect = testing.ASGIResponseEventCollector()
+
+    await collect({'type': 'http.response.start', 'status': 200})
+    await collect({'type': 'http.response.body', 'more_body': False})
+
+    # NOTE(kgriffs): Events after more_body is False are ignored to conform
+    #   to the ASGI spec.
+    await collect({'type': 'http.response.body'})
+    assert len(collect.events) == 2
+
+
+@pytest.mark.asyncio
+async def test_invalid_asgi_events():
+    collect = testing.ASGIResponseEventCollector()
+
+    def make_event(headers=None, status=200):
+        return {
+            'type': 'http.response.start',
+            'headers': headers or [],
+            'status': status
+        }
+
+    with pytest.raises(TypeError):
+        await collect({'type': 123})
+
+    with pytest.raises(TypeError):
+        headers = [
+            ('notbytes', b'bytes')
+        ]
+        await collect(make_event(headers))
+
+    with pytest.raises(TypeError):
+        headers = [
+            (b'bytes', 'notbytes')
+        ]
+        await collect(make_event(headers))
+
+    with pytest.raises(ValueError):
+        headers = [
+            # NOTE(kgriffs): Name must be lowercase
+            (b'Content-Type', b'application/json')
+        ]
+        await collect(make_event(headers))
+
+    with pytest.raises(TypeError):
+        await collect(make_event(status='200'))
+
+    with pytest.raises(TypeError):
+        await collect(make_event(status=200.1))
+
+    with pytest.raises(TypeError):
+        await collect({'type': 'http.response.body', 'body': 'notbytes'})
+
+    with pytest.raises(TypeError):
+        await collect({'type': 'http.response.body', 'more_body': ''})
+
+    with pytest.raises(ValueError):
+        # NOTE(kgriffs): Invalid type
+        await collect({'type': 'http.response.bod'})
+
+
+def test_is_asgi_app_cls():
+    class Foo:
+        @classmethod
+        def class_meth(cls, scope, receive, send):
+            pass
+
+    assert testing.client._is_asgi_app(Foo.class_meth)
diff --git a/tests/test_testing.py b/tests/test_testing.py
index 221c159..f013ddc 100644
--- a/tests/test_testing.py
+++ b/tests/test_testing.py
@@ -1,5 +1,3 @@
-import time
-
 import pytest
 
 import falcon
@@ -61,111 +59,6 @@ def test_simulate_request_http_version(version, valid):
             testing.simulate_request(app, http_version=version)
 
 
-def test_asgi_request_event_emitter_hang():
-    # NOTE(kgriffs): This tests the ASGI server behavior that
-    #   ASGIRequestEventEmitter simulates when emit() is called
-    #   again after there are not more events available.
-
-    expected_elasped_min = 1
-    disconnect_at = time.time() + expected_elasped_min
-
-    emit = testing.ASGIRequestEventEmitter(disconnect_at=disconnect_at)
-
-    async def t():
-        start = time.time()
-        while True:
-            event = await emit()
-            if not event.get('more_body', False):
-                break
-        elapsed = time.time() - start
-
-        assert elapsed < 0.1
-
-        start = time.time()
-        await emit()
-        elapsed = time.time() - start
-
-        assert (elapsed + 0.1) > expected_elasped_min
-
-    falcon.invoke_coroutine_sync(t)
-
-
-def test_ignore_extra_asgi_events():
-    collect = testing.ASGIResponseEventCollector()
-
-    async def t():
-        await collect({'type': 'http.response.start', 'status': 200})
-        await collect({'type': 'http.response.body', 'more_body': False})
-
-        # NOTE(kgriffs): Events after more_body is False are ignored to conform
-        #   to the ASGI spec.
-        await collect({'type': 'http.response.body'})
-        assert len(collect.events) == 2
-
-    falcon.invoke_coroutine_sync(t)
-
-
-def test_invalid_asgi_events():
-    collect = testing.ASGIResponseEventCollector()
-
-    def make_event(headers=None, status=200):
-        return {
-            'type': 'http.response.start',
-            'headers': headers or [],
-            'status': status
-        }
-
-    async def t():
-        with pytest.raises(TypeError):
-            await collect({'type': 123})
-
-        with pytest.raises(TypeError):
-            headers = [
-                ('notbytes', b'bytes')
-            ]
-            await collect(make_event(headers))
-
-        with pytest.raises(TypeError):
-            headers = [
-                (b'bytes', 'notbytes')
-            ]
-            await collect(make_event(headers))
-
-        with pytest.raises(ValueError):
-            headers = [
-                # NOTE(kgriffs): Name must be lowercase
-                (b'Content-Type', b'application/json')
-            ]
-            await collect(make_event(headers))
-
-        with pytest.raises(TypeError):
-            await collect(make_event(status='200'))
-
-        with pytest.raises(TypeError):
-            await collect(make_event(status=200.1))
-
-        with pytest.raises(TypeError):
-            await collect({'type': 'http.response.body', 'body': 'notbytes'})
-
-        with pytest.raises(TypeError):
-            await collect({'type': 'http.response.body', 'more_body': ''})
-
-        with pytest.raises(ValueError):
-            # NOTE(kgriffs): Invalid type
-            await collect({'type': 'http.response.bod'})
-
-    falcon.invoke_coroutine_sync(t)
-
-
-def test_is_asgi_app_cls():
-    class Foo:
-        @classmethod
-        def class_meth(cls, scope, receive, send):
-            pass
-
-    assert testing.client._is_asgi_app(Foo.class_meth)
-
-
 def test_simulate_request_content_type():
     class Foo:
         def on_post(self, req, resp):
diff --git a/tools/mintest.sh b/tools/mintest.sh
index 771974d..90d6c66 100755
--- a/tools/mintest.sh
+++ b/tools/mintest.sh
@@ -3,4 +3,4 @@
 pip install -U tox coverage
 
 rm -f .coverage.*
-tox -e pep8 && tox -e py35,py38 && tools/testing/combine_coverage.sh
+tox -e pep8 && tox -e mypy && tox -e py35,py38 && tools/testing/combine_coverage.sh