Repository: falconry/falcon

Issue:
ASGI+multipart/form-data
Add ASGI support to [MultipartFormHandler](https://falcon.readthedocs.io/en/latest/api/media.html#falcon.media.MultipartFormHandler).

The parsing interface could look similar to the [WSGI flavour](https://falcon.readthedocs.io/en/latest/api/multipart.html), just async:
```python
form = await req.get_media()

async for part in form:
    async with aiofiles.open('mydest', 'wb') as dest:
        await part.stream.pipe(dest)
```

There's an open question what to do with properties that would become async, such as [`part.text`](https://falcon.readthedocs.io/en/latest/api/multipart.html#falcon.media.multipart.BodyPart.text). We may need to refactor them to `async def BodyPart.get_text()` etc. Suggestions welcome!

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

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

```diff
@@             Coverage Diff             @@
##            master    #1728      +/-   ##
===========================================
- Coverage   100.00%   99.78%   -0.22%     
===========================================
  Files           49       51       +2     
  Lines         3915     4174     +259     
  Branches       608      659      +51     
===========================================
+ Hits          3915     4165     +250     
- Misses           0        8       +8     
- Partials         0        1       +1     
```


| [Impacted Files](https://codecov.io/gh/falconry/falcon/pull/1728?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [falcon/asgi/app.py](https://codecov.io/gh/falconry/falcon/pull/1728/diff?src=pr&el=tree#diff-ZmFsY29uL2FzZ2kvYXBwLnB5) | `100.00% <100.00%> (ø)` | |
| [falcon/asgi/multipart.py](https://codecov.io/gh/falconry/falcon/pull/1728/diff?src=pr&el=tree#diff-ZmFsY29uL2FzZ2kvbXVsdGlwYXJ0LnB5) | `100.00% <100.00%> (ø)` | |
| [falcon/asgi/reader.py](https://codecov.io/gh/falconry/falcon/pull/1728/diff?src=pr&el=tree#diff-ZmFsY29uL2FzZ2kvcmVhZGVyLnB5) | `100.00% <100.00%> (ø)` | |
| [falcon/asgi/request.py](https://codecov.io/gh/falconry/falcon/pull/1728/diff?src=pr&el=tree#diff-ZmFsY29uL2FzZ2kvcmVxdWVzdC5weQ==) | `100.00% <100.00%> (ø)` | |
| [falcon/media/multipart.py](https://codecov.io/gh/falconry/falcon/pull/1728/diff?src=pr&el=tree#diff-ZmFsY29uL21lZGlhL211bHRpcGFydC5weQ==) | `100.00% <100.00%> (ø)` | |
| [falcon/testing/helpers.py](https://codecov.io/gh/falconry/falcon/pull/1728/diff?src=pr&el=tree#diff-ZmFsY29uL3Rlc3RpbmcvaGVscGVycy5weQ==) | `100.00% <100.00%> (ø)` | |
| [falcon/asgi/stream.py](https://codecov.io/gh/falconry/falcon/pull/1728/diff?src=pr&el=tree#diff-ZmFsY29uL2FzZ2kvc3RyZWFtLnB5) | `94.00% <0.00%> (-6.00%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/falconry/falcon/pull/1728?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/1728?src=pr&el=footer). Last update [7242b08...7137eea](https://codecov.io/gh/falconry/falcon/pull/1728?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/docs/api/multipart.rst b/docs/api/multipart.rst
index 78570bb..aafae43 100644
--- a/docs/api/multipart.rst
+++ b/docs/api/multipart.rst
@@ -11,25 +11,54 @@ Falcon features easy and efficient access to submitted multipart forms by using
 default, allowing you to use ``req.get_media()`` to iterate over the
 :class:`body parts <falcon.media.multipart.BodyPart>` in a form:
 
-.. code:: python
-
-    for part in req.get_media():
-        if part.content_type == 'application/json':
-            # TODO: Body part is a JSON document, do something useful with it
-            resp.media = part.media
-        elif part.name == 'datafile':
-            while True:
-                # Do something with the uploaded data (file)
-                chunk = part.stream.read(8192)
-                if not chunk:
-                    break
-                feed_data(chunk)
-        elif part.name == 'imagedata':
-            # Store this body part in a file.
-            with open(path.join(UPLOADS, part.secure_filename), 'wb') as dest:
-                part.stream.pipe(dest)
-        else:
-            # TODO: Do something else
+.. tabs::
+
+    .. group-tab:: WSGI
+
+        .. code:: python
+
+            form = req.get_media()
+            for part in form:
+                if part.content_type == 'application/json':
+                    # Body part is a JSON document, do something useful with it
+                    resp.media = part.get_media()
+                elif part.name == 'datafile':
+                    while True:
+                        # Do something with the uploaded data (file)
+                        chunk = part.stream.read(8192)
+                        if not chunk:
+                            break
+                        feed_data(chunk)
+                elif part.name == 'imagedata':
+                    # Store this body part in a file.
+                    filename = os.path.join(UPLOAD_PATH, part.secure_filename)
+                    with open(filename, 'wb') as dest:
+                        part.stream.pipe(dest)
+                else:
+                    # Do something else
+                    form_data[part.name] = part.text
+
+    .. group-tab:: ASGI
+
+        .. code:: python
+
+            form = await req.get_media()
+            async for part in form:
+                if part.content_type == 'application/json':
+                    # Body part is a JSON document, do something useful with it
+                    resp.media = await part.get_media()
+                elif part.name == 'datafile':
+                    # Do something with the uploaded data (file)
+                    async for chunk in part.stream:
+                        await feed_data(chunk)
+                elif part.name == 'imagedata':
+                    # Store this body part in a file.
+                    filename = os.path.join(UPLOAD_PATH, part.secure_filename)
+                    async with aiofiles.open(filename, 'wb') as dest:
+                        await part.stream.pipe(dest)
+                else:
+                    # Do something else
+                    form_data[part.name] = await part.text
 
 .. note::
    Rather than being read in and buffered all at once, the request stream is
@@ -45,6 +74,7 @@ Body Part Type
 
 .. autoclass:: falcon.media.multipart.BodyPart
     :members:
+    :exclude-members: data, media, text
 
 Parsing Options
 ---------------
diff --git a/docs/api/util.rst b/docs/api/util.rst
index 4234092..3eda5a0 100644
--- a/docs/api/util.rst
+++ b/docs/api/util.rst
@@ -39,6 +39,8 @@ Async
 .. autofunction:: falcon.sync_to_async
 .. autofunction:: falcon.wrap_sync_to_async
 .. autofunction:: falcon.wrap_sync_to_async_unsafe
+.. autofunction:: falcon.invoke_coroutine_sync
+.. autofunction:: falcon.runs_sync
 
 Other
 -----
diff --git a/falcon/app.py b/falcon/app.py
index 5b85a67..85b2963 100644
--- a/falcon/app.py
+++ b/falcon/app.py
@@ -28,6 +28,7 @@ from falcon.request import Request, RequestOptions
 import falcon.responders
 from falcon.response import Response, ResponseOptions
 import falcon.status_codes as status
+from falcon.util import deprecation
 from falcon.util import misc
 from falcon.util.misc import code_to_http_status
 
@@ -1014,6 +1015,7 @@ class API(App):
     removed in a future release.
     """
 
-    @misc.deprecated('API class may be removed in a future release, use falcon.App instead.')
+    @deprecation.deprecated('API class may be removed in a future release, '
+                            'use falcon.App instead.')
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
diff --git a/falcon/asgi/app.py b/falcon/asgi/app.py
index 29d018c..d6fa7b5 100644
--- a/falcon/asgi/app.py
+++ b/falcon/asgi/app.py
@@ -22,13 +22,17 @@ from falcon.app_helpers import prepare_middleware
 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 .multipart import MultipartForm
 from .request import Request
 from .response import Response
 from .structures import SSEvent
 
+# TODO(vytas): Clean up these foul workarounds when we drop Python 3.5 support.
+MultipartFormHandler._ASGI_MULTIPART_FORM = MultipartForm  # type: ignore
 
 __all__ = ['App']
 
diff --git a/falcon/asgi/multipart.py b/falcon/asgi/multipart.py
new file mode 100644
index 0000000..8ca8569
--- /dev/null
+++ b/falcon/asgi/multipart.py
@@ -0,0 +1,160 @@
+# Copyright 2019-2020 by Vytautas Liuolia.
+#
+# 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.
+
+"""ASGI multipart form media handler components."""
+
+import cgi
+
+from falcon.asgi.reader import BufferedReader
+from falcon.errors import DelimiterError
+from falcon.media import multipart
+
+_ALLOWED_CONTENT_HEADERS = multipart._ALLOWED_CONTENT_HEADERS
+_CRLF = multipart._CRLF
+_CRLF_CRLF = multipart._CRLF_CRLF
+
+MultipartParseError = multipart.MultipartParseError
+
+
+class BodyPart(multipart.BodyPart):
+
+    async def get_data(self):
+        if self._data is None:
+            max_size = self._parse_options.max_body_part_buffer_size + 1
+            self._data = await self.stream.read(max_size)
+            if len(self._data) >= max_size:
+                raise MultipartParseError(description='body part is too large')
+
+        return self._data
+
+    async def get_media(self):
+        if self._media is None:
+            handler = self._parse_options.media_handlers.find_by_media_type(
+                self.content_type, 'text/plain')
+
+            try:
+                self._media = await handler.deserialize_async(
+                    self.stream, self.content_type, None)
+            finally:
+                if handler.exhaust_stream:
+                    await self.stream.exhaust()
+
+        return self._media
+
+    async def get_text(self):
+        content_type, options = cgi.parse_header(self.content_type)
+        if content_type != 'text/plain':
+            return None
+
+        charset = options.get('charset', self._parse_options.default_charset)
+        try:
+            return (await self.get_data()).decode(charset)
+        except (ValueError, LookupError):
+            raise MultipartParseError(
+                description='invalid text or charset: {}'.format(charset))
+
+    data = property(get_data)
+    media = property(get_media)
+    text = property(get_text)
+
+
+class MultipartForm:
+
+    def __init__(self, stream, boundary, content_length, parse_options):
+        self._stream = stream if isinstance(stream, BufferedReader) else BufferedReader(stream)
+        self._boundary = boundary
+        # NOTE(vytas): Here self._dash_boundary is not prepended with CRLF
+        #   (yet) for parsing the prologue. The CRLF will be prepended later to
+        #   construct the inter-part delimiter as per RFC 7578, section 4.1
+        #   (see the note below).
+        self._dash_boundary = b'--' + boundary
+        self._parse_options = parse_options
+
+    def __aiter__(self):
+        return self._iterate_parts()
+
+    async def _iterate_parts(self):
+        prologue = True
+        delimiter = self._dash_boundary
+        stream = self._stream
+        max_headers_size = self._parse_options.max_body_part_headers_size
+        remaining_parts = self._parse_options.max_body_part_count
+
+        while True:
+            # NOTE(vytas): Either exhaust the unused stream part, or skip
+            #   the prologue.
+            try:
+                await stream.pipe_until(delimiter, consume_delimiter=True)
+
+                if prologue:
+                    # NOTE(vytas): RFC 7578, section 4.1.
+                    #   As with other multipart types, the parts are delimited
+                    #   with a boundary delimiter, constructed using CRLF,
+                    #   "--", and the value of the "boundary" parameter.
+                    delimiter = _CRLF + delimiter
+                    prologue = False
+
+                separator = await stream.read_until(
+                    _CRLF, 2, consume_delimiter=True)
+                if separator == b'--':
+                    # NOTE(vytas): boundary delimiter + '--\r\n' signals the
+                    # end of a multipart form.
+                    break
+                elif separator:
+                    raise MultipartParseError(
+                        description='unexpected form structure')
+
+            except DelimiterError:
+                raise MultipartParseError(
+                    description='unexpected form structure')
+
+            headers = {}
+            try:
+                headers_block = await stream.read_until(
+                    _CRLF_CRLF, max_headers_size, consume_delimiter=True)
+            except DelimiterError:
+                raise MultipartParseError(
+                    description='incomplete body part headers')
+
+            for line in headers_block.split(_CRLF):
+                name, sep, value = line.partition(b': ')
+                if sep:
+                    name = name.lower()
+
+                    # NOTE(vytas): RFC 7578, section 4.5.
+                    #   This use is deprecated for use in contexts that support
+                    #   binary data such as HTTP. Senders SHOULD NOT generate
+                    #   any parts with a Content-Transfer-Encoding header
+                    #   field.
+                    #
+                    #   Currently, no deployed implementations that send such
+                    #   bodies have been discovered.
+                    if name == b'content-transfer-encoding':
+                        raise MultipartParseError(
+                            description=(
+                                'the deprecated Content-Transfer-Encoding '
+                                'header field is unsupported'))
+                    # NOTE(vytas): RFC 7578, section 4.8.
+                    #   Other header fields MUST NOT be included and MUST be
+                    #   ignored.
+                    elif name in _ALLOWED_CONTENT_HEADERS:
+                        headers[name] = value
+
+            remaining_parts -= 1
+            if remaining_parts < 0 < self._parse_options.max_body_part_count:
+                raise MultipartParseError(
+                    description='maximum number of form body parts exceeded')
+
+            yield BodyPart(stream.delimit(delimiter), headers,
+                           self._parse_options)
diff --git a/falcon/asgi/reader.py b/falcon/asgi/reader.py
new file mode 100644
index 0000000..49712d3
--- /dev/null
+++ b/falcon/asgi/reader.py
@@ -0,0 +1,338 @@
+# Copyright 2019-2020 by Vytautas Liuolia.
+#
+# 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.
+
+"""Buffered ASGI stream reader."""
+
+import io
+
+from falcon.errors import DelimiterError, OperationNotAllowed
+
+DEFAULT_CHUNK_SIZE = 8192
+"""Default minimum chunk size for :class:`BufferedReader` (8 KiB)."""
+
+_MAX_JOIN_CHUNKS = 1024
+
+
+class BufferedReader:
+    """File-like input object wrapping asynchronous iterator over bytes.
+
+    This class implements coroutine functions for asynchronous reading or
+    iteration, but otherwise provides an interface similar to that defined by
+    :class:`io.IOBase`.
+    """
+
+    __slots__ = [
+        '_buffer',
+        '_buffer_len',
+        '_buffer_pos',
+        '_chunk_size',
+        '_consumed',
+        '_exhausted',
+        '_iteration_started',
+        '_max_join_size',
+        '_source',
+    ]
+
+    def __init__(self, source, chunk_size=None):
+        self._source = self._iter_normalized(source)
+        self._chunk_size = chunk_size or DEFAULT_CHUNK_SIZE
+        self._max_join_size = self._chunk_size * _MAX_JOIN_CHUNKS
+
+        self._buffer = b''
+        self._buffer_len = 0
+        self._buffer_pos = 0
+        self._consumed = 0
+        self._exhausted = False
+        self._iteration_started = False
+
+    async def _iter_normalized(self, source):
+        chunk = b''
+        chunk_size = self._chunk_size
+
+        async for item in source:
+            chunk_len = len(chunk)
+            if chunk_len >= chunk_size:
+                self._consumed += chunk_len
+                yield chunk
+                chunk = item
+                continue
+
+            chunk += item
+
+        if chunk:
+            self._consumed += len(chunk)
+            yield chunk
+
+        self._exhausted = True
+
+    async def _iter_with_buffer(self, size_hint=0):
+        if self._buffer_len > self._buffer_pos:
+            if 0 < size_hint < self._buffer_len - self._buffer_pos:
+                buffer_pos = self._buffer_pos
+                self._buffer_pos += size_hint
+                yield self._buffer[buffer_pos:self._buffer_pos]
+
+            buffer_pos = self._buffer_pos
+            self._buffer_pos = self._buffer_len
+            yield self._buffer[buffer_pos:self._buffer_len]
+
+        async for chunk in self._source:
+            yield chunk
+
+    async def _iter_delimited(self, delimiter, size_hint=0):
+        delimiter_len_1 = len(delimiter) - 1
+        if not 0 <= delimiter_len_1 < self._chunk_size:
+            raise ValueError('delimiter length must be within [1, chunk_size]')
+
+        if self._buffer_len > self._buffer_pos:
+            pos = self._buffer.find(delimiter, self._buffer_pos)
+            if pos == 0:
+                return
+            if pos > 0:
+                if 0 < size_hint < pos - self._buffer_pos:
+                    buffer_pos = self._buffer_pos
+                    self._buffer_pos += size_hint
+                    yield self._buffer[buffer_pos:self._buffer_pos]
+                buffer_pos = self._buffer_pos
+                self._buffer_pos = pos
+                yield self._buffer[buffer_pos:pos]
+                return
+
+            if 0 < size_hint < (
+                    self._buffer_len - self._buffer_pos - delimiter_len_1):
+                buffer_pos = self._buffer_pos
+                self._buffer_pos += size_hint
+                yield self._buffer[buffer_pos:self._buffer_pos]
+
+        if self._buffer_pos > 0:
+            self._trim_buffer()
+
+        async for chunk in self._source:
+            offset = self._buffer_len - delimiter_len_1
+            if offset > 0:
+                fragment = self._buffer[offset:] + chunk[:delimiter_len_1]
+                pos = fragment.find(delimiter)
+                if pos < 0:
+                    output = self._buffer
+                    self._buffer = chunk
+                    self._buffer_len = len(chunk)
+                    yield output
+                else:
+                    self._buffer += chunk
+                    self._buffer_len += len(chunk)
+                    self._buffer_pos = offset + pos
+                    # PERF(vytas): local1 + local2 is faster than self._attr
+                    #   (still true on CPython 3.8)
+                    yield self._buffer[:offset + pos]
+                    return
+            elif self._buffer:
+                self._buffer += chunk
+                self._buffer_len += len(chunk)
+            else:
+                self._buffer = chunk
+                self._buffer_len = len(chunk)
+
+            pos = self._buffer.find(delimiter)
+            if pos >= 0:
+                if pos > 0:
+                    self._buffer_pos = pos
+                    yield self._buffer[:pos]
+                return
+
+        yield self._buffer
+
+    async def _consume_delimiter(self, delimiter):
+        delimiter_len = len(delimiter)
+        if await self.peek(delimiter_len) != delimiter:
+            raise DelimiterError('expected delimiter missing')
+        self._buffer_pos += delimiter_len
+
+    def _prepend_buffer(self, chunk):
+        if self._buffer_len > self._buffer_pos:
+            self._buffer = chunk + self._buffer[self._buffer_pos:]
+            self._buffer_len = len(self._buffer)
+        else:
+            self._buffer = chunk
+            self._buffer_len = len(chunk)
+
+        self._buffer_pos = 0
+
+    def _trim_buffer(self):
+        self._buffer = self._buffer[self._buffer_pos:]
+        self._buffer_len -= self._buffer_pos
+        self._buffer_pos = 0
+
+    async def _read_from(self, source, size=-1):
+        if size == -1 or size is None:
+            result = io.BytesIO()
+            async for chunk in source:
+                result.write(chunk)
+            return result.getvalue()
+
+        if size <= 0:
+            return b''
+
+        remaining = size
+
+        if size <= self._max_join_size:
+            result = []
+            async for chunk in source:
+                chunk_len = len(chunk)
+                if remaining < chunk_len:
+                    result.append(chunk[:remaining])
+                    self._prepend_buffer(chunk[remaining:])
+                    break
+
+                result.append(chunk)
+                remaining -= chunk_len
+                if remaining == 0:
+                    break
+
+            # PERF(vytas) Don't join unless necessary.
+            return result[0] if len(result) == 1 else b''.join(result)
+
+        # NOTE(vytas): size > self._max_join_size
+        result = io.BytesIO()
+        async for chunk in source:
+            chunk_len = len(chunk)
+            if remaining < chunk_len:
+                result.write(chunk[:remaining])
+                self._prepend_buffer(chunk[remaining:])
+                break
+
+            result.write(chunk)
+            remaining -= chunk_len
+            if remaining == 0:
+                break
+
+        return result.getvalue()
+
+    def delimit(self, delimiter):
+        return type(self)(self._iter_delimited(delimiter),
+                          chunk_size=self._chunk_size)
+
+    # -------------------------------------------------------------------------
+    # Asynchronous IO interface.
+    # -------------------------------------------------------------------------
+
+    def __aiter__(self):
+        if self._iteration_started:
+            raise OperationNotAllowed(
+                'This stream is already being iterated over.')
+
+        self._iteration_started = True
+
+        # PERF(vytas): Do not wrap unless needed.
+        if self._buffer_len > self._buffer_pos:
+            return self._iter_with_buffer()
+        return self._source
+
+    async def exhaust(self):
+        await self.pipe()
+
+    async def peek(self, size=-1):
+        if size < 0 or size > self._chunk_size:
+            size = self._chunk_size
+
+        if self._buffer_pos > 0:
+            self._trim_buffer()
+
+        if self._buffer_len < size:
+            async for chunk in self._source:
+                self._buffer += chunk
+                self._buffer_len = len(self._buffer)
+                if self._buffer_len >= size:
+                    break
+
+        return self._buffer[:size]
+
+    async def pipe(self, destination=None):
+        async for chunk in self._iter_with_buffer():
+            if destination is not None:
+                await destination.write(chunk)
+
+    async def pipe_until(self, delimiter, destination=None,
+                         consume_delimiter=False):
+        async for chunk in self._iter_delimited(delimiter):
+            if destination is not None:
+                await destination.write(chunk)
+
+        if consume_delimiter:
+            await self._consume_delimiter(delimiter)
+
+    async def read(self, size=-1):
+        return await self._read_from(
+            self._iter_with_buffer(size_hint=size or 0), size)
+
+    async def readall(self):
+        """Read and return all remaining data in the request body.
+
+        Warning:
+            Only use this method when you can be certain that you have
+            enough free memory for the entire request body, and that you
+            have configured your web server to limit request bodies to a
+            reasonable size (to guard against malicious requests).
+
+        Returns:
+            bytes: The request body data, or ``b''`` if the body is empty or
+            has already been consumed.
+        """
+        return await self._read_from(self._iter_with_buffer())
+
+    async def read_until(self, delimiter, size=-1, consume_delimiter=False):
+        result = await self._read_from(
+            self._iter_delimited(delimiter, size_hint=size or 0), size)
+
+        if consume_delimiter:
+            await self._consume_delimiter(delimiter)
+
+        return result
+
+    # -------------------------------------------------------------------------
+    # These methods are included to improve compatibility with Python's
+    #   standard "file-like" IO interface.
+    # -------------------------------------------------------------------------
+
+    # TODO(vytas): Implement the same close() machinery as in asgi.stream?
+    # def close(self):
+    #     pass
+
+    @property
+    def eof(self):
+        """Whether the stream is at EOF."""
+        return self._exhausted and self._buffer_len == self._buffer_pos
+
+    def fileno(self):
+        """Raise an instance of OSError since a file descriptor is not used."""
+        raise OSError('This IO object does not use a file descriptor')
+
+    def isatty(self):
+        """Always returns ``False``."""
+        return False
+
+    def readable(self):
+        """Always returns ``True``."""
+        return True
+
+    def seekable(self):
+        """Always returns ``False``."""
+        return False
+
+    def writable(self):
+        """Always returns ``False``."""
+        return False
+
+    def tell(self):
+        """Returns the number of bytes read from the stream so far."""
+        return self._consumed - (self._buffer_len - self._buffer_pos)
diff --git a/falcon/asgi/request.py b/falcon/asgi/request.py
index 975c3a7..a42056b 100644
--- a/falcon/asgi/request.py
+++ b/falcon/asgi/request.py
@@ -717,7 +717,8 @@ class Request(falcon.request.Request):
                 self.content_length
             )
         finally:
-            await self.stream.exhaust()
+            if handler.exhaust_stream:
+                await self.stream.exhaust()
 
         return self._media
 
diff --git a/falcon/cyutil/reader.pyx b/falcon/cyutil/reader.pyx
index 46be840..039e72a 100644
--- a/falcon/cyutil/reader.pyx
+++ b/falcon/cyutil/reader.pyx
@@ -19,7 +19,8 @@ from libc.stdint cimport uint32_t
 import functools
 import io
 
-from falcon.util.reader import DEFAULT_CHUNK_SIZE, DelimiterError, _MAX_JOIN_CHUNKS
+from falcon.errors import DelimiterError
+from falcon.util.reader import DEFAULT_CHUNK_SIZE, _MAX_JOIN_CHUNKS
 
 
 cdef class BufferedReader:
diff --git a/falcon/errors.py b/falcon/errors.py
index a9f76f1..abc26ae 100644
--- a/falcon/errors.py
+++ b/falcon/errors.py
@@ -37,9 +37,10 @@ package namespace::
 
 from datetime import datetime
 
-from falcon import util
 from falcon.http_error import HTTPError
 import falcon.status_codes as status
+from falcon.util.deprecation import deprecated_args
+from falcon.util.misc import dt_to_http
 
 
 class HeaderNotSupported(ValueError):
@@ -64,6 +65,10 @@ class OperationNotAllowed(ValueError):
     """The requested operation is not allowed."""
 
 
+class DelimiterError(IOError):
+    """The read operation did not find the requested stream delimiter."""
+
+
 class HTTPBadRequest(HTTPError):
     """400 Bad Request.
 
@@ -109,7 +114,7 @@ class HTTPBadRequest(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_400,
@@ -186,7 +191,7 @@ class HTTPUnauthorized(HTTPError):
 
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, challenges=None, **kwargs):
         if challenges:
             headers = _load_headers(headers)
@@ -257,7 +262,7 @@ class HTTPForbidden(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_403,
@@ -322,7 +327,7 @@ class HTTPNotFound(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_404,
@@ -442,7 +447,7 @@ class HTTPMethodNotAllowed(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=1)
+    @deprecated_args(allowed_positional=1)
     def __init__(self, allowed_methods, title=None, description=None, headers=None, **kwargs):
         headers = _load_headers(headers)
         headers['Allow'] = ', '.join(allowed_methods)
@@ -507,7 +512,7 @@ class HTTPNotAcceptable(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_406,
@@ -574,7 +579,7 @@ class HTTPConflict(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_409,
@@ -647,7 +652,7 @@ class HTTPGone(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_410,
@@ -705,7 +710,7 @@ class HTTPLengthRequired(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_411,
@@ -764,7 +769,7 @@ class HTTPPreconditionFailed(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_412,
@@ -833,7 +838,7 @@ class HTTPPayloadTooLarge(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, retry_after=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_413,
@@ -897,7 +902,7 @@ class HTTPUriTooLong(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_414,
@@ -956,7 +961,7 @@ class HTTPUnsupportedMediaType(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_415,
@@ -1028,7 +1033,7 @@ class HTTPRangeNotSatisfiable(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=1)
+    @deprecated_args(allowed_positional=1)
     def __init__(self, resource_length, title=None, description=None, headers=None, **kwargs):
         headers = _load_headers(headers)
         headers['Content-Range'] = 'bytes */' + str(resource_length)
@@ -1092,7 +1097,7 @@ class HTTPUnprocessableEntity(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_422,
@@ -1148,7 +1153,7 @@ class HTTPLocked(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_423,
@@ -1203,7 +1208,7 @@ class HTTPFailedDependency(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_424,
@@ -1266,7 +1271,7 @@ class HTTPPreconditionRequired(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_428,
@@ -1334,7 +1339,7 @@ class HTTPTooManyRequests(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, retry_after=None, **kwargs):
         super().__init__(
             status.HTTP_429,
@@ -1396,7 +1401,7 @@ class HTTPRequestHeaderFieldsTooLarge(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_431,
@@ -1465,7 +1470,7 @@ class HTTPUnavailableForLegalReasons(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_451,
@@ -1520,7 +1525,7 @@ class HTTPInternalServerError(HTTPError):
 
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_500,
@@ -1582,7 +1587,7 @@ class HTTPNotImplemented(HTTPError):
 
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_501,
@@ -1637,7 +1642,7 @@ class HTTPBadGateway(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_502,
@@ -1708,7 +1713,7 @@ class HTTPServiceUnavailable(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, retry_after=None, **kwargs):
         super().__init__(
             status.HTTP_503,
@@ -1764,7 +1769,7 @@ class HTTPGatewayTimeout(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_504,
@@ -1825,7 +1830,7 @@ class HTTPVersionNotSupported(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_505,
@@ -1884,7 +1889,7 @@ class HTTPInsufficientStorage(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_507,
@@ -1940,7 +1945,7 @@ class HTTPLoopDetected(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_508,
@@ -2008,7 +2013,7 @@ class HTTPNetworkAuthenticationRequired(HTTPError):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=0)
+    @deprecated_args(allowed_positional=0)
     def __init__(self, title=None, description=None, headers=None, **kwargs):
         super().__init__(
             status.HTTP_511,
@@ -2061,7 +2066,7 @@ class HTTPInvalidHeader(HTTPBadRequest):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=2)
+    @deprecated_args(allowed_positional=2)
     def __init__(self, msg, header_name, headers=None, **kwargs):
         description = 'The value provided for the "{0}" header is invalid. {1}'
         description = description.format(header_name, msg)
@@ -2115,7 +2120,7 @@ class HTTPMissingHeader(HTTPBadRequest):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=1)
+    @deprecated_args(allowed_positional=1)
     def __init__(self, header_name, headers=None, **kwargs):
         description = 'The "{0}" header is required.'
         description = description.format(header_name)
@@ -2172,7 +2177,7 @@ class HTTPInvalidParam(HTTPBadRequest):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=2)
+    @deprecated_args(allowed_positional=2)
     def __init__(self, msg, param_name, headers=None, **kwargs):
         description = 'The "{0}" parameter is invalid. {1}'
         description = description.format(param_name, msg)
@@ -2228,7 +2233,7 @@ class HTTPMissingParam(HTTPBadRequest):
             base articles related to this error (default ``None``).
     """
 
-    @util.deprecated_args(allowed_positional=1)
+    @deprecated_args(allowed_positional=1)
     def __init__(self, param_name, headers=None, **kwargs):
         description = 'The "{0}" parameter is required.'
         description = description.format(param_name)
@@ -2260,7 +2265,7 @@ def _parse_retry_after(headers, retry_after):
         return headers
     headers = _load_headers(headers)
     if isinstance(retry_after, datetime):
-        headers['Retry-After'] = util.dt_to_http(retry_after)
+        headers['Retry-After'] = dt_to_http(retry_after)
     else:
         headers['Retry-After'] = str(retry_after)
     return headers
diff --git a/falcon/http_error.py b/falcon/http_error.py
index ee2c2e4..d8b35ee 100644
--- a/falcon/http_error.py
+++ b/falcon/http_error.py
@@ -17,7 +17,8 @@
 from collections import OrderedDict
 import xml.etree.ElementTree as et
 
-from falcon.util import deprecated, deprecated_args, json, uri
+from falcon.util import json, uri
+from falcon.util.deprecation import deprecated, deprecated_args
 
 
 class HTTPError(Exception):
diff --git a/falcon/media/multipart.py b/falcon/media/multipart.py
index d69d526..ff4ca4d 100644
--- a/falcon/media/multipart.py
+++ b/falcon/media/multipart.py
@@ -21,8 +21,9 @@ from urllib.parse import unquote_to_bytes
 from falcon import errors
 from falcon import request_helpers
 from falcon.media.base import BaseHandler
-from falcon.util import BufferedReader, DelimiterError
-from falcon.util import deprecated_args, misc
+from falcon.util import BufferedReader
+from falcon.util import misc
+from falcon.util.deprecation import deprecated_args
 
 # TODO(vytas):
 #   * Better support for form-wide charset setting
@@ -71,18 +72,26 @@ class BodyPart:
         content_type (str): Value of the Content-Type header, or the multipart
             form default ``text/plain`` if the header is missing.
 
-        data (bytes): Body part content bytes. The maximum number of bytes that
-            could be read is configurable via :class:`MultipartParseOptions`,
-            and a :class:`.MultipartParseError` is raised if the body part is
-            larger that this size.
+        data (bytes): Property that acts as a convenience alias for
+            :meth:`~.get_data`.
 
-            For large bodies, such as attached files, use the input `stream`
-            directly.
 
-            .. note::
-               Accessing this property the first time would consume the part
-               input stream.
-               The value is cached for subsequent access.
+            .. tabs::
+
+                .. tab:: WSGI
+
+                    .. code:: python
+
+                        # Equivalent to: content = part.get_data()
+                        content = part.data
+
+                .. tab:: ASGI
+
+                    The ``await`` keyword must still be added when referencing
+                    the property::
+
+                        # Equivalent to: content = await part.get_data()
+                        content = await part.data
 
         name(str): The name parameter of the Content-Disposition header.
             The value of the "name" parameter is the original field name from
@@ -114,31 +123,79 @@ class BodyPart:
             to the server's data stream and is non-seekable. The stream is
             automatically delimited according to the multipart stream boundary.
 
-        media (object): Returns a deserialized form of the multipart body part.
-            When called, it will attempt to deserialize the body part stream
-            using the Content-Type header as well as the media-type handlers
-            configured via :class:`MultipartParseOptions`.
+            With the exception of being buffered to keep track of the boundary,
+            the wrapped body part stream interface and behavior mimic
+            :attr:`Request.bounded_stream <falcon.Request.bounded_stream>`
+            (WSGI) and :attr:`Request.stream <falcon.asgi.Request.stream>`
+            (ASGI), respectively:
 
+            .. tabs::
 
-        text (str): The part decoded as a text string provided the part is
-            encoded as ``text/plain``, ``None`` otherwise.
+                .. tab:: WSGI
 
-            Text is decoded from `data` using the charset specified in the
-            `Content-Type` header, or, if omitted, the
-            :data:`default charset <MultipartParseOptions.default_charset>`.
-            The charset must be supported by Python's ``bytes.decode()``
-            function. The list of standard encodings (charsets) supported by
-            the Python 3 standard library can be found `here
-            <https://docs.python.org/3/library/codecs.html#standard-encodings>`__.
+                    Reading the whole part content:
 
-            If decoding fails due to invalid `data` bytes (for the specified
-            encoding), or the specified encoding itself is unsupported, a
-            :class:`MultipartParseError` will be raised when referencing this
-            property.
+                    .. code:: python
 
-            .. note::
-               As this property builds upon `data`, it would consume the part
-               input stream in the same way.
+                        data = part.stream.read()
+
+                    This is also safe:
+
+                    .. code:: python
+
+                        doc = yaml.safe_load(part.stream)
+
+                .. tab:: ASGI
+
+                    Similarly to
+                    :attr:`BoundedStream <falcon.asgi.BoundedStream>`, the most
+                    efficient way to read the body part content is asynchronous
+                    iteration over part data chunks:
+
+                    .. code:: python
+
+                        async for data_chunk in part.stream:
+                            pass
+
+        media (object): Property that acts as a convenience alias for
+            :meth:`~.get_media`.
+
+            .. tabs::
+
+                .. tab:: WSGI
+
+                    .. code:: python
+
+                        # Equivalent to: deserialized_media = part.get_media()
+                        deserialized_media = req.media
+
+                .. tab:: ASGI
+
+                    The ``await`` keyword must still be added when referencing
+                    the property::
+
+                        # Equivalent to: deserialized_media = await part.get_media()
+                        deserialized_media = await part.media
+
+        text (str): Property that acts as a convenience alias for
+            :meth:`~.get_text`.
+
+            .. tabs::
+
+                .. tab:: WSGI
+
+                    .. code:: python
+
+                        # Equivalent to: decoded_text = part.get_text()
+                        decoded_text = part.text
+
+                .. tab:: ASGI
+
+                    The ``await`` keyword must still be added when referencing
+                    the property::
+
+                        # Equivalent to: decoded_text = await part.get_text()
+                        decoded_text = await part.text
     """
 
     _content_disposition = None
@@ -152,8 +209,27 @@ class BodyPart:
         self._headers = headers
         self._parse_options = parse_options
 
-    @property
-    def data(self):
+    def get_data(self):
+        """Returns the body part content bytes.
+
+        The maximum number of bytes that may be read is configurable via
+        :class:`MultipartParseOptions`, and a :class:`.MultipartParseError` is
+        raised if the body part is larger that this size.
+
+        The size limit guards against reading unexpectedly large amount of data
+        into memory by referencing :attr:`data` and :attr:`text` properties
+        that build upon this method.
+        For large bodies, such as attached files, use the input :attr:`stream`
+        directly.
+
+        Note:
+            Calling this method the first time will consume the part's input
+            stream. The result is cached for subsequent access, and follow-up
+            calls will just retrieve the cached content.
+
+        Returns:
+            bytes: The body part content.
+        """
         if self._data is None:
             max_size = self._parse_options.max_body_part_buffer_size + 1
             self._data = self.stream.read(max_size)
@@ -162,8 +238,31 @@ class BodyPart:
 
         return self._data
 
-    @property
-    def text(self):
+    def get_text(self):
+        """Returns the body part content decoded as a text string.
+
+        Text is decoded from the part content (as returned by
+        :meth:`~.get_data`) using the charset specified in the `Content-Type`
+        header, or, if omitted, the
+        :data:`default charset <MultipartParseOptions.default_charset>`.
+        The charset must be supported by Python's ``bytes.decode()``
+        function. The list of standard encodings (charsets) supported by the
+        Python 3 standard library can be found `here
+        <https://docs.python.org/3/library/codecs.html#standard-encodings>`__.
+
+        If decoding fails due to invalid `data` bytes (for the specified
+        encoding), or the specified encoding itself is unsupported, a
+        :class:`MultipartParseError` will be raised when referencing this
+        property.
+
+        Note:
+            As this method builds upon :meth:`~.get_data`, it will consume the
+            part's input stream in the same way.
+
+        Returns:
+            str: The part decoded as a text string provided the part is
+            encoded as ``text/plain``, ``None`` otherwise.
+        """
         content_type, options = cgi.parse_header(self.content_type)
         if content_type != 'text/plain':
             return None
@@ -232,16 +331,47 @@ class BodyPart:
 
         return self._name
 
-    @property
-    def media(self):
+    def get_media(self):
+        """Returns a deserialized form of the multipart body part.
+
+        When called, this method will attempt to deserialize the body part
+        stream using the Content-Type header as well as the media-type handlers
+        configured via :class:`MultipartParseOptions`.
+
+        .. tabs::
+
+            .. tab:: WSGI
+
+                The result will be cached and returned in subsequent calls::
+
+                    deserialized_media = part.get_media()
+
+            .. tab:: ASGI
+
+                The result will be cached and returned in subsequent calls::
+
+                    deserialized_media = await part.get_media()
+
+        Returns:
+            object: The deserialized media representation.
+        """
         if self._media is None:
             handler = self._parse_options.media_handlers.find_by_media_type(
                 self.content_type, 'text/plain')
-            self._media = handler.deserialize(
-                self.stream, self.content_type, None)
+
+            try:
+                self._media = handler.deserialize(
+                    self.stream, self.content_type, None)
+            finally:
+                if handler.exhaust_stream:
+                    self.stream.exhaust()
 
         return self._media
 
+    data = property(get_data)
+    media = property(get_media)
+    text = property(get_text)
+
 
 class MultipartForm:
 
@@ -295,7 +425,7 @@ class MultipartForm:
                     raise MultipartParseError(
                         description='unexpected form structure')
 
-            except DelimiterError:
+            except errors.DelimiterError:
                 raise MultipartParseError(
                     description='unexpected form structure')
 
@@ -303,7 +433,7 @@ class MultipartForm:
             try:
                 headers_block = stream.read_until(
                     _CRLF_CRLF, max_headers_size, consume_delimiter=True)
-            except DelimiterError:
+            except errors.DelimiterError:
                 raise MultipartParseError(
                     description='incomplete body part headers')
 
@@ -354,12 +484,20 @@ class MultipartFormHandler(BaseHandler):
        handler does not consume the stream immediately. Rather, the stream is
        consumed on-demand and parsed into individual body parts while iterating
        over the media object.
+
+    For examples on parsing the request form, see also: :ref:`multipart`.
     """
 
+    _ASGI_MULTIPART_FORM = None
+
     def __init__(self, parse_options=None):
         self.parse_options = parse_options or MultipartParseOptions()
 
-    def deserialize(self, stream, content_type, content_length):
+    def _deserialize_form(self, stream, content_type, content_length,
+                          form_cls=MultipartForm):
+        if not form_cls:
+            raise NotImplementedError
+
         _, options = cgi.parse_header(content_type)
         try:
             boundary = options['boundary']
@@ -383,8 +521,15 @@ class MultipartFormHandler(BaseHandler):
                 'The boundary parameter must consist of 1 to 70 characters',
                 'Content-Type')
 
-        return MultipartForm(stream, boundary.encode(), content_length,
-                             self.parse_options)
+        return form_cls(stream, boundary.encode(), content_length,
+                        self.parse_options)
+
+    def deserialize(self, stream, content_type, content_length):
+        return self._deserialize_form(stream, content_type, content_length)
+
+    async def deserialize_async(self, stream, content_type, content_length):
+        return self._deserialize_form(stream, content_type, content_length,
+                                      form_cls=self._ASGI_MULTIPART_FORM)
 
     def serialize(self, media, content_type):
         raise NotImplementedError('multipart form serialization unsupported')
@@ -397,7 +542,7 @@ class MultipartParseOptions:
 
     Attributes:
         default_charset (str): The default character encoding for
-            :data:`text fields <BodyPart.text>` (default: ``utf-8``).
+            :meth:`text fields <BodyPart.get_text>` (default: ``utf-8``).
 
         max_body_part_count (int): The maximum number of body parts in the form
             (default: 64). If the form contains more parts than this number,
@@ -405,9 +550,9 @@ class MultipartParseOptions:
             option is set to 0, no limit will be imposed by the parser.
 
         max_body_part_buffer_size (int): The maximum number of bytes to buffer
-            and return when the :data:`BodyPart.data` property is
-            referenced (default: 1 MiB). If the body part size exceeds this
-            value, an instance of :class:`MultipartParseError` will be raised.
+            and return when the :meth:`BodyPart.get_data` method is called
+            (default: 1 MiB). If the body part size exceeds this value, an
+            instance of :class:`MultipartParseError` will be raised.
 
         max_body_part_headers_size (int): The maximum size (in bytes) of the
             body part headers structure (default: 8192). If the body part
diff --git a/falcon/util/__init__.py b/falcon/util/__init__.py
index 41250c5..fce326c 100644
--- a/falcon/util/__init__.py
+++ b/falcon/util/__init__.py
@@ -26,8 +26,6 @@ import sys
 
 # Hoist misc. utils
 from falcon.util.misc import *  # NOQA
-from falcon.util.reader import BufferedReader as _PyBufferedReader
-from falcon.util.reader import DelimiterError  # NOQA
 from falcon.util.structures import *  # NOQA
 from falcon.util.sync import *  # NOQA
 from falcon.util.time import *  # NOQA
@@ -44,6 +42,8 @@ if 'samesite' not in _reserved_cookie_attrs:  # pragma: no cover
 
 IS_64_BITS = sys.maxsize > 2**32
 
+from falcon.util.reader import BufferedReader as _PyBufferedReader  # NOQA
+
 try:
     from falcon.cyutil.reader import BufferedReader as _CyBufferedReader
 except ImportError:
diff --git a/falcon/util/deprecation.py b/falcon/util/deprecation.py
new file mode 100644
index 0000000..f1e3dbd
--- /dev/null
+++ b/falcon/util/deprecation.py
@@ -0,0 +1,105 @@
+# Copyright 2013 by Rackspace Hosting, Inc.
+#
+# 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.
+
+"""Miscellaneous deprecation utilities.
+
+This module provides decorators to mark functions and classes as deprecated.
+"""
+
+import functools
+import inspect
+import warnings
+
+
+__all__ = (
+    'DeprecatedWarning',
+    'deprecated',
+    'deprecated_args',
+)
+
+
+# NOTE(kgriffs): We don't want our deprecations to be ignored by default,
+# so create our own type.
+#
+# TODO(kgriffs): Revisit this decision if users complain.
+class DeprecatedWarning(UserWarning):
+    pass
+
+
+def deprecated(instructions, is_property=False):
+    """Flags a method as deprecated.
+
+    This function returns a decorator which can be used to mark deprecated
+    functions. Applying this decorator will result in a warning being
+    emitted when the function is used.
+
+    Args:
+        instructions (str): Specific guidance for the developer, e.g.:
+            'Please migrate to add_proxy(...)'
+        is_property (bool): If the deprecated object is a property. It
+            will omit the ``(...)`` from the generated documentation
+    """
+
+    def decorator(func):
+
+        object_name = 'property' if is_property else 'function'
+        post_name = '' if is_property else '(...)'
+        message = 'Call to deprecated {} {}{}. {}'.format(
+            object_name, func.__name__, post_name, instructions)
+
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            frame = inspect.currentframe().f_back
+
+            warnings.warn_explicit(message,
+                                   category=DeprecatedWarning,
+                                   filename=inspect.getfile(frame.f_code),
+                                   lineno=frame.f_lineno)
+
+            return func(*args, **kwargs)
+
+        return wrapper
+
+    return decorator
+
+
+def deprecated_args(*, allowed_positional, is_method=True):
+    """Flags a method call with positional args as deprecated
+
+    Keyword Args:
+        allowed_positional (int): Number of allowed positional arguments
+        is_method (bool, optional): The decorated function is a method. Will
+          add one to the number of allowed positional args to account for
+          ``self``. Defaults to True.
+    """
+
+    template = (
+        'Calls with{} positional args are deprecated.'
+        ' Please specify them as keyword arguments instead.'
+    )
+    text = ' more than {}'.format(allowed_positional) if allowed_positional else ''
+    warn_text = template.format(text)
+    if is_method:
+        allowed_positional += 1
+
+    def deprecated_args(fn):
+        @functools.wraps(fn)
+        def wraps(*args, **kwargs):
+            if len(args) > allowed_positional:
+                warnings.warn(warn_text, DeprecatedWarning, stacklevel=2)
+            return fn(*args, **kwargs)
+
+        return wraps
+
+    return deprecated_args
diff --git a/falcon/util/misc.py b/falcon/util/misc.py
index e7ba957..0f1b585 100644
--- a/falcon/util/misc.py
+++ b/falcon/util/misc.py
@@ -31,9 +31,11 @@ import inspect
 import re
 import sys
 import unicodedata
-import warnings
 
 from falcon import status_codes
+# NOTE(vytas): Hoist `deprecated` here since it is documented as part of the
+# public Falcon interface.
+from .deprecation import deprecated
 
 try:
     from falcon.cyutil.misc import isascii as _cy_isascii
@@ -52,7 +54,6 @@ __all__ = (
     'get_http_status',
     'http_status_to_code',
     'code_to_http_status',
-    'deprecated_args',
     'secure_filename',
 )
 
@@ -113,51 +114,6 @@ def is_python_func(func):
     return inspect.isfunction(func)
 
 
-# NOTE(kgriffs): We don't want our deprecations to be ignored by default,
-# so create our own type.
-#
-# TODO(kgriffs): Revisit this decision if users complain.
-class DeprecatedWarning(UserWarning):
-    pass
-
-
-def deprecated(instructions, is_property=False):
-    """Flags a method as deprecated.
-
-    This function returns a decorator which can be used to mark deprecated
-    functions. Applying this decorator will result in a warning being
-    emitted when the function is used.
-
-    Args:
-        instructions (str): Specific guidance for the developer, e.g.:
-            'Please migrate to add_proxy(...)'
-        is_property (bool): If the deprecated object is a property. It
-            will omit the ``(...)`` from the generated documentation
-    """
-
-    def decorator(func):
-
-        object_name = 'property' if is_property else 'function'
-        post_name = '' if is_property else '(...)'
-        message = 'Call to deprecated {} {}{}. {}'.format(
-            object_name, func.__name__, post_name, instructions)
-
-        @functools.wraps(func)
-        def wrapper(*args, **kwargs):
-            frame = inspect.currentframe().f_back
-
-            warnings.warn_explicit(message,
-                                   category=DeprecatedWarning,
-                                   filename=inspect.getfile(frame.f_code),
-                                   lineno=frame.f_lineno)
-
-            return func(*args, **kwargs)
-
-        return wrapper
-
-    return decorator
-
-
 def http_now():
     """Returns the current UTC time as an IMF-fixdate.
 
@@ -506,37 +462,6 @@ def code_to_http_status(status):
         return '{} {}'.format(code, _DEFAULT_HTTP_REASON)
 
 
-def deprecated_args(*, allowed_positional, is_method=True):
-    """Flags a method call with positional args as deprecated
-
-    Keyword Args:
-        allowed_positional (int): Number of allowed positional arguments
-        is_method (bool, optional): The decorated function is a method. Will
-          add one to the number of allowed positional args to account for
-          ``self``. Defaults to True.
-    """
-
-    template = (
-        'Calls with{} positional args are deprecated.'
-        ' Please specify them as keyword arguments instead.'
-    )
-    text = ' more than {}'.format(allowed_positional) if allowed_positional else ''
-    warn_text = template.format(text)
-    if is_method:
-        allowed_positional += 1
-
-    def deprecated_args(fn):
-        @functools.wraps(fn)
-        def wraps(*args, **kwargs):
-            if len(args) > allowed_positional:
-                warnings.warn(warn_text, DeprecatedWarning, stacklevel=2)
-            return fn(*args, **kwargs)
-
-        return wraps
-
-    return deprecated_args
-
-
 def _isascii(string):
     """Return ``True`` if all characters in the string are ASCII.
 
diff --git a/falcon/util/reader.py b/falcon/util/reader.py
index f51e3f3..6b66bc3 100644
--- a/falcon/util/reader.py
+++ b/falcon/util/reader.py
@@ -17,16 +17,14 @@
 import functools
 import io
 
+from falcon.errors import DelimiterError
+
 DEFAULT_CHUNK_SIZE = 32768
 """Default chunk size for :class:`BufferedReader` (32 KiB)."""
 
 _MAX_JOIN_CHUNKS = 128
 
 
-class DelimiterError(IOError):
-    pass
-
-
 class BufferedReader:
 
     def __init__(self, read, max_stream_len, chunk_size=None):
diff --git a/falcon/util/sync.py b/falcon/util/sync.py
index bc557b3..d63bc5b 100644
--- a/falcon/util/sync.py
+++ b/falcon/util/sync.py
@@ -8,6 +8,8 @@ from typing import Callable
 
 __all__ = [
     'get_loop',
+    'invoke_coroutine_sync',
+    'runs_sync',
     'sync_to_async',
     'wrap_sync_to_async',
     'wrap_sync_to_async_unsafe',
@@ -180,3 +182,55 @@ def _wrap_non_coroutine_unsafe(func):
         return func
 
     return wrap_sync_to_async_unsafe(func)
+
+
+def invoke_coroutine_sync(coroutine, *args, **kwargs):
+    """Invokes a coroutine function from a synchronous caller and runs until complete.
+
+    Warning:
+        This method is very inefficient and should only be used
+        for testing purposes. It will create an event loop for the current
+        thread if one is not already running.
+
+    Additional arguments not mentioned below are bound to the given
+    coroutine function via ``functools.partial()``.
+
+    Args:
+        coroutine: A coroutine function to invoke.
+        *args: Additional args are passed through to the coroutine function.
+
+    Keyword Args:
+        **kwargs: Additional args are passed through to the coroutine function.
+    """
+
+    loop = asyncio.get_event_loop()
+    return loop.run_until_complete(
+        partial(coroutine, *args, **kwargs)()
+    )
+
+
+def runs_sync(coroutine):
+    """A decorator to transform a coroutine function into a synchronous method.
+
+    This is achieved by always invoking the decorated coroutine function via
+    :meth:`invoke_coroutine_sync`.
+
+    Warning:
+        This decorator is very inefficient and should only be used for adapting
+        asynchronous test functions for use with synchronous test runners such
+        as ``pytest`` or the ``unittest`` module.
+
+        It will create an event loop for the current thread if one is not
+        already running.
+
+    Args:
+        coroutine: A coroutine function to masquerade as a synchronous one.
+
+    Returns:
+        callable: A synchronous function.
+    """
+    @wraps(coroutine)
+    def invoke(*args, **kwargs):
+        return invoke_coroutine_sync(coroutine, *args, **kwargs)
+
+    return invoke
diff --git a/requirements/docs b/requirements/docs
index ac3c718..e59d178 100644
--- a/requirements/docs
+++ b/requirements/docs
@@ -4,6 +4,6 @@ jinja2
 markupsafe
 pygments
 pygments-style-github
-sphinx>=1.4.4,!=1.7.3
+sphinx
 sphinx_rtd_theme
 sphinx-tabs
diff --git a/setup.py b/setup.py
index 62c8c11..bce7098 100644
--- a/setup.py
+++ b/setup.py
@@ -63,8 +63,11 @@ if CYTHON:
         #       * https://github.com/cython/cython/issues/2273
         #       * https://bugs.python.org/issue38225
         #
+        # NOTE(vytas): It is pointless to cythonize reader.py, since cythonized
+        #   Falcon is using reader.pyx instead.
         'falcon.hooks',
         'falcon.responders',
+        'falcon.util.reader',
         'falcon.util.sync',
     ]
 


Test patch:
diff --git a/docs/api/testing.rst b/docs/api/testing.rst
index 4d67942..d2b7ccb 100644
--- a/docs/api/testing.rst
+++ b/docs/api/testing.rst
@@ -15,8 +15,6 @@ General Helpers
 
 .. autoclass:: TestCase
 .. autoclass:: SimpleTestResource
-
-.. autofunction:: invoke_coroutine_sync
 .. autofunction:: rand_string
 .. autofunction:: get_unused_port
 .. autofunction:: redirected
diff --git a/falcon/testing/client.py b/falcon/testing/client.py
index b43cd82..b811b4f 100644
--- a/falcon/testing/client.py
+++ b/falcon/testing/client.py
@@ -35,6 +35,7 @@ from falcon.util import (
     get_loop,
     http_cookies,
     http_date_to_dt,
+    invoke_coroutine_sync,
     json as util_json,
     to_query_str,
 )
@@ -502,7 +503,7 @@ def simulate_request(app, method='GET', path='/', query_string=None,
         await _wait_for_shutdown(lifespan_event_collector.events)
         await t
 
-    helpers.invoke_coroutine_sync(conductor)
+    invoke_coroutine_sync(conductor)
 
     return Result(resp_event_collector.body_chunks,
                   code_to_http_status(resp_event_collector.status),
diff --git a/falcon/testing/helpers.py b/falcon/testing/helpers.py
index ec86557..9bf4a5b 100644
--- a/falcon/testing/helpers.py
+++ b/falcon/testing/helpers.py
@@ -27,7 +27,6 @@ import asyncio
 import cgi
 from collections import defaultdict
 import contextlib
-import functools
 import io
 import itertools
 import random
@@ -289,33 +288,6 @@ class ASGIResponseEventCollector:
     __call__ = collect
 
 
-def invoke_coroutine_sync(coroutine, *args, **kwargs):
-    """Invokes a coroutine function from a synchronous caller and runs until complete.
-
-    Warning:
-        This method is very inefficient and should only be used
-        for testing purposes. It will create an event loop for the current
-        thread if one is not already running.
-
-    Additional arguments not mentioned below are bound to the given
-    coroutine function via ``functools.partial()``.
-
-    Args:
-        coroutine: A coroutine function to invoke.
-        *args: Additional args are passed through to the coroutine function.
-
-    Keyword Args:
-        **kwargs: Additional args are passed through to the coroutine function.
-    """
-
-    loop = asyncio.get_event_loop()
-    return loop.run_until_complete(
-        functools.partial(
-            coroutine, *args, **kwargs
-        )()
-    )
-
-
 # get_encoding_from_headers() is Copyright 2016 Kenneth Reitz, and is
 # used here under the terms of the Apache License, Version 2.0.
 def get_encoding_from_headers(headers):
diff --git a/tests/asgi/_asgi_test_app.py b/tests/asgi/_asgi_test_app.py
index c6dc672..60b2ebd 100644
--- a/tests/asgi/_asgi_test_app.py
+++ b/tests/asgi/_asgi_test_app.py
@@ -1,5 +1,6 @@
 import asyncio
 from collections import Counter
+import hashlib
 import sys
 import time
 
@@ -111,6 +112,26 @@ class Events:
         resp.sse = emit()
 
 
+class Multipart:
+    async def on_post(self, req, resp):
+        parts = {}
+
+        form = await req.get_media()
+        async for part in form:
+            # NOTE(vytas): SHA1 is no longer recommended for cryptographic
+            #   purposes, but here we are only using it for integrity checking.
+            sha1 = hashlib.sha1()
+            async for chunk in part.stream:
+                sha1.update(chunk)
+
+            parts[part.name] = {
+                'filename': part.filename,
+                'sha1': sha1.hexdigest(),
+            }
+
+        resp.media = parts
+
+
 class LifespanHandler:
     def __init__(self):
         self.startup_succeeded = False
@@ -132,6 +153,7 @@ def create_app():
     app.add_route('/', Things())
     app.add_route('/bucket', Bucket())
     app.add_route('/events', Events())
+    app.add_route('/forms', Multipart())
 
     lifespan_handler = LifespanHandler()
     app.add_middleware(lifespan_handler)
diff --git a/tests/asgi/test_asgi_servers.py b/tests/asgi/test_asgi_servers.py
index 58201ea..213357c 100644
--- a/tests/asgi/test_asgi_servers.py
+++ b/tests/asgi/test_asgi_servers.py
@@ -1,6 +1,8 @@
 from contextlib import contextmanager
+import hashlib
 import os
 import platform
+import random
 import subprocess
 import sys
 import time
@@ -19,6 +21,7 @@ _WIN32 = sys.platform.startswith('win')
 
 _SERVER_HOST = '127.0.0.1'
 _SIZE_1_KB = 1024
+_SIZE_1_MB = _SIZE_1_KB ** 2
 
 
 _REQUEST_TIMEOUT = 10
@@ -42,6 +45,29 @@ class TestASGIServer:
         resp = requests.head(server_base_url, data=body, timeout=_REQUEST_TIMEOUT)
         assert resp.status_code == 405
 
+    def test_post_multipart_form(self, server_base_url):
+        size = random.randint(16 * _SIZE_1_MB, 32 * _SIZE_1_MB)
+        data = os.urandom(size)
+        digest = hashlib.sha1(data).hexdigest()
+        files = {
+            'random': ('random.dat', data),
+            'message': ('hello.txt', b'Hello, World!\n'),
+        }
+
+        resp = requests.post(
+            server_base_url + 'forms', files=files, timeout=_REQUEST_TIMEOUT)
+        assert resp.status_code == 200
+        assert resp.json() == {
+            'message': {
+                'filename': 'hello.txt',
+                'sha1': '60fde9c2310b0d4cad4dab8d126b04387efba289',
+            },
+            'random': {
+                'filename': 'random.dat',
+                'sha1': digest,
+            },
+        }
+
     def test_post_multiple(self, server_base_url):
         body = testing.rand_string(_SIZE_1_KB / 2, _SIZE_1_KB)
         resp = requests.post(server_base_url, data=body, timeout=_REQUEST_TIMEOUT)
diff --git a/tests/asgi/test_boundedstream_asgi.py b/tests/asgi/test_boundedstream_asgi.py
index 618bdf6..7d38607 100644
--- a/tests/asgi/test_boundedstream_asgi.py
+++ b/tests/asgi/test_boundedstream_asgi.py
@@ -3,6 +3,7 @@ import time
 
 import pytest
 
+import falcon
 from falcon import asgi, testing
 
 
@@ -108,7 +109,7 @@ def test_read_all(body, extra_body, set_content_length):
         assert s.closed
 
     for t in (test_iteration, test_readall_a, test_readall_b, test_readall_c, test_readall_d):
-        testing.invoke_coroutine_sync(t)
+        falcon.invoke_coroutine_sync(t)
 
 
 def test_filelike():
@@ -145,7 +146,7 @@ def test_filelike():
             async for chunk in s:
                 pass
 
-    testing.invoke_coroutine_sync(test_iteration)
+    falcon.invoke_coroutine_sync(test_iteration)
 
 
 @pytest.mark.parametrize('body', [
@@ -204,8 +205,8 @@ def test_read_chunks(body, chunk_size):
         assert b''.join(chunks) == body
 
     for t in (test_nonmixed, test_mixed_a, test_mixed_b, test_mixed_iter):
-        testing.invoke_coroutine_sync(t)
-        testing.invoke_coroutine_sync(t)
+        falcon.invoke_coroutine_sync(t)
+        falcon.invoke_coroutine_sync(t)
 
 
 def test_exhaust_with_disconnect():
@@ -224,7 +225,24 @@ def test_exhaust_with_disconnect():
         assert await s.read(100) == b''
         assert s.eof
 
-    testing.invoke_coroutine_sync(t)
+    falcon.invoke_coroutine_sync(t)
+
+
+@falcon.runs_sync
+async def test_exhaust():
+    emitter = testing.ASGIRequestEventEmitter(b'123456798' * 1024)
+    stream = asgi.BoundedStream(emitter)
+
+    assert await stream.read(1) == b'1'
+    assert await stream.read(6) == b'234567'
+    assert await stream.read(101) == b'98' + b'123456798' * 11
+
+    await stream.exhaust()
+
+    assert await stream.read(1) == b''
+    assert await stream.read(6) == b''
+    assert await stream.read(101) == b''
+    assert stream.eof
 
 
 def test_iteration_already_started():
@@ -248,7 +266,7 @@ def test_iteration_already_started():
 
         assert b''.join(chunks) == body
 
-    testing.invoke_coroutine_sync(t)
+    falcon.invoke_coroutine_sync(t)
 
 
 def _stream(body, content_length=None):
diff --git a/tests/asgi/test_buffered_reader.py b/tests/asgi/test_buffered_reader.py
new file mode 100644
index 0000000..6ff8fb1
--- /dev/null
+++ b/tests/asgi/test_buffered_reader.py
@@ -0,0 +1,387 @@
+import io
+
+import pytest
+
+import falcon
+from falcon.asgi import reader
+from falcon.errors import OperationNotAllowed
+
+
+async def async_iter(data):
+    for item in data:
+        yield item
+
+
+def async_take(source, count=None):
+    async def collect():
+        result = []
+        async for item in source:
+            result.append(item)
+            if count is not None and count <= len(result):
+                return result
+        return result
+
+    return falcon.invoke_coroutine_sync(collect)
+
+
+async def chop_data(data, min_size=1024, max_size=64 * 1024):
+    index = 0
+    size = min_size
+
+    while True:
+        chunk = data[index:index + size]
+        index += size
+        if not chunk:
+            break
+        yield chunk
+
+        size = min(max_size, size + 1)
+
+
+SOURCE1 = (
+    b'Hello',
+    b'',
+    b'',
+    b',',
+    b' ',
+    b'World!',
+    b'\n',
+    b'Jus',
+    b't tes',
+    b'ting some iterato',
+    b'r goodne',
+    b'',
+    b'ss.',
+    b'',
+    b'\n',
+)
+DATA1 = b''.join(SOURCE1)
+
+DATA2 = (
+    b'123456789ABCDEF\n' * 64 * 1024 * 64 +
+    b'--boundary1234567890--' +
+    b'123456789ABCDEF\n' * 64 * 1024 * 63 +
+    b'--boundary1234567890--' +
+    b'123456789ABCDEF\n' * 64 * 1024 * 62 +
+    b'--boundary1234567890--'
+)
+SOURCE2 = tuple(async_take(chop_data(DATA2)))
+
+SOURCE3 = (
+    b'1' * 1024 * 1024 + b'333',
+    b'2' * 2 * 1024 * 1024 + b'444'
+)
+DATA3 = b''.join(SOURCE3)
+
+
+class AsyncSink:
+
+    def __init__(self):
+        self._sink = io.BytesIO()
+
+    async def write(self, data):
+        self._sink.write(data)
+
+    @property
+    def accumulated(self):
+        return self._sink.getvalue()
+
+
+@pytest.fixture()
+def reader1():
+    return reader.BufferedReader(async_iter(SOURCE1), chunk_size=8)
+
+
+@pytest.fixture()
+def reader2():
+    return reader.BufferedReader(async_iter(SOURCE2), chunk_size=2048)
+
+
+@pytest.fixture()
+def reader3():
+    return reader.BufferedReader(async_iter(SOURCE3), chunk_size=2048)
+
+
+def test_basic_aiter(reader1):
+    assert async_take(reader1) == [
+        b'Hello, World!',
+        b'\nJust tes',
+        b'ting some iterato',
+        b'r goodne',
+        b'ss.\n',
+    ]
+
+
+@falcon.runs_sync
+async def test_aiter_from_buffer(reader1):
+    assert await reader1.read(4) == b'Hell'
+
+    collected = []
+    async for chunk in reader1:
+        collected.append(chunk)
+    assert collected == [
+        b'o, World!',
+        b'\nJust tes',
+        b'ting some iterato',
+        b'r goodne',
+        b'ss.\n',
+    ]
+
+
+@pytest.mark.parametrize('delimiter,expected', [
+    (b'H', []),
+    (b'Hello', []),
+    (b'o', [b'Hell']),
+    (b'ting', [b'Hello, World!', b'\nJust tes']),
+    (
+        b'404',
+        [
+            b'Hello, World!',
+            b'\nJust tes',
+            b'ting some iterato',
+            b'r goodne',
+            b'ss.\n',
+        ],
+    ),
+])
+def test_delimit(reader1, delimiter, expected):
+    delimited = reader1.delimit(delimiter)
+    assert async_take(delimited) == expected
+
+
+@falcon.runs_sync
+async def test_exhaust(reader1):
+    await reader1.exhaust()
+    assert await reader1.peek() == b''
+
+
+@pytest.mark.parametrize('size', [1, 2, 3, 5, 7, 8])
+@falcon.runs_sync
+async def test_peek(reader1, size):
+    assert await reader1.peek(size) == b'Hello, World'[:size]
+    assert reader1.tell() == 0
+
+
+@falcon.runs_sync
+async def test_peek_at_eof():
+    source = chop_data(b'Hello!')
+    stream = reader.BufferedReader(source)
+    assert await stream.peek(16) == b'Hello!'
+
+
+@falcon.runs_sync
+async def test_pipe(reader1):
+    sink = AsyncSink()
+    await reader1.pipe(sink)
+    assert sink.accumulated == DATA1
+    assert reader1.eof
+    assert reader1.tell() == len(sink.accumulated)
+
+
+@falcon.runs_sync
+async def test_pipe_until_delimiter_not_found(reader1):
+    sink = AsyncSink()
+    await reader1.pipe_until(b'404', sink)
+    assert sink.accumulated == DATA1
+
+
+@pytest.mark.parametrize('sizes,expected', [
+    ((0, 1, 2, 5), [b'', b'H', b'el', b'lo, W']),
+    (
+        (20, 21, 22, 23, 25),
+        [
+            b'Hello, World!\nJust t',
+            b'esting some iterator ',
+            b'goodness.\n',
+            b'',
+            b'',
+        ],
+    ),
+    ((1, 50), [b'H', b'ello, World!\nJust testing some iterator goodness.\n']),
+    ((50, 1), [b'Hello, World!\nJust testing some iterator goodness.', b'\n']),
+])
+@falcon.runs_sync
+async def test_read(reader1, sizes, expected):
+    results = []
+    for size in sizes:
+        results.append(await reader1.read(size))
+
+    assert results == expected
+
+
+@pytest.mark.parametrize('start_size', [1, 16777216])
+@falcon.runs_sync
+async def test_varying_read_size(reader2, start_size):
+    size = start_size
+    result = io.BytesIO()
+
+    while True:
+        chunk = await reader2.read(size)
+        if not chunk:
+            break
+
+        result.write(chunk)
+        size += 1
+
+    assert result.getvalue() == DATA2
+
+
+@pytest.mark.parametrize('peek', [0, 1, 8])
+@falcon.runs_sync
+async def test_readall(reader1, peek):
+    if peek:
+        await reader1.peek(peek)
+    assert await reader1.readall() == DATA1
+    assert reader1.eof
+
+
+@pytest.mark.parametrize('fork', [False, True])
+@pytest.mark.parametrize('offset,delimiter,size,expected', [
+    (0, b', ', 4, b'Hell'),
+    (0, b', ', 5, b'Hello'),
+    (0, b', ', -1, b'Hello'),
+    (20, b' ', 4, b'esti'),
+    (20, b' ', 5, b'estin'),
+    (20, b' ', 6, b'esting'),
+    (20, b' ', 20, b'esting'),
+    (20, b' ', None, b'esting'),
+    (0, b'Hell', 13, b''),
+    (1, b'ell', 13, b''),
+    (2, b'll', 13, b''),
+    (3, b'l', 13, b''),
+    (2, b'l', 13, b''),
+    (0, b'good', 13, b'Hello, World!'),
+    (7, b'good', 19, b'World!\nJust testing'),
+    (7, b'good', 33, b'World!\nJust testing some iterator'),
+    (7, b'good', 34, b'World!\nJust testing some iterator '),
+    (7, b'good', 1337, b'World!\nJust testing some iterator '),
+    (7, b'good', -1, b'World!\nJust testing some iterator '),
+])
+@falcon.runs_sync
+async def test_read_until(reader1, offset, delimiter, size, expected, fork):
+    if offset:
+        await reader1.read(offset)
+
+    if fork:
+        assert await reader1.delimit(delimiter).read(size) == expected
+    else:
+        assert await reader1.read_until(delimiter, size) == expected
+
+
+@falcon.runs_sync
+async def test_read_until_with_buffer_edge_case(reader1):
+    assert await reader1.read(12) == b'Hello, World'
+    assert await reader1.peek(1) == b'!'
+    assert await reader1.read_until(b'404', 1) == b'!'
+    assert await reader1.read(13) == b'\nJust testing'
+
+
+def test_placeholder_methods(reader1):
+    with pytest.raises(OSError):
+        reader1.fileno()
+
+    assert not reader1.isatty()
+    assert reader1.readable()
+    assert not reader1.seekable()
+    assert not reader1.writable()
+
+
+@falcon.runs_sync
+async def test_iteration_started(reader1):
+    async for chunk in reader1:
+        with pytest.raises(OperationNotAllowed):
+            async for nested in reader1:
+                pass
+
+
+@falcon.runs_sync
+async def test_invalid_delimiter_length(reader1):
+    with pytest.raises(ValueError):
+        await reader1.read_until(b'')
+
+    with pytest.raises(ValueError):
+        await reader1.pipe_until(b'')
+
+    with pytest.raises(ValueError):
+        await reader1.delimit(b'').read()
+
+
+@pytest.mark.parametrize('size1,size2', [
+    (11003077, 22000721),
+    (13372477, 51637898),
+])
+@falcon.runs_sync
+async def test_irregular_large_read_until(reader2, size1, size2):
+    delimiter = b'--boundary1234567890--'
+
+    await reader2.pipe_until(delimiter, consume_delimiter=True)
+    await reader2.pipe_until(delimiter, consume_delimiter=True)
+
+    expected = b'123456789ABCDEF\n' * 64 * 1024 * 62
+
+    assert await reader2.read_until(delimiter, 1337) == expected[:1337]
+
+    chunk1 = await reader2.read_until(delimiter, size1)
+    assert len(chunk1) == size1
+
+    chunk2 = await reader2.read_until(delimiter, size2)
+    assert len(chunk2) == size2
+
+    remainder = await reader2.read_until(delimiter, 62 * 1024 * 1024)
+    assert chunk1 + chunk2 + remainder == expected[1337:]
+
+
+@falcon.runs_sync
+@pytest.mark.parametrize('chunk_size', list(range(46, 63)))
+async def test_read_until_shared_boundary(chunk_size):
+    source = chop_data(
+        b'-boundary-like-' * 4 + b'--some junk--\n' + b'\n' * 1024,
+        min_size=chunk_size,
+        max_size=chunk_size)
+    stream = reader.BufferedReader(source, chunk_size)
+    assert await stream.read_until(b'-boundary-like---') == (
+        b'-boundary-like-' * 3)
+    assert await stream.peek(17) == b'-boundary-like---'
+
+
+# NOTE(vytas): This is woefully unoptimized, and this test highlights that.
+#   Work in progress.
+@falcon.runs_sync
+async def test_small_reads(reader3):
+    ops = 0
+    read = 0
+    last = b''
+    size = 0
+
+    while True:
+        size = max(1, (size + ops) % 137)
+        chunk = await reader3.read(size)
+        if not chunk:
+            break
+
+        ops += 1
+        read += len(chunk)
+        last = chunk
+
+    assert ops == 50740
+    assert read == len(DATA3)
+    assert last.endswith(b'4')
+
+
+@falcon.runs_sync
+async def test_small_reads_with_delimiter(reader3):
+    ops = 0
+    read = 0
+    size = 0
+
+    while True:
+        size = max(1, (size + ops) % 1337)
+        chunk = await reader3.read_until(b'33', size)
+        assert chunk.strip(b'1') == b''
+        if not chunk:
+            break
+
+        ops += 1
+        read += len(chunk)
+
+    assert read == 1024 * 1024
diff --git a/tests/asgi/test_scope.py b/tests/asgi/test_scope.py
index f3d6035..71c21b5 100644
--- a/tests/asgi/test_scope.py
+++ b/tests/asgi/test_scope.py
@@ -2,6 +2,7 @@ import asyncio
 
 import pytest
 
+import falcon
 from falcon import testing
 from falcon.asgi import App
 from falcon.errors import UnsupportedScopeError
@@ -107,7 +108,7 @@ def test_lifespan_scope_default_version():
 
         await t
 
-    testing.invoke_coroutine_sync(t)
+    falcon.invoke_coroutine_sync(t)
 
     assert not resource.called
 
@@ -139,7 +140,7 @@ def test_lifespan_scope_version(spec_version, supported):
 
     if not supported:
         with pytest.raises(UnsupportedScopeError):
-            testing.invoke_coroutine_sync(
+            falcon.invoke_coroutine_sync(
                 app.__call__, scope, req_event_emitter, resp_event_collector
             )
 
@@ -158,7 +159,7 @@ def test_lifespan_scope_version(spec_version, supported):
 
         await t
 
-    testing.invoke_coroutine_sync(t)
+    falcon.invoke_coroutine_sync(t)
 
 
 def test_query_string_values():
@@ -210,7 +211,7 @@ def _call_with_scope(scope):
     req_event_emitter = testing.ASGIRequestEventEmitter()
     resp_event_collector = testing.ASGIResponseEventCollector()
 
-    testing.invoke_coroutine_sync(
+    falcon.invoke_coroutine_sync(
         app.__call__, scope, req_event_emitter, resp_event_collector
     )
 
diff --git a/tests/test_after_hooks.py b/tests/test_after_hooks.py
index 0905047..1d26b37 100644
--- a/tests/test_after_hooks.py
+++ b/tests/test_after_hooks.py
@@ -297,7 +297,7 @@ def test_resource_with_uri_fields_async():
         await resource.on_get(req, resp, '1', '2')
         assert resource.fields == ('1', '2')
 
-    testing.invoke_coroutine_sync(test_direct)
+    falcon.invoke_coroutine_sync(test_direct)
 
 
 @pytest.mark.parametrize(
diff --git a/tests/test_before_hooks.py b/tests/test_before_hooks.py
index 8a8aa0f..7376f6f 100644
--- a/tests/test_before_hooks.py
+++ b/tests/test_before_hooks.py
@@ -385,7 +385,7 @@ def test_parser_async(body, doc):
         await resource.on_get(req, resp, doc)
         assert resource.doc == doc
 
-    testing.invoke_coroutine_sync(test_direct)
+    falcon.invoke_coroutine_sync(test_direct)
 
 
 def test_wrapped_resource(client, wrapped_resource):
diff --git a/tests/test_buffered_reader.py b/tests/test_buffered_reader.py
index f972798..3051121 100644
--- a/tests/test_buffered_reader.py
+++ b/tests/test_buffered_reader.py
@@ -2,7 +2,8 @@ import io
 
 import pytest
 
-from falcon.util import BufferedReader, DelimiterError
+from falcon.errors import DelimiterError
+from falcon.util import BufferedReader
 
 
 class WouldHang(RuntimeError):
diff --git a/tests/test_error.py b/tests/test_error.py
index d7ef36f..fde6e0d 100644
--- a/tests/test_error.py
+++ b/tests/test_error.py
@@ -3,7 +3,7 @@ import pytest
 import falcon
 import falcon.errors as errors
 import falcon.status_codes as status
-from falcon.util.misc import DeprecatedWarning
+from falcon.util.deprecation import DeprecatedWarning
 
 
 @pytest.mark.parametrize('err, title', [
diff --git a/tests/test_httperror.py b/tests/test_httperror.py
index 502ec90..75adc65 100644
--- a/tests/test_httperror.py
+++ b/tests/test_httperror.py
@@ -10,7 +10,8 @@ import yaml
 import falcon
 from falcon.http_error import NoRepresentation, OptionalRepresentation
 import falcon.testing as testing
-from falcon.util import json, misc
+from falcon.util import json
+from falcon.util.deprecation import DeprecatedWarning
 
 from _util import create_app  # NOQA
 
@@ -288,7 +289,7 @@ class TestHTTPError:
         assert response.json == expected_body
 
     def test_has_representation(self):
-        with pytest.warns(misc.DeprecatedWarning, match='has_representation is deprecated'):
+        with pytest.warns(DeprecatedWarning, match='has_representation is deprecated'):
             assert falcon.HTTPError(falcon.HTTP_701).has_representation is True
 
     def test_no_description_json(self, client):
@@ -876,13 +877,13 @@ def test_kw_only():
     # only deprecated for now
     # with pytest.raises(TypeError, match='positional argument'):
     #     falcon.HTTPError(falcon.HTTP_BAD_REQUEST, 'foo', 'bar')
-    with pytest.warns(misc.DeprecatedWarning, match='positional args are deprecated'):
+    with pytest.warns(DeprecatedWarning, match='positional args are deprecated'):
         falcon.HTTPError(falcon.HTTP_BAD_REQUEST, 'foo', 'bar')
 
 
 def test_NoRepresentation():
     with pytest.warns(
-        misc.DeprecatedWarning,
+        DeprecatedWarning,
         match='has_representation is deprecated.*The class NoRepresentation'
     ):
         assert NoRepresentation().has_representation is False
@@ -891,7 +892,7 @@ def test_NoRepresentation():
 class TestOptionalRepresentation:
     def test_OptionalRepresentation_false(self):
         with pytest.warns(
-            misc.DeprecatedWarning,
+            DeprecatedWarning,
             match='has_representation is deprecated.*The class OptionalRepresentation'
         ):
             or_ = OptionalRepresentation()
@@ -900,7 +901,7 @@ class TestOptionalRepresentation:
 
     def test_OptionalRepresentation_true(self):
         with pytest.warns(
-            misc.DeprecatedWarning,
+            DeprecatedWarning,
             match='has_representation is deprecated.*The class OptionalRepresentation'
         ):
             or_ = OptionalRepresentation()
diff --git a/tests/test_media_handlers.py b/tests/test_media_handlers.py
index 4670100..c9fcc83 100644
--- a/tests/test_media_handlers.py
+++ b/tests/test_media_handlers.py
@@ -8,6 +8,7 @@ import mujson
 import pytest
 import ujson
 
+import falcon
 from falcon import ASGI_SUPPORTED, media, testing
 
 from _util import create_app  # NOQA
@@ -90,7 +91,7 @@ def test_serialization(asgi, func, body, expected):
     args = (body, b'application/javacript')
 
     if asgi:
-        result = testing.invoke_coroutine_sync(handler.serialize_async, *args)
+        result = falcon.invoke_coroutine_sync(handler.serialize_async, *args)
     else:
         result = handler.serialize(*args)
 
@@ -114,7 +115,7 @@ def test_deserialization(asgi, func, body, expected):
         s = BoundedStream(testing.ASGIRequestEventEmitter(body))
         args.insert(0, s)
 
-        result = testing.invoke_coroutine_sync(handler.deserialize_async, *args)
+        result = falcon.invoke_coroutine_sync(handler.deserialize_async, *args)
     else:
         args.insert(0, io.BytesIO(body))
         result = handler.deserialize(*args)
diff --git a/tests/test_media_multipart.py b/tests/test_media_multipart.py
index 8683bfb..cafd674 100644
--- a/tests/test_media_multipart.py
+++ b/tests/test_media_multipart.py
@@ -3,6 +3,7 @@ import itertools
 import os
 import random
 
+import msgpack
 import pytest
 
 import falcon
@@ -10,6 +11,8 @@ from falcon import media
 from falcon import testing
 from falcon.util import BufferedReader
 
+from _util import create_app  # NOQA: I100
+
 
 EXAMPLE1 = (
     b'--5b11af82ab65407ba8cdccf37d2a9c4f\r\n'
@@ -191,51 +194,6 @@ def test_serialize():
         handler.serialize({'key': 'value'}, 'multipart/form-data')
 
 
-def test_content_transfer_encoding_header():
-    data = (
-        b'--BOUNDARY\r\n'
-        b'Content-Disposition: form-data; name="file"; filename="bytes"\r\n'
-        b'Content-Transfer-Encoding: Base64'
-        b'Content-Type: application/x-falcon\r\n\r\n'
-        b'UGVyZWdyaW5lIEZhbGNvbiADLgA='
-        b'\r\n'
-        b'--BOUNDARY\r\n'
-        b'Content-Disposition: form-data; name="empty"\r\n'
-        b'Content-Type: text/plain\r\n\r\n'
-        b'\r\n'
-        b'--BOUNDARY--\r\n'
-    )
-
-    handler = media.MultipartFormHandler()
-    form = handler.deserialize(
-        io.BytesIO(data), 'multipart/form-data; boundary=BOUNDARY', len(data))
-    with pytest.raises(falcon.HTTPBadRequest):
-        for part in form:
-            pass
-
-
-def test_unsupported_charset():
-    data = (
-        b'--BOUNDARY\r\n'
-        b'Content-Disposition: form-data; name="text"\r\n'
-        b'Content-Type: text/plain; charset=pecyn\r\n\r\n'
-        b'AAHEHlRoZSBGYWxjb24gV2ViIEZyYW1ld29yaywgMjAxOQ=='
-        b'\r\n'
-        b'--BOUNDARY\r\n'
-        b'Content-Disposition: form-data; name="empty"\r\n'
-        b'Content-Type: text/plain\r\n\r\n'
-        b'\r\n'
-        b'--BOUNDARY--\r\n'
-    )
-
-    handler = media.MultipartFormHandler()
-    form = handler.deserialize(
-        io.BytesIO(data), 'multipart/form-data; boundary=BOUNDARY', len(data))
-    with pytest.raises(falcon.HTTPBadRequest):
-        for part in form:
-            part.text
-
-
 @pytest.mark.parametrize('charset,data', [
     ('utf-8', b'Impossible byte: \xff'),
     ('utf-8', b'Overlong... \xfc\x83\xbf\xbf\xbf\xbf ... sequence'),
@@ -334,35 +292,6 @@ def test_body_part_properties():
             assert part.secure_filename == part.filename
 
 
-def test_filename_star():
-    # NOTE(vytas): Generated by requests_toolbelt 0.9.1 on Py2
-    #   (interestingly, one gets a "normal" filename on Py3.7).
-
-    data = (
-        b'--a0d738bcdb30449eb0d13f4b72c2897e\r\n'
-        b'Content-Disposition: form-data; name="file"; '
-        b"filename*=utf-8''%E2%AC%85%20Arrow.txt\r\n\r\n"
-        b'A unicode arrow in the filename.\r\n'
-        b'--a0d738bcdb30449eb0d13f4b72c2897e--\r\n'
-    )
-
-    handler = media.MultipartFormHandler()
-    content_type = ('multipart/form-data; boundary=' +
-                    'a0d738bcdb30449eb0d13f4b72c2897e')
-    stream = BufferedReader(io.BytesIO(data).read, len(data))
-    form = handler.deserialize(stream, content_type, len(data))
-    for part in form:
-        assert part.filename == '⬅ Arrow.txt'
-        assert part.secure_filename == '__Arrow.txt'
-
-    data = data.replace(b'*=utf-8', b'*=esoteric')
-    stream = BufferedReader(io.BytesIO(data).read, len(data))
-    form = handler.deserialize(stream, content_type, len(data))
-    for part in form:
-        with pytest.raises(falcon.HTTPBadRequest):
-            part.filename
-
-
 def test_empty_filename():
     data = (
         b'--a0d738bcdb30449eb0d13f4b72c2897e\r\n'
@@ -383,31 +312,15 @@ def test_empty_filename():
             part.secure_filename
 
 
-@pytest.mark.parametrize('max_headers_size', [64, 140, 141, 142, 256, 1024])
-def test_headers_edge_cases(max_headers_size):
-    data = (
-        b'--a0d738bcdb30449eb0d13f4b72c2897e\r\n'
-        b'X-Falcon: Peregrine\r\n'
-        b'Content-Type: application/vnd.oasis.opendocument.text\r\n'
-        b'Junk\r\n'
-        b'Content-Disposition: form-data; name="file"; filename=hd.txt\r\n\r\n'
-        b'No, it is not an ODT document...\r\n'
-        b'--a0d738bcdb30449eb0d13f4b72c2897e--\r\n'
-    )
+@falcon.runs_sync
+async def test_async_unsupported():
+    if falcon.ASGI_SUPPORTED:
+        pytest.skip('Testing missing ASGI support')
 
     handler = media.MultipartFormHandler()
-    handler.parse_options.max_body_part_headers_size = max_headers_size
-
-    content_type = ('multipart/form-data; boundary=' +
-                    'a0d738bcdb30449eb0d13f4b72c2897e')
-    stream = BufferedReader(io.BytesIO(data).read, len(data))
-    form = handler.deserialize(stream, content_type, len(data))
-
-    if max_headers_size < 142:
-        with pytest.raises(falcon.HTTPBadRequest):
-            list(form)
-    else:
-        assert len(list(form)) == 1
+    with pytest.raises(NotImplementedError):
+        await handler.deserialize_async(
+            'Dummy', 'multipart/form-data; boundary=BOUNDARY', None)
 
 
 class MultipartAnalyzer:
@@ -419,43 +332,110 @@ class MultipartAnalyzer:
                 'data': part.data.decode(),
                 'filename': part.filename,
                 'name': part.name,
+                'secure_filename': part.secure_filename if part.filename else None,
                 'text': part.text,
             })
 
         resp.media = values
 
+    def on_post_media(self, req, resp):
+        deserialized = []
+        for part in req.media:
+            part_media = part.get_media()
+            assert part_media == part.media
+            deserialized.append(part_media)
 
-@pytest.fixture
-def client():
-    app = falcon.App()
-    app.add_route('/media', MultipartAnalyzer())
+        resp.media = deserialized
+
+    def on_post_mirror(self, req, resp):
+        parts = []
+        for part in req.get_media():
+            parts.append({
+                'content': part.stream.read(),
+                'content_type': part.content_type,
+                'name': part.name,
+            })
+
+        resp.content_type = falcon.MEDIA_MSGPACK
+        resp.media = parts
+
+
+class AsyncMultipartAnalyzer:
+    async def on_post(self, req, resp):
+        values = []
+        form = await req.get_media()
+        async for part in form:
+            # NOTE(vytas): use the async property form to exercise that too.
+            values.append({
+                'content_type': part.content_type,
+                'data': (await part.data).decode(),
+                'filename': part.filename,
+                'name': part.name,
+                'secure_filename': part.secure_filename if part.filename else None,
+                'text': (await part.text),
+            })
+
+        resp.media = values
 
-    return testing.TestClient(app)
+    async def on_post_media(self, req, resp):
+        deserialized = []
+        form = await req.media
+        async for part in form:
+            part_media = await part.get_media()
+            assert part_media == await part.media
+            deserialized.append(part_media)
+
+        resp.media = deserialized
+
+    async def on_post_mirror(self, req, resp):
+        parts = []
+        async for part in await req.get_media():
+            parts.append({
+                'content': await part.stream.read(),
+                'content_type': part.content_type,
+                'name': part.name,
+            })
+
+        resp.content_type = falcon.MEDIA_MSGPACK
+        resp.media = parts
 
 
 @pytest.fixture
-def custom_client():
+def custom_client(asgi):
     def _factory(options):
         multipart_handler = media.MultipartFormHandler()
         for key, value in options.items():
             setattr(multipart_handler.parse_options, key, value)
-        handlers = media.Handlers({
+        req_handlers = media.Handlers({
             falcon.MEDIA_JSON: media.JSONHandler(),
             falcon.MEDIA_MULTIPART: multipart_handler,
         })
 
-        app = falcon.App()
-        app.req_options.media_handlers = handlers
-        app.add_route('/media', MultipartAnalyzer())
+        app = create_app(asgi)
+        app.req_options.media_handlers = req_handlers
+        app.resp_options.media_handlers = media.Handlers({
+            falcon.MEDIA_JSON: media.JSONHandler(),
+            falcon.MEDIA_MSGPACK: media.MessagePackHandler(),
+        })
+
+        resource = AsyncMultipartAnalyzer() if asgi else MultipartAnalyzer()
+        app.add_route('/submit', resource)
+        app.add_route('/media', resource, suffix='media')
+        app.add_route('/mirror', resource, suffix='mirror')
 
         return testing.TestClient(app)
 
     return _factory
 
 
+@pytest.fixture
+def client(custom_client):
+    return custom_client({})
+
+
 def test_upload_multipart(client):
     resp = client.simulate_post(
-        '/media',
+        '/submit',
         headers={
             'Content-Type':
             'multipart/form-data; boundary=5b11af82ab65407ba8cdccf37d2a9c4f',
@@ -469,6 +449,7 @@ def test_upload_multipart(client):
             'data': 'world',
             'filename': None,
             'name': 'hello',
+            'secure_filename': None,
             'text': 'world',
         },
         {
@@ -476,6 +457,7 @@ def test_upload_multipart(client):
             'data': '{"debug": true, "message": "Hello, world!", "score": 7}',
             'filename': None,
             'name': 'document',
+            'secure_filename': None,
             'text': None,
         },
         {
@@ -483,6 +465,7 @@ def test_upload_multipart(client):
             'data': 'Hello, world!\n',
             'filename': 'test.txt',
             'name': 'file1',
+            'secure_filename': 'test.txt',
             'text': 'Hello, world!\n',
         },
     ]
@@ -491,7 +474,7 @@ def test_upload_multipart(client):
 @pytest.mark.parametrize('truncated_by', [1, 2, 3, 4])
 def test_truncated_form(client, truncated_by):
     resp = client.simulate_post(
-        '/media',
+        '/submit',
         headers={
             'Content-Type':
             'multipart/form-data; boundary=5b11af82ab65407ba8cdccf37d2a9c4f',
@@ -507,7 +490,7 @@ def test_truncated_form(client, truncated_by):
 
 def test_unexected_form_structure(client):
     resp1 = client.simulate_post(
-        '/media',
+        '/submit',
         headers={
             'Content-Type':
             'multipart/form-data; boundary=5b11af82ab65407ba8cdccf37d2a9c4f',
@@ -521,7 +504,7 @@ def test_unexected_form_structure(client):
     }
 
     resp2 = client.simulate_post(
-        '/media',
+        '/submit',
         headers={
             'Content-Type':
             'multipart/form-data; boundary=5b11af82ab65407ba8cdccf37d2a9c4f',
@@ -537,7 +520,7 @@ def test_unexected_form_structure(client):
 
 def test_data_too_large(client):
     resp = client.simulate_post(
-        '/media',
+        '/submit',
         headers={
             'Content-Type':
             'multipart/form-data; boundary=BOUNDARY',
@@ -556,7 +539,7 @@ def test_too_many_body_parts(custom_client, max_body_part_count):
     client = custom_client({'max_body_part_count': max_body_part_count})
     boundary = '---------------------------1574247108204320607285918568'
     resp = client.simulate_post(
-        '/media',
+        '/submit',
         headers={
             'Content-Type':
             'multipart/form-data; boundary=' + boundary,
@@ -590,15 +573,20 @@ def test_random_form(client):
     form = handler.deserialize(
         io.BytesIO(form_data), content_type, len(form_data))
 
+    resp = client.simulate_post(
+        '/mirror', headers={'Content-Type': content_type}, body=form_data)
+    assert resp.status_code == 200
+    form = msgpack.unpackb(resp.content, raw=False)
+
     for index, part in enumerate(form):
-        assert part.content_type == 'application/x-falcon-urandom'
-        assert part.data == part_data[index]
+        assert part['content'] == part_data[index]
+        assert part['content_type'] == 'application/x-falcon-urandom'
 
 
 def test_invalid_random_form(client):
     length = random.randint(2**20, 2**21)
     resp = client.simulate_post(
-        '/media',
+        '/submit',
         headers={
             'Content-Type':
             'multipart/form-data; boundary=' + HASH_BOUNDARY,
@@ -606,3 +594,189 @@ def test_invalid_random_form(client):
         body=os.urandom(length))
 
     assert resp.status_code == 400
+
+
+def test_content_transfer_encoding_header(client):
+    data = (
+        b'--BOUNDARY\r\n'
+        b'Content-Disposition: form-data; name="file"; filename="bytes"\r\n'
+        b'Content-Transfer-Encoding: Base64'
+        b'Content-Type: application/x-falcon\r\n\r\n'
+        b'UGVyZWdyaW5lIEZhbGNvbiADLgA='
+        b'\r\n'
+        b'--BOUNDARY\r\n'
+        b'Content-Disposition: form-data; name="empty"\r\n'
+        b'Content-Type: text/plain\r\n\r\n'
+        b'\r\n'
+        b'--BOUNDARY--\r\n'
+    )
+
+    resp = client.simulate_post(
+        '/submit',
+        headers={'Content-Type': 'multipart/form-data; boundary=BOUNDARY'},
+        body=data)
+
+    assert resp.status_code == 400
+    assert resp.json == {
+        'description': ('the deprecated Content-Transfer-Encoding header '
+                        'field is unsupported'),
+        'title': 'Malformed multipart/form-data request media',
+    }
+
+
+def test_unsupported_charset(client):
+    data = (
+        b'--BOUNDARY\r\n'
+        b'Content-Disposition: form-data; name="text"\r\n'
+        b'Content-Type: text/plain; charset=pecyn\r\n\r\n'
+        b'AAHEHlRoZSBGYWxjb24gV2ViIEZyYW1ld29yaywgMjAxOQ=='
+        b'\r\n'
+        b'--BOUNDARY\r\n'
+        b'Content-Disposition: form-data; name="empty"\r\n'
+        b'Content-Type: text/plain\r\n\r\n'
+        b'\r\n'
+        b'--BOUNDARY--\r\n'
+    )
+
+    resp = client.simulate_post(
+        '/submit',
+        headers={'Content-Type': 'multipart/form-data; boundary=BOUNDARY'},
+        body=data)
+
+    assert resp.status_code == 400
+    assert resp.json == {
+        'description': 'invalid text or charset: pecyn',
+        'title': 'Malformed multipart/form-data request media',
+    }
+
+
+def test_filename_star(client):
+    # NOTE(vytas): Generated by requests_toolbelt 0.9.1 on Py2
+    #   (interestingly, one gets a "normal" filename on Py3.7).
+
+    data = (
+        b'--a0d738bcdb30449eb0d13f4b72c2897e\r\n'
+        b'Content-Disposition: form-data; name="file"; '
+        b"filename*=utf-8''%E2%AC%85%20Arrow.txt\r\n\r\n"
+        b'A unicode arrow in the filename.\r\n'
+        b'--a0d738bcdb30449eb0d13f4b72c2897e--\r\n'
+    )
+    content_type = ('multipart/form-data; boundary=' +
+                    'a0d738bcdb30449eb0d13f4b72c2897e')
+
+    resp = client.simulate_post(
+        '/submit', headers={'Content-Type': content_type}, body=data)
+    assert resp.status_code == 200
+    assert resp.json == [{
+        'content_type': 'text/plain',
+        'data': 'A unicode arrow in the filename.',
+        'filename': '⬅ Arrow.txt',
+        'name': 'file',
+        'secure_filename': '__Arrow.txt',
+        'text': 'A unicode arrow in the filename.',
+    }]
+
+    data = data.replace(b'*=utf-8', b'*=esoteric')
+    resp = client.simulate_post(
+        '/submit', headers={'Content-Type': content_type}, body=data)
+    assert resp.status_code == 400
+    assert resp.json == {
+        'description': 'invalid text or charset: esoteric',
+        'title': 'Malformed multipart/form-data request media',
+    }
+
+
+@pytest.mark.parametrize('max_headers_size', [64, 140, 141, 142, 256, 1024])
+def test_headers_edge_cases(custom_client, max_headers_size):
+    client = custom_client({'max_body_part_headers_size': max_headers_size})
+
+    data = (
+        b'--a0d738bcdb30449eb0d13f4b72c2897e\r\n'
+        b'X-Falcon: Peregrine\r\n'
+        b'Content-Type: application/vnd.oasis.opendocument.text\r\n'
+        b'Junk\r\n'
+        b'Content-Disposition: form-data; name="file"; filename=hd.txt\r\n\r\n'
+        b'No, it is not an ODT document...\r\n'
+        b'--a0d738bcdb30449eb0d13f4b72c2897e--\r\n'
+    )
+    content_type = ('multipart/form-data; boundary=' +
+                    'a0d738bcdb30449eb0d13f4b72c2897e')
+
+    resp = client.simulate_post(
+        '/submit', headers={'Content-Type': content_type}, body=data)
+
+    if max_headers_size < 142:
+        assert resp.status_code == 400
+        assert resp.json == {
+            'description': 'incomplete body part headers',
+            'title': 'Malformed multipart/form-data request media',
+        }
+    else:
+        assert len(resp.json) == 1
+
+
+def test_deserialize_part_media(client):
+    data = (
+        b'--BOUNDARY\r\n'
+        b'Content-Disposition: form-data; name="factorials"\r\n'
+        b'Content-Type: application/json\r\n\r\n'
+        b'{"count": 6, "numbers": [1, 2, 6, 24, 120, 720]}\r\n'
+        b'--BOUNDARY\r\n'
+        b'Content-Disposition: form-data; name="person"\r\n'
+        b'Content-Type: application/x-www-form-urlencoded\r\n\r\n'
+        b'name=Jane&surname=Doe&fruit=%F0%9F%8D%8F\r\n'
+        b'--BOUNDARY--\r\n'
+    )
+
+    resp = client.simulate_post(
+        '/media',
+        headers={'Content-Type': 'multipart/form-data; boundary=BOUNDARY'},
+        body=data)
+
+    assert resp.status_code == 200
+    assert resp.json == [
+        {'count': 6, 'numbers': [1, 2, 6, 24, 120, 720]},
+        {'fruit': '🍏', 'name': 'Jane', 'surname': 'Doe'},
+    ]
+
+
+def test_deserialize_custom_media(custom_client):
+    class FirstByteHandler(media.BaseHandler):
+        exhaust_stream = True
+
+        def deserialize(self, stream, content_type, content_length):
+            first_byte = stream.read(1)
+            if first_byte:
+                return '0x{:02x}'.format(first_byte[0])
+            return ''
+
+        async def deserialize_async(self, stream, content_type,
+                                    content_length):
+            first_byte = await stream.read(1)
+            if first_byte:
+                return '0x{:02x}'.format(first_byte[0])
+            return ''
+
+    handlers = media.Handlers(
+        {'application/x-falcon-first-byte': FirstByteHandler()})
+    client = custom_client({'media_handlers': handlers})
+
+    data = (
+        b'--BOUNDARY\r\n'
+        b'Content-Disposition: form-data; name="first"\r\n'
+        b'Content-Type: application/x-falcon-first-byte\r\n\r\n'
+        b'\r\n'
+        b'--BOUNDARY\r\n'
+        b'Content-Disposition: form-data; name="second"\r\n'
+        b'Content-Type: application/x-falcon-first-byte\r\n\r\n'
+        b'Hi!\r\n'
+        b'--BOUNDARY--\r\n'
+    )
+
+    resp = client.simulate_post(
+        '/media',
+        headers={'Content-Type': 'multipart/form-data; boundary=BOUNDARY'},
+        body=data)
+
+    assert resp.status_code == 200
+    assert resp.json == ['', '0x48']
diff --git a/tests/test_media_urlencoded.py b/tests/test_media_urlencoded.py
index c312346..c5c2125 100644
--- a/tests/test_media_urlencoded.py
+++ b/tests/test_media_urlencoded.py
@@ -30,7 +30,7 @@ def test_urlencoded_form_handler_serialize(data, expected):
     handler = media.URLEncodedFormHandler()
     assert handler.serialize(data, falcon.MEDIA_URLENCODED) == expected
 
-    value = testing.invoke_coroutine_sync(handler.serialize_async, data, falcon.MEDIA_URLENCODED)
+    value = falcon.invoke_coroutine_sync(handler.serialize_async, data, falcon.MEDIA_URLENCODED)
     assert value == expected
 
 
diff --git a/tests/test_static.py b/tests/test_static.py
index c51e9c8..a534842 100644
--- a/tests/test_static.py
+++ b/tests/test_static.py
@@ -102,7 +102,7 @@ def test_bad_path(asgi, uri, monkeypatch):
 
     with pytest.raises(falcon.HTTPNotFound):
         if asgi:
-            testing.invoke_coroutine_sync(sr, req, resp)
+            falcon.invoke_coroutine_sync(sr, req, resp)
         else:
             sr(req, resp)
 
@@ -181,7 +181,7 @@ def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch
             await sr(req, resp)
             return await resp.stream.read()
 
-        body = testing.invoke_coroutine_sync(run)
+        body = falcon.invoke_coroutine_sync(run)
     else:
         sr(req, resp)
         body = resp.stream.read()
@@ -291,7 +291,7 @@ def test_fallback_filename(asgi, uri, default, expected, content_type, downloada
             await sr(req, resp)
             return await resp.stream.read()
 
-        body = testing.invoke_coroutine_sync(run)
+        body = falcon.invoke_coroutine_sync(run)
     else:
         sr(req, resp)
         body = resp.stream.read()
diff --git a/tests/test_testing.py b/tests/test_testing.py
index 26a7514..17cdcd2 100644
--- a/tests/test_testing.py
+++ b/tests/test_testing.py
@@ -82,7 +82,7 @@ def test_asgi_request_event_emitter_hang():
 
         assert (elapsed + 0.1) > expected_elasped_min
 
-    testing.invoke_coroutine_sync(t)
+    falcon.invoke_coroutine_sync(t)
 
 
 def test_ignore_extra_asgi_events():
@@ -97,7 +97,7 @@ def test_ignore_extra_asgi_events():
         await collect({'type': 'http.response.body'})
         assert len(collect.events) == 2
 
-    testing.invoke_coroutine_sync(t)
+    falcon.invoke_coroutine_sync(t)
 
 
 def test_invalid_asgi_events():
@@ -149,7 +149,7 @@ def test_invalid_asgi_events():
             # NOTE(kgriffs): Invalid type
             await collect({'type': 'http.response.bod'})
 
-    testing.invoke_coroutine_sync(t)
+    falcon.invoke_coroutine_sync(t)
 
 
 def test_is_asgi_app_cls():
diff --git a/tests/test_utils.py b/tests/test_utils.py
index 6cd5866..af4b5cb 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -12,7 +12,7 @@ import pytest
 import falcon
 from falcon import testing
 from falcon import util
-from falcon.util import json, misc, structures, uri
+from falcon.util import deprecation, json, misc, structures, uri
 
 from _util import create_app, to_coroutine  # NOQA
 
@@ -1052,7 +1052,7 @@ class TestContextType:
 class TestDeprecatedArgs:
     def test_method(self, recwarn):
         class C:
-            @misc.deprecated_args(allowed_positional=0)
+            @deprecation.deprecated_args(allowed_positional=0)
             def a_method(self, a=1, b=2):
                 pass
 
@@ -1062,7 +1062,7 @@ class TestDeprecatedArgs:
         assert len(recwarn) == 1
 
     def test_function(self, recwarn):
-        @misc.deprecated_args(allowed_positional=0, is_method=False)
+        @deprecation.deprecated_args(allowed_positional=0, is_method=False)
         def a_function(a=1, b=2):
             pass
 
diff --git a/tests/test_validators.py b/tests/test_validators.py
index 7bf6e0b..c330deb 100644
--- a/tests/test_validators.py
+++ b/tests/test_validators.py
@@ -121,7 +121,7 @@ def call_method(asgi, method_name, *args):
     resource = ResourceAsync() if asgi else Resource()
 
     if asgi:
-        return testing.invoke_coroutine_sync(getattr(resource, method_name), *args)
+        return falcon.invoke_coroutine_sync(getattr(resource, method_name), *args)
 
     return getattr(resource, method_name)(*args)