# 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.

"""Testing utilities.

This module contains various testing utilities that can be accessed
directly from the `testing` package::

    from falcon import testing

    wsgi_environ = testing.create_environ()
"""

from __future__ import annotations

import asyncio
from collections import defaultdict
from collections import deque
import contextlib
from enum import auto
from enum import Enum
import io
import itertools
import json
import random
import re
import socket
import sys
import time
from typing import (
    Any,
    Callable,
    Deque,
    Dict,
    Iterable,
    Iterator,
    List,
    Mapping,
    Optional,
    TextIO,
    Tuple,
    Type,
    Union,
)

import falcon
from falcon import errors as falcon_errors
from falcon._typing import CookieArg
from falcon._typing import HeaderArg
from falcon._typing import ResponseStatus
import falcon.asgi
from falcon.asgi_spec import AsgiEvent
from falcon.asgi_spec import EventType
from falcon.asgi_spec import ScopeType
from falcon.asgi_spec import WSCloseCode
from falcon.constants import SINGLETON_HEADERS
import falcon.request
from falcon.util import code_to_http_status
from falcon.util import uri
from falcon.util.mediatypes import parse_header

# NOTE(kgriffs): Changed in 3.0 from 'curl/7.24.0 (x86_64-apple-darwin12.0)'
DEFAULT_UA = 'falcon-client/' + falcon.__version__
DEFAULT_HOST = 'falconframework.org'


class ASGILifespanEventEmitter:
    """Emits ASGI lifespan events to an ASGI app.

    This class can be used to drive a standard ASGI app callable in order to
    perform functional tests on the app in question.

    When simulating both lifespan and per-request events, each event stream
    will require a separate invocation of the ASGI callable; one with a
    lifespan event emitter, and one with a request event emitter. An
    asyncio :class:`~asyncio.Condition` can be used to pause the
    lifespan emitter until all of the desired request events have been
    emitted.

    Keyword Args:
        shutting_down (asyncio.Condition): An instance of
            :class:`asyncio.Condition` that will be awaited before
            emitting the final shutdown event (``'lifespan.shutdown``).
    """

    def __init__(self, shutting_down: asyncio.Condition) -> None:
        self._state = 0
        self._shutting_down = shutting_down

    async def emit(self) -> AsgiEvent:
        if self._state == 0:
            self._state += 1
            return {'type': EventType.LIFESPAN_STARTUP}

        if self._state == 1:
            self._state += 1
            # NOTE(kgriffs): This verifies the app ignores events it does
            #   not recognize.
            return {'type': 'lifespan._nonstandard_event'}

        async with self._shutting_down:
            await self._shutting_down.wait()

        return {'type': EventType.LIFESPAN_SHUTDOWN}

    __call__ = emit


class ASGIRequestEventEmitter:
    """Emits events on-demand to an ASGI app.

    This class can be used to drive a standard ASGI app callable in order to
    perform functional tests on the app in question.

    Note:
        In order to ensure the app is able to handle subtle variations
        in the ASGI events that are allowed by the specification, such
        variations are applied to the emitted events at unspecified
        intervals. This includes whether or not the `more_body` field
        is explicitly set, or whether or not the request `body` chunk in
        the event is occasionally empty,

    Keyword Args:
        body (str): The body content to use when emitting http.request
            events. May be an empty string. If a byte string, it will
            be used as-is; otherwise it will be encoded as UTF-8
            (default ``b''``).
        chunk_size (int): The maximum number of bytes to include in
            a single http.request event (default 4096).
        disconnect_at (float): The Unix timestamp after which to begin
            emitting ``'http.disconnect'`` events (default now + 30s). The
            value may be either an ``int`` or a ``float``, depending
            on the precision required. Setting `disconnect_at` to
            ``0`` is treated as a special case, and will result in an
            ``'http.disconnect'`` event being immediately emitted (rather than
            first emitting an ``'http.request'`` event).
    """

    # TODO(kgriffs): If this pattern later becomes useful elsewhere,
    #   factor out into a standalone helper class.
    _branch_decider: dict[str, bool] = defaultdict(bool)

    def __init__(
        self,
        body: Optional[Union[str, bytes]] = None,
        chunk_size: Optional[int] = None,
        disconnect_at: Optional[Union[int, float]] = None,
    ) -> None:
        if body is None:
            body = b''
        elif not isinstance(body, bytes):
            body = body.encode()

        body = memoryview(body)

        if disconnect_at is None:
            disconnect_at = time.time() + 30

        if chunk_size is None:
            chunk_size = 4096

        self._body: Optional[memoryview] = body
        self._chunk_size = chunk_size
        self._emit_empty_chunks = True
        self._disconnect_at = disconnect_at
        self._disconnected = False
        self._exhaust_body = True

        self._emitted_empty_chunk_a = False
        self._emitted_empty_chunk_b = False

    @property
    def disconnected(self) -> bool:
        """Returns ``True`` if the simulated client connection is in a
        "disconnected" state.
        """  # noqa: D205
        return self._disconnected or (self._disconnect_at <= time.time())

    def disconnect(self, exhaust_body: Optional[bool] = None) -> None:
        """Set the client connection state to disconnected.

        Call this method to simulate an immediate client disconnect and
        begin emitting ``'http.disconnect'`` events.

        Arguments:
            exhaust_body (bool): Set to ``False`` in order to
                begin emitting ``'http.disconnect'`` events without first
                emitting at least one ``'http.request'`` event.
        """

        if exhaust_body is not None:
            self._exhaust_body = exhaust_body

        self._disconnected = True

    async def emit(self) -> AsgiEvent:
        # NOTE(kgriffs): Special case: if we are immediately disconnected,
        #   the first event should be 'http.disconnnect'
        if self._disconnect_at == 0:
            return {'type': EventType.HTTP_DISCONNECT}

        #
        # NOTE(kgriffs): Based on my reading of the ASGI spec, at least one
        #   'http.request' event should be emitted before 'http.disconnect'
        #   for normal requests. However, the server may choose to
        #   immediately abandon a connection for some reason, in which case
        #   an 'http.request' event may never be sent.
        #
        #   See also: https://asgi.readthedocs.io/en/latest/specs/main.html#events
        #
        if self._body is None or not self._exhaust_body:
            # NOTE(kgriffs): When there are no more events, an ASGI
            #   server will hang until the client connection
            #   disconnects.
            while not self.disconnected:
                await asyncio.sleep(0.001)

            return {'type': EventType.HTTP_DISCONNECT}

        event: Dict[str, Any] = {'type': EventType.HTTP_REQUEST}

        if self._emit_empty_chunks:
            # NOTE(kgriffs): Return a couple variations on empty chunks
            #   every time, to ensure test coverage.
            if not self._emitted_empty_chunk_a:
                self._emitted_empty_chunk_a = True
                event['more_body'] = True
                return event

            if not self._emitted_empty_chunk_b:
                self._emitted_empty_chunk_b = True
                event['more_body'] = True
                event['body'] = b''
                return event

            # NOTE(kgriffs): Part of the time just return an
            #   empty chunk to make sure the app handles that
            #   correctly.
            if self._toggle_branch('return_empty_chunk'):
                event['more_body'] = True

                # NOTE(kgriffs): Since ASGI specifies that
                #   'body' is optional, we toggle whether
                #   or not to explicitly set it to b'' to ensure
                #   the app handles both correctly.
                if self._toggle_branch('explicit_empty_body_1'):
                    event['body'] = b''

                return event

        chunk = self._body[: self._chunk_size]
        self._body = self._body[self._chunk_size :] or None

        if chunk:
            event['body'] = bytes(chunk)
        elif self._toggle_branch('explicit_empty_body_2'):
            # NOTE(kgriffs): Since ASGI specifies that
            #   'body' is optional, we toggle whether
            #   or not to explicitly set it to b'' to ensure
            #   the app handles both correctly.
            event['body'] = b''

        if self._body:
            event['more_body'] = True
        elif self._toggle_branch('set_more_body_false'):
            # NOTE(kgriffs): The ASGI spec allows leaving off
            #   the 'more_body' key when it would be set to
            #   False, so toggle one of the approaches
            #   to make sure the app handles both cases.
            event['more_body'] = False

        return event

    __call__ = emit

    def _toggle_branch(self, name: str) -> bool:
        self._branch_decider[name] = not self._branch_decider[name]
        return self._branch_decider[name]


class ASGIResponseEventCollector:
    """Collects and validates ASGI events returned by an app.

    Raises:
        TypeError: An event field emitted by the app was of an unexpected type.
        ValueError: Invalid event name or field value.
    """

    _LIFESPAN_EVENT_TYPES = frozenset(
        [
            'lifespan.startup.complete',
            'lifespan.startup.failed',
            'lifespan.shutdown.complete',
            'lifespan.shutdown.failed',
        ]
    )

    _HEADER_NAME_RE = re.compile(rb'^[a-zA-Z][a-zA-Z0-9\-_]*$')
    _BAD_HEADER_VALUE_RE = re.compile(rb'[\000-\037]')

    events: List[AsgiEvent]
    """An iterable of events that were emitted by the app,
    collected as-is from the app.
    """
    headers: List[Tuple[str, str]]
    """An iterable of (str, str) tuples representing the ISO-8859-1 decoded
    headers emitted by the app in the body of the ``'http.response.start'`` event.
    """
    status: Optional[ResponseStatus]
    """HTTP status code emitted by the app in the body of the
    ``'http.response.start'`` event.
    """
    body_chunks: List[bytes]
    """An iterable of ``bytes`` objects emitted by the app via
    ``'http.response.body'`` events.
    """
    more_body: Optional[bool]
    """Whether or not the app expects to emit more body chunks.

    Will be ``None`` if unknown (i.e., the app has not yet emitted
    any ``'http.response.body'`` events.)
    """

    def __init__(self) -> None:
        self.events = []
        self.headers = []
        self.status = None
        self.body_chunks = []
        self.more_body = None

    async def collect(self, event: AsgiEvent) -> None:
        if self.more_body is False:
            # NOTE(kgriffs): According to the ASGI spec, once we get a
            #   message setting more_body to False, any further messages
            #   on the channel are ignored.
            return

        self.events.append(event)

        event_type = event['type']
        if not isinstance(event_type, str):
            raise TypeError('ASGI event type must be a Unicode string')

        if event_type == EventType.HTTP_RESPONSE_START:
            for name, value in event.get('headers', []):
                if not isinstance(name, bytes):
                    raise TypeError('ASGI header names must be byte strings')
                if not isinstance(value, bytes):
                    raise TypeError('ASGI header names must be byte strings')

                # NOTE(vytas): Ported basic validation from wsgiref.validate.
                if not self._HEADER_NAME_RE.match(name):
                    raise ValueError('Bad header name: {!r}'.format(name))
                if self._BAD_HEADER_VALUE_RE.search(value):
                    raise ValueError('Bad header value: {!r}'.format(value))

                # NOTE(vytas): After the name validation above, the name is
                #   guaranteed to only contain a subset of ASCII.
                name_decoded = name.decode()
                if not name_decoded.islower():
                    raise ValueError('ASGI header names must be lowercase')

                self.headers.append((name_decoded, value.decode('latin1')))

            self.status = event['status']

            if not isinstance(self.status, int):
                raise TypeError('ASGI status must be an int')

        elif event_type == EventType.HTTP_RESPONSE_BODY:
            chunk = event.get('body', b'')
            if not isinstance(chunk, bytes):
                raise TypeError('ASGI body content must be a byte string')

            self.body_chunks.append(chunk)

            self.more_body = event.get('more_body', False)
            if not isinstance(self.more_body, bool):
                raise TypeError('ASGI more_body flag must be a bool')

        elif event_type not in self._LIFESPAN_EVENT_TYPES:
            raise ValueError('Invalid ASGI event type: ' + event_type)

    __call__ = collect


class _WebSocketState(Enum):
    CONNECT = auto()
    HANDSHAKE = auto()
    ACCEPTED = auto()
    DENIED = auto()
    CLOSED = auto()


class ASGIWebSocketSimulator:
    """Simulates a WebSocket client for testing a Falcon ASGI app.

    This class provides a way to test WebSocket endpoints in a Falcon ASGI app
    without having to interact with an actual ASGI server. While it is certainly
    important to test against a real server, a number of functional tests can be
    satisfied more efficiently and transparently with a simulated connection.

    Note:

        The ASGIWebSocketSimulator class is not designed to be instantiated
        directly; rather it should be obtained via
        :meth:`~falcon.testing.ASGIConductor.simulate_ws`.
    """

    _DEFAULT_WAIT_READY_TIMEOUT = 5

    def __init__(self) -> None:
        self.__msgpack = None

        self._state = _WebSocketState.CONNECT
        self._disconnect_emitted = False
        self._close_code: Optional[int] = None
        self._close_reason: Optional[str] = None
        self._accepted_subprotocol: Optional[str] = None
        self._accepted_headers: Optional[List[Tuple[bytes, bytes]]] = None
        self._collected_server_events: Deque[AsgiEvent] = deque()
        self._collected_client_events: Deque[AsgiEvent] = deque()

        self._event_handshake_complete = asyncio.Event()

    @property
    def ready(self) -> bool:
        """``True`` if the WebSocket connection has been accepted and the client is
        still connected, ``False`` otherwise.
        """  # noqa: D205
        return self._state == _WebSocketState.ACCEPTED

    @property
    def closed(self) -> bool:
        """``True`` if the WebSocket connection has been denied or closed by the app,
        or the client has disconnected.
        """  # noqa: D205
        return self._state in {_WebSocketState.DENIED, _WebSocketState.CLOSED}

    @property
    def close_code(self) -> Optional[int]:
        """The WebSocket close code provided by the app if the connection is closed.

        Returns ``None`` if the connection is still open.
        """
        return self._close_code

    @property
    def close_reason(self) -> Optional[str]:
        """The WebSocket close reason provided by the app if the connection is closed.

        Returns ``None`` if the connection is still open.
        """
        return self._close_reason

    @property
    def subprotocol(self) -> Optional[str]:
        """The subprotocol the app wishes to accept, or ``None`` if not specified."""
        return self._accepted_subprotocol

    @property
    def headers(self) -> Optional[List[Tuple[bytes, bytes]]]:
        """An iterable of ``[name, value]`` two-item tuples, where *name* is the
        header name, and *value* is the header value for each header returned by
        the app when it accepted the WebSocket connection.
        This property resolves to ``None`` if the connection has not been accepted.
        """  # noqa: D205
        return self._accepted_headers

    async def wait_ready(self, timeout: Optional[int] = None) -> None:
        """Wait until the connection has been accepted or denied.

        This coroutine can be awaited in order to pause execution until the
        app has accepted or denied the connection. In the latter case, an
        error will be raised to the caller.

        Keyword Args:
            timeout (int): Number of seconds to wait before giving up and
                raising an error (default: ``5``).
        """

        timeout = timeout or self._DEFAULT_WAIT_READY_TIMEOUT

        try:
            await asyncio.wait_for(self._event_handshake_complete.wait(), timeout)
        except asyncio.TimeoutError:
            msg = (
                f'Timed out after waiting {timeout} seconds for the WebSocket '
                f'handshake to complete. Check the on_websocket responder and '
                f'any middleware for any conditions that may be stalling the '
                f'request flow.'
            )
            raise asyncio.TimeoutError(msg)

        self._require_accepted()

    # NOTE(kgriffs): This is a coroutine just in case we need it to be
    #   in a future code revision. It also makes it more consistent
    #   with the other methods.
    async def close(
        self, code: Optional[int] = None, reason: Optional[str] = None
    ) -> None:
        """Close the simulated connection.

        Keyword Args:
            code (int): The WebSocket close code to send to the application
                per the WebSocket spec (default: ``1000``).
            reason (str): The WebSocket close reason to send to the application
                per the WebSocket spec (default: empty string).
        """

        # NOTE(kgriffs): Give our collector a chance in case the
        #   server is trying to close at the same time (e.g., there was an
        #   unhandled error and the server wants to disconnect with an error
        #   code.) We await a few times to let the server app settle across
        #   multiple of its own await's.
        for __ in range(3):
            await asyncio.sleep(0)

        if self.closed:
            return

        assert self._close_code is None

        if code is None:
            code = WSCloseCode.NORMAL

        if reason is None:
            reason = ''

        self._state = _WebSocketState.CLOSED
        self._close_code = code
        self._close_reason = reason

    async def send_text(self, payload: str) -> None:
        """Send a message to the app with a Unicode string payload.

        Arguments:
            payload (str): The string to send.
        """

        # NOTE(kgriffs): We have to check ourselves because some ASGI
        #   servers are not very strict which can lead to hard-to-debug
        #   errors.
        if not isinstance(payload, str):
            raise TypeError('payload must be a string')

        # NOTE(kgriffs): From the client's perspective, it was a send,
        #   but the server will be expecting websocket.receive
        await self._send(text=payload)

    async def send_data(self, payload: Union[bytes, bytearray, memoryview]) -> None:
        """Send a message to the app with a binary data payload.

        Arguments:
            payload (Union[bytes, bytearray, memoryview]): The binary data to send.
        """

        # NOTE(kgriffs): We have to check ourselves because some ASGI
        #   servers are not very strict which can lead to hard-to-debug
        #   errors.
        if not isinstance(payload, (bytes, bytearray, memoryview)):
            raise TypeError('payload must be a byte string')

        # NOTE(kgriffs): From the client's perspective, it was a send,
        #   but the server will be expecting websocket.receive
        await self._send(data=bytes(payload))

    async def send_json(self, media: object) -> None:
        """Send a message to the app with a JSON-encoded payload.

        Arguments:
            media: A JSON-encodable object to send as a TEXT (0x01) payload.
        """

        text = json.dumps(media)
        await self.send_text(text)

    async def send_msgpack(self, media: object) -> None:
        """Send a message to the app with a MessagePack-encoded payload.

        Arguments:
            media: A MessagePack-encodable object to send as a BINARY (0x02) payload.
        """

        data = self._msgpack.packb(media, use_bin_type=True)
        await self.send_data(data)

    async def receive_text(self) -> str:
        """Receive a message from the app with a Unicode string payload.

        Awaiting this coroutine will block until a message is available or
        the WebSocket is disconnected.
        """

        event = await self._receive()

        # PERF(kgriffs): When we normally expect the key to be
        #   present, this pattern is faster than get()
        try:
            text = event['text']
        except KeyError:
            text = None

        # NOTE(kgriffs): Even if the key is present, it may be None
        if text is None:
            raise falcon_errors.PayloadTypeError(
                'Expected TEXT payload but got BINARY instead'
            )

        return text

    async def receive_data(self) -> bytes:
        """Receive a message from the app with a binary data payload.

        Awaiting this coroutine will block until a message is available or
        the WebSocket is disconnected.
        """

        event = await self._receive()

        # PERF(kgriffs): When we normally expect the key to be
        #   present, EAFP is faster than get()
        try:
            data = event['bytes']
        except KeyError:
            data = None

        # NOTE(kgriffs): Even if the key is present, it may be None
        if data is None:
            raise falcon_errors.PayloadTypeError(
                'Expected BINARY payload but got TEXT instead'
            )

        return data

    async def receive_json(self) -> Any:
        """Receive a message from the app with a JSON-encoded TEXT payload.

        Awaiting this coroutine will block until a message is available or
        the WebSocket is disconnected.
        """

        text = await self.receive_text()
        return json.loads(text)

    async def receive_msgpack(self) -> Any:
        """Receive a message from the app with a MessagePack-encoded BINARY payload.

        Awaiting this coroutine will block until a message is available or
        the WebSocket is disconnected.
        """

        data = await self.receive_data()
        return self._msgpack.unpackb(data, use_list=True, raw=False)

    @property
    def _msgpack(self) -> Any:
        # NOTE(kgriffs): A property is used in lieu of referencing
        #   the msgpack module directly, in order to bubble up the
        #   import error in an obvious way, when the package has
        #   not been installed.

        if not self.__msgpack:
            import msgpack

            self.__msgpack = msgpack

        return self.__msgpack

    def _require_accepted(self) -> None:
        if self._state == _WebSocketState.ACCEPTED:
            return

        if self._state in {_WebSocketState.CONNECT, _WebSocketState.HANDSHAKE}:
            raise falcon_errors.OperationNotAllowed(
                'WebSocket connection has not yet been accepted'
            )
        elif self._state == _WebSocketState.CLOSED:
            raise falcon_errors.WebSocketDisconnected(self._close_code)

        assert self._state == _WebSocketState.DENIED

        if self._close_code == WSCloseCode.PATH_NOT_FOUND:
            raise falcon_errors.WebSocketPathNotFound(WSCloseCode.PATH_NOT_FOUND)

        if self._close_code == WSCloseCode.SERVER_ERROR:
            raise falcon_errors.WebSocketServerError(WSCloseCode.SERVER_ERROR)

        if self._close_code == WSCloseCode.HANDLER_NOT_FOUND:
            raise falcon_errors.WebSocketHandlerNotFound(WSCloseCode.HANDLER_NOT_FOUND)

        raise falcon_errors.WebSocketDisconnected(self._close_code)

    # NOTE(kgriffs): This is a coroutine just in case we need it to be
    #   in a future code revision. It also makes it more consistent
    #   with the other methods.
    async def _send(
        self, data: Optional[bytes] = None, text: Optional[str] = None
    ) -> None:
        self._require_accepted()

        # NOTE(kgriffs): From the client's perspective, it was a send,
        #   but the server will be expecting websocket.receive
        event: Dict[str, Any] = {'type': EventType.WS_RECEIVE}
        if data is not None:
            event['bytes'] = data

        if text is not None:
            event['text'] = text

        self._collected_client_events.append(event)

        # NOTE(kgriffs): If something is waiting to read this data on the
        #   other side, give it a chance to progress (because we like to party
        #   like it's 1992.)
        await asyncio.sleep(0)

    async def _receive(self) -> AsgiEvent:
        while not self._collected_server_events:
            self._require_accepted()
            await asyncio.sleep(0)

        self._require_accepted()
        return self._collected_server_events.popleft()

    async def _emit(self) -> AsgiEvent:
        if self._state == _WebSocketState.CONNECT:
            self._state = _WebSocketState.HANDSHAKE
            return {'type': EventType.WS_CONNECT}

        if self._state == _WebSocketState.HANDSHAKE:
            # NOTE(kgriffs): We need to wait for the handshake to
            #   complete, before proceeding.
            await self._event_handshake_complete.wait()

        while not self._collected_client_events:
            await asyncio.sleep(0)
            if self.closed:
                return self._create_checked_disconnect()

        return self._collected_client_events.popleft()

    async def _collect(self, event: AsgiEvent) -> None:
        assert event

        if self._state == _WebSocketState.CONNECT:
            raise falcon_errors.OperationNotAllowed(
                'An ASGI application must receive the first websocket.connect '
                'event before attempting to send any events.'
            )

        event_type = event['type']
        if self._state == _WebSocketState.HANDSHAKE:
            if event_type == EventType.WS_ACCEPT:
                self._state = _WebSocketState.ACCEPTED
                self._accepted_subprotocol = event.get('subprotocol')
                self._accepted_headers = event.get('headers')
                self._event_handshake_complete.set()

                # NOTE(kgriffs): Yield to other pending tasks that may be
                #   waiting on the completion of the handshake. This ensures
                #   that the simulated client connection can enter its context
                #   before the app logic continues and potentially closes the
                #   connection from that side.
                await asyncio.sleep(0)

            elif event_type == EventType.WS_CLOSE:
                self._state = _WebSocketState.DENIED

                desired_code = event.get('code', WSCloseCode.NORMAL)
                reason = event.get('reason', '')
                if desired_code == WSCloseCode.SERVER_ERROR or (
                    3000 <= desired_code < 4000
                ):
                    # NOTE(kgriffs): Pass this code through since it is a
                    #   special code we have set in the framework to trigger
                    #   different raised error types or to pass through a
                    #   raised HTTPError status code.
                    self._close_code = desired_code
                    self._close_reason = reason
                else:
                    # NOTE(kgriffs): Force the close code to this since it is
                    #   similar to what happens with a real web server (the HTTP
                    #   connection is closed with a 403 and there is no websocket
                    #   close code).
                    self._close_code = WSCloseCode.FORBIDDEN
                    self._close_reason = code_to_http_status(
                        WSCloseCode.FORBIDDEN - 3000
                    )

                self._event_handshake_complete.set()

            else:
                raise falcon_errors.OperationNotAllowed(
                    'An ASGI application must send either websocket.accept or '
                    'websocket.close before sending any other event types (got '
                    '{0})'.format(event_type)
                )

        elif self._state == _WebSocketState.ACCEPTED:
            if event_type == EventType.WS_CLOSE:
                self._state = _WebSocketState.CLOSED
                self._close_code = event.get('code', WSCloseCode.NORMAL)
                self._close_reason = event.get('reason', '')
            else:
                assert event_type == EventType.WS_SEND
                self._collected_server_events.append(event)
        else:
            assert self.closed

            # NOTE(vytas): Tweaked in Falcon 4.0: we now simulate ASGI
            #   WebSocket protocol 2.4+, raising an instance of OSError upon
            #   send if the client has already disconnected.
            raise falcon_errors.WebSocketDisconnected(self._close_code)

        # NOTE(kgriffs): Give whatever is waiting on the handshake or a
        #   collected data/text event a chance to progress.
        await asyncio.sleep(0)

    def _create_checked_disconnect(self) -> AsgiEvent:
        if self._disconnect_emitted:
            raise falcon_errors.OperationNotAllowed(
                'The websocket.disconnect event has already been emitted, '
                'and so the app should not attempt to receive any more '
                'events, since ASGI servers will likely block indefinitely '
                'rather than re-emitting websocket.disconnect events.'
            )

        self._disconnect_emitted = True
        response = {'type': EventType.WS_DISCONNECT, 'code': self._close_code}

        if self._close_reason:
            response['reason'] = self._close_reason

        return response


# 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: Mapping[str, str]) -> Optional[str]:
    """Return encoding from given HTTP Header Dict.

    Args:
        headers(dict): Dictionary from which to extract encoding. Header
            names must either be lowercase or the dict must support
            case-insensitive lookups.
    """

    content_type = headers.get('content-type')

    if not content_type:
        return None

    content_type, params = parse_header(content_type)

    if 'charset' in params:
        return params['charset'].strip('\'"')

    # NOTE(kgriffs): Added checks for text/event-stream and application/json
    if content_type in ('text/event-stream', 'application/json'):
        return 'UTF-8'

    if 'text' in content_type:
        return 'ISO-8859-1'

    return None


def get_unused_port() -> int:
    """Get an unused localhost port for use by a test server.

    Warning:
        It is possible for a third party to bind to the returned port
        before the caller is able to do so. The caller will need to
        retry with a different port in that case.

    Warning:
        This method has only be tested on POSIX systems and may not
        work elsewhere.
    """

    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(('localhost', 0))
        return s.getsockname()[1]


def rand_string(min: int, max: int) -> str:
    """Return a randomly-generated string, of a random length.

    Args:
        min (int): Minimum string length to return, inclusive
        max (int): Maximum string length to return, inclusive

    """

    int_gen = random.randint
    string_length = int_gen(min, max)
    return ''.join([chr(int_gen(ord(' '), ord('~'))) for __ in range(string_length)])


def create_scope(
    path: str = '/',
    query_string: str = '',
    method: str = 'GET',
    headers: Optional[HeaderArg] = None,
    host: str = DEFAULT_HOST,
    scheme: Optional[str] = None,
    port: Optional[int] = None,
    http_version: str = '1.1',
    remote_addr: Optional[str] = None,
    root_path: Optional[str] = None,
    content_length: Optional[int] = None,
    include_server: bool = True,
    cookies: Optional[CookieArg] = None,
) -> Dict[str, Any]:
    """Create a mock ASGI scope ``dict`` for simulating HTTP requests.

    Keyword Args:
        path (str): The path for the request (default ``'/'``)
        query_string (str): The query string to simulate, without a
            leading ``'?'`` (default ``''``). The query string is passed as-is
            (it will not be percent-encoded).
        method (str): The HTTP method to use (default ``'GET'``)
        headers (dict): Headers as a dict-like (Mapping) object, or an
            iterable yielding a series of two-member (*name*, *value*)
            iterables. Each pair of strings provides the name and value
            for an HTTP header. If desired, multiple header values may be
            combined into a single (*name*, *value*) pair by joining the values
            with a comma when the header in question supports the list
            format (see also RFC 7230 and RFC 7231). When the
            request will include a body, the Content-Length header should be
            included in this list. Header names are not case-sensitive.

            Note:
                If a User-Agent header is not provided, it will default to::

                    f'falcon-client/{falcon.__version__}'

        host(str): Hostname for the request (default ``'falconframework.org'``).
            This also determines the value of the Host header in the
            request.
        scheme (str): URL scheme, either ``'http'`` or ``'https'``
            (default ``'http'``)
        port (int): The TCP port to simulate. Defaults to
            the standard port used by the given scheme (i.e., 80 for ``'http'``
            and 443 for ``'https'``). A string may also be passed, as long as
            it can be parsed as an int.
        http_version (str): The HTTP version to simulate. Must be either
            ``'2'``, ``'2.0'``, ``'1.1'``, ``'1.0'``, or ``'1'``
            (default ``'1.1'``). If set to ``'1.0'``, the Host header will not
            be added to the scope.
        remote_addr (str): Remote address for the request to use for
            the 'client' field in the connection scope (default None)
        root_path (str): The root path this application is mounted at; same as
            SCRIPT_NAME in WSGI (default ``''``).
        content_length (int): The expected content length of the request
            body (default ``None``). If specified, this value will be
            used to set the Content-Length header in the request.
        include_server (bool): Set to ``False`` to not set the 'server' key
            in the scope ``dict`` (default ``True``).
        cookies (dict): Cookies as a dict-like (Mapping) object, or an
            iterable yielding a series of two-member (*name*, *value*)
            iterables. Each pair of items provides the name and value
            for the 'Set-Cookie' header.
    """

    http_version = _fixup_http_version(http_version)

    path = uri.decode(path, unquote_plus=False)

    # NOTE(kgriffs): Handles both None and ''
    query_string_bytes = query_string.encode() if query_string else b''

    if query_string_bytes and query_string_bytes.startswith(b'?'):
        raise ValueError("query_string should not start with '?'")

    scope: Dict[str, Any] = {
        'type': ScopeType.HTTP,
        'asgi': {
            'version': '3.0',
            'spec_version': '2.1',
        },
        'http_version': http_version,
        'method': method.upper(),
        'path': path,
        'query_string': query_string_bytes,
    }

    # NOTE(kgriffs): Explicitly test against None so that the caller
    #   is able to simulate setting app to an empty string if they
    #   need to cover that branch in their code.
    if root_path is not None:
        # NOTE(kgriffs): Judging by the algorithm given in PEP-3333 for
        #   reconstructing the URL, SCRIPT_NAME is expected to contain a
        #   preceding slash character. Since ASGI states that this value is
        #   the same as WSGI's SCRIPT_NAME, we will follow suit here.
        if root_path and not root_path.startswith('/'):
            scope['root_path'] = '/' + root_path
        else:
            scope['root_path'] = root_path

    if scheme:
        if scheme not in {'http', 'https', 'ws', 'wss'}:
            raise ValueError("scheme must be either 'http', 'https', 'ws', or 'wss'")

        scope['scheme'] = scheme

    if port is None:
        if (scheme or 'http') in {'http', 'ws'}:
            port = 80
        else:
            port = 443
    else:
        port = int(port)

    if remote_addr:
        # NOTE(kgriffs): Choose from the standard IANA dynamic range
        remote_port = random.randint(49152, 65535)

        # NOTE(kgriffs): Expose as an iterable to ensure the framework/app
        #   isn't hard-coded to only work with a list or tuple.
        scope['client'] = iter([remote_addr, remote_port])

    if include_server:
        scope['server'] = iter([host, port])

    # NOTE(myusko): Clients discard Set-Cookie header
    #  in the response to the OPTIONS method.
    if method == 'OPTIONS' and cookies is not None:
        cookies = None

    _add_headers_to_scope(
        scope, headers, content_length, host, port, scheme, http_version, cookies
    )

    return scope


def create_scope_ws(
    path: str = '/',
    query_string: str = '',
    headers: Optional[HeaderArg] = None,
    host: str = DEFAULT_HOST,
    scheme: Optional[str] = None,
    port: Optional[int] = None,
    http_version: str = '1.1',
    remote_addr: Optional[str] = None,
    root_path: Optional[str] = None,
    include_server: bool = True,
    subprotocols: Optional[str] = None,
    spec_version: str = '2.1',
) -> Dict[str, Any]:
    """Create a mock ASGI scope ``dict`` for simulating WebSocket requests.

    Keyword Args:
        path (str): The path for the request (default ``'/'``)
        query_string (str): The query string to simulate, without a
            leading ``'?'`` (default ``''``). The query string is passed as-is
            (it will not be percent-encoded).
        headers (dict): Headers as a dict-like (Mapping) object, or an
            iterable yielding a series of two-member (*name*, *value*)
            iterables. Each pair of strings provides the name and value
            for an HTTP header. If desired, multiple header values may be
            combined into a single (*name*, *value*) pair by joining the values
            with a comma when the header in question supports the list
            format (see also RFC 7230 and RFC 7231). When the
            request will include a body, the Content-Length header should be
            included in this list. Header names are not case-sensitive.

            Note:
                If a User-Agent header is not provided, it will default to::

                    f'falcon-client/{falcon.__version__}'

        host(str): Hostname for the request (default ``'falconframework.org'``).
            This also determines the value of the Host header in the
            request.
        scheme (str): URL scheme, either ``'ws'`` or ``'wss'``
            (default ``'ws'``)
        port (int): The TCP port to simulate. Defaults to
            the standard port used by the given scheme (i.e., 80 for ``'ws'``
            and 443 for ``'wss'``). A string may also be passed, as long as
            it can be parsed as an int.
        http_version (str): The HTTP version to simulate. Must be either
            ``'2'``, ``'2.0'``, or ``'1.1'`` (default ``'1.1'``).
        remote_addr (str): Remote address for the request to use for
            the 'client' field in the connection scope (default None)
        root_path (str): The root path this application is mounted at; same as
            SCRIPT_NAME in WSGI (default ``''``).
        include_server (bool): Set to ``False`` to not set the 'server' key
            in the scope ``dict`` (default ``True``).
        spec_version (str): The ASGI spec version to emulate (default ``'2.1'``).
        subprotocols (Iterable[str]): Subprotocols the client wishes to
            advertise to the server (default ``[]``).
    """

    scope = create_scope(
        path=path,
        query_string=query_string,
        headers=headers,
        host=host,
        scheme=(scheme or 'ws'),
        port=port,
        http_version=http_version,
        remote_addr=remote_addr,
        root_path=root_path,
        include_server=include_server,
    )

    scope['type'] = ScopeType.WS
    scope['asgi']['spec_version'] = spec_version
    del scope['method']

    # NOTE(kgriffiths): Explicit check against None affords simulating a request
    #   with a scope that does not contain the optional 'subprotocols' key.
    if subprotocols is not None:
        scope['subprotocols'] = subprotocols

    return scope


def create_environ(
    path: str = '/',
    query_string: str = '',
    http_version: str = '1.1',
    scheme: str = 'http',
    host: str = DEFAULT_HOST,
    port: Optional[int] = None,
    headers: Optional[HeaderArg] = None,
    app: Optional[str] = None,
    body: Union[str, bytes] = b'',
    method: str = 'GET',
    wsgierrors: Optional[TextIO] = None,
    file_wrapper: Optional[Callable[..., Any]] = None,
    remote_addr: Optional[str] = None,
    root_path: Optional[str] = None,
    cookies: Optional[CookieArg] = None,
) -> Dict[str, Any]:
    """Create a mock PEP-3333 environ ``dict`` for simulating WSGI requests.

    Keyword Args:
        path (str): The path for the request (default ``'/'``)
        query_string (str): The query string to simulate, without a
            leading ``'?'`` (default ``''``). The query string is passed as-is
            (it will not be percent-encoded).
        http_version (str): The HTTP version to simulate. Must be either
            ``'2'``, ``'2.0'``, ``'1.1'``, ``'1.0'``, or ``'1'``
            (default ``'1.1'``). If set to ``'1.0'``, the Host header will not
            be added to the scope.
        scheme (str): URL scheme, either ``'http'`` or ``'https'``
            (default ``'http'``)
        host(str): Hostname for the request (default ``'falconframework.org'``)
        port (int): The TCP port to simulate. Defaults to
            the standard port used by the given scheme (i.e., 80 for ``'http'``
            and 443 for ``'https'``). A string may also be passed, as long as
            it can be parsed as an int.
        headers (dict): Headers as a dict-like (Mapping) object, or an
            iterable yielding a series of two-member (*name*, *value*)
            iterables. Each pair of strings provides the name and value
            for an HTTP header. If desired, multiple header values may be
            combined into a single (*name*, *value*) pair by joining the values
            with a comma when the header in question supports the list
            format (see also RFC 7230 and RFC 7231). Header names are not
            case-sensitive.

            Note:
                If a User-Agent header is not provided, it will default to::

                    f'falcon-client/{falcon.__version__}'

        root_path (str): Value for the ``SCRIPT_NAME`` environ variable, described in
            PEP-3333: 'The initial portion of the request URL's "path" that
            corresponds to the application object, so that the application
            knows its virtual "location". This may be an empty string, if the
            application corresponds to the "root" of the server.' (default ``''``)
        app (str): Deprecated alias for `root_path`. If both kwargs are passed,
            `root_path` takes precedence.
        body (str): The body of the request (default ``''``). The value will be
            encoded as UTF-8 in the WSGI environ. Alternatively, a byte string
            may be passed, in which case it will be used as-is.
        method (str): The HTTP method to use (default ``'GET'``)
        wsgierrors (io): The stream to use as *wsgierrors*
            (default ``sys.stderr``)
        file_wrapper: Callable that returns an iterable, to be used as
            the value for *wsgi.file_wrapper* in the environ.
        remote_addr (str): Remote address for the request to use as the
            ``'REMOTE_ADDR'`` environ variable (default ``None``)
        cookies (dict): Cookies as a dict-like (Mapping) object, or an
            iterable yielding a series of two-member (*name*, *value*)
            iterables. Each pair of items provides the name and value
            for the Set-Cookie header.

    """

    http_version = _fixup_http_version(http_version)

    if query_string and query_string.startswith('?'):
        raise ValueError("query_string should not start with '?'")

    body_bytes = io.BytesIO(body.encode() if isinstance(body, str) else body)

    # NOTE(kgriffs): wsgiref, gunicorn, and uWSGI all unescape
    # the paths before setting PATH_INFO but preserve raw original
    raw_path = path
    path = uri.decode(path, unquote_plus=False)

    # NOTE(kgriffs): The decoded path may contain UTF-8 characters.
    # But according to the WSGI spec, no strings can contain chars
    # outside ISO-8859-1. Therefore, to reconcile the URI
    # encoding standard that allows UTF-8 with the WSGI spec
    # that does not, WSGI servers tunnel the string via
    # ISO-8859-1. falcon.testing.create_environ() mimics this
    # behavior, e.g.:
    #
    #   tunnelled_path = path.encode('utf-8').decode('iso-8859-1')
    #
    # falcon.Request does the following to reverse the process:
    #
    #   path = tunnelled_path.encode('iso-8859-1').decode('utf-8', 'replace')
    #
    path = path.encode().decode('iso-8859-1')

    scheme = scheme.lower()
    if port is None:
        port_str = '80' if scheme == 'http' else '443'
    else:
        # NOTE(kgriffs): Running it through int() first ensures that if
        #   a string was passed, it is a valid integer.
        port_str = str(int(port))

    root_path = root_path or app or ''

    # NOTE(kgriffs): Judging by the algorithm given in PEP-3333 for
    # reconstructing the URL, SCRIPT_NAME is expected to contain a
    # preceding slash character.
    if root_path and not root_path.startswith('/'):
        root_path = '/' + root_path

    env: dict[str, Any] = {
        'SERVER_PROTOCOL': 'HTTP/' + http_version,
        'SERVER_SOFTWARE': 'gunicorn/0.17.0',
        'SCRIPT_NAME': (root_path or ''),
        'REQUEST_METHOD': method,
        'PATH_INFO': path,
        'QUERY_STRING': query_string,
        'REMOTE_PORT': '65133',
        'RAW_URI': raw_path,
        'SERVER_NAME': host,
        'SERVER_PORT': port_str,
        'wsgi.version': (1, 0),
        'wsgi.url_scheme': scheme,
        'wsgi.input': body_bytes,
        'wsgi.errors': wsgierrors or sys.stderr,
        'wsgi.multithread': False,
        'wsgi.multiprocess': True,
        'wsgi.run_once': False,
    }

    # NOTE(kgriffs): It has been observed that WSGI servers do not always
    #   set the REMOTE_ADDR variable, so we don't always set it either, to
    #   ensure the framework/app handles that case correctly.
    if remote_addr:
        env['REMOTE_ADDR'] = remote_addr

    if file_wrapper is not None:
        env['wsgi.file_wrapper'] = file_wrapper

    if http_version != '1.0':
        host_header = host

        if scheme == 'https':
            if port_str != '443':
                host_header += ':' + port_str
        else:
            if port_str != '80':
                host_header += ':' + port_str

        env['HTTP_HOST'] = host_header

    content_length = body_bytes.seek(0, 2)
    body_bytes.seek(0)

    if content_length != 0:
        env['CONTENT_LENGTH'] = str(content_length)

    # NOTE(myusko): Clients discard Set-Cookie header
    #  in the response to the OPTIONS method.
    if cookies is not None and method != 'OPTIONS':
        env['HTTP_COOKIE'] = _make_cookie_values(cookies)

    _add_headers_to_environ(env, headers)

    return env


def create_req(
    options: Optional[falcon.request.RequestOptions] = None, **kwargs: Any
) -> falcon.Request:
    """Create and return a new Request instance.

    This function can be used to conveniently create a WSGI environ
    and use it to instantiate a :class:`falcon.Request` object in one go.

    The arguments for this function are identical to those
    of :meth:`falcon.testing.create_environ`, except an additional
    `options` keyword argument may be set to an instance of
    :class:`falcon.RequestOptions` to configure certain
    aspects of request parsing in lieu of the defaults.
    """

    env = create_environ(**kwargs)
    return falcon.request.Request(env, options=options)


def create_asgi_req(
    body: Optional[bytes] = None,
    req_type: Optional[Type[falcon.asgi.Request]] = None,
    options: Optional[falcon.request.RequestOptions] = None,
    **kwargs: Any,
) -> falcon.asgi.Request:
    """Create and return a new ASGI Request instance.

    This function can be used to conveniently create an ASGI scope
    and use it to instantiate a :class:`falcon.asgi.Request` object
    in one go.

    The arguments for this function are identical to those
    of :meth:`falcon.testing.create_scope`, with the addition of
    `body`, `req_type`, and `options` arguments as documented below.

    Keyword Arguments:
        body (bytes): The body data to use for the request (default b''). If
            the value is a :class:`str`, it will be UTF-8 encoded to
            a byte string.
        req_type (object): A subclass of :class:`falcon.asgi.Request`
            to instantiate. If not specified, the standard
            :class:`falcon.asgi.Request` class will simply be used.
        options (falcon.RequestOptions): An instance of
            :class:`falcon.RequestOptions` that should be used to determine
            certain aspects of request parsing in lieu of the defaults.
    """

    scope = create_scope(**kwargs)

    body = body or b''
    disconnect_at = time.time() + 300

    req_event_emitter = ASGIRequestEventEmitter(body, disconnect_at=disconnect_at)

    req_type = req_type or falcon.asgi.Request
    return req_type(scope, req_event_emitter, options=options)


@contextlib.contextmanager
def redirected(
    stdout: TextIO = sys.stdout, stderr: TextIO = sys.stderr
) -> Iterator[None]:
    """Redirect stdout or stderr temporarily.

    e.g.:

    with redirected(stderr=os.devnull):
        ...
    """

    old_stdout, old_stderr = sys.stdout, sys.stderr
    sys.stdout, sys.stderr = stdout, stderr
    try:
        yield
    finally:
        sys.stderr, sys.stdout = old_stderr, old_stdout


def closed_wsgi_iterable(iterable: Iterable[bytes]) -> Iterable[bytes]:
    """Wrap an iterable to ensure its ``close()`` method is called.

    Wraps the given `iterable` in an iterator utilizing a ``for`` loop as
    illustrated in
    `the PEP-3333 server/gateway side example
    <https://www.python.org/dev/peps/pep-3333/#the-server-gateway-side>`_.
    Finally, if the iterable has a ``close()`` method, it is called upon
    exception or exhausting iteration.

    Furthermore, the first bytestring yielded from iteration, if any, is
    prefetched before returning the wrapped iterator in order to ensure the
    WSGI ``start_response`` function is called even if the WSGI application is
    a generator.

    Args:
        iterable (iterable): An iterable that yields zero or more
            bytestrings, per PEP-3333

    Returns:
        iterator: An iterator yielding the same bytestrings as `iterable`
    """

    def wrapper() -> Iterator[bytes]:
        try:
            for item in iterable:
                yield item
        finally:
            if hasattr(iterable, 'close'):
                iterable.close()

    wrapped = wrapper()
    head: Tuple[bytes, ...]
    try:
        head = (next(wrapped),)
    except StopIteration:
        head = ()
    return itertools.chain(head, wrapped)


# ---------------------------------------------------------------------
# Private
# ---------------------------------------------------------------------


def _add_headers_to_environ(env: Dict[str, Any], headers: Optional[HeaderArg]) -> None:
    if headers:
        try:
            items = headers.items()  # type: ignore[union-attr]
        except AttributeError:
            items = headers

        for name, value in items:
            name_wsgi = name.upper().replace('-', '_')
            if name_wsgi not in ('CONTENT_TYPE', 'CONTENT_LENGTH'):
                name_wsgi = 'HTTP_' + name_wsgi

            if value is None:
                value = ''
            else:
                value = value.strip()

            if name_wsgi not in env or name.lower() in SINGLETON_HEADERS:
                env[name_wsgi] = value
            else:
                env[name_wsgi] += ',' + value

    env.setdefault('HTTP_USER_AGENT', DEFAULT_UA)


def _add_headers_to_scope(
    scope: dict[str, Any],
    headers: Optional[HeaderArg],
    content_length: Optional[int],
    host: str,
    port: int,
    scheme: Optional[str],
    http_version: str,
    cookies: Optional[CookieArg],
) -> None:
    found_ua = False
    prepared_headers: List[Iterable[bytes]] = []

    if headers:
        try:
            items = headers.items()  # type: ignore[union-attr]
        except AttributeError:
            items = headers

        for name, value in items:
            n = name.lower().encode('latin1')
            found_ua = found_ua or (n == b'user-agent')

            # NOTE(kgriffs): Value is stripped if not empty, otherwise defaults
            #   to b'' to be consistent with _add_headers_to_environ().
            v = b'' if value is None else value.strip().encode('latin1')

            # NOTE(kgriffs): Expose as an iterable to ensure the framework/app
            #   isn't hard-coded to only work with a list or tuple.
            prepared_headers.append(iter([n, v]))

    if not found_ua:
        prepared_headers.append([b'user-agent', DEFAULT_UA.encode()])

    if content_length is not None:
        value = str(content_length).encode()
        prepared_headers.append((b'content-length', value))

    if http_version != '1.0':
        host_header = host

        if scheme == 'https':
            if port != 443:
                host_header += ':' + str(port)
        else:
            if port != 80:
                host_header += ':' + str(port)

        prepared_headers.append([b'host', host_header.encode()])

    if cookies is not None:
        prepared_headers.append([b'cookie', _make_cookie_values(cookies).encode()])

    # NOTE(kgriffs): Make it an iterator to ensure the app is not expecting
    #   a specific type (ASGI only specified that it is an iterable).
    scope['headers'] = iter(prepared_headers)


def _fixup_http_version(http_version: str) -> str:
    if http_version not in ('2', '2.0', '1.1', '1.0', '1'):
        raise ValueError('Invalid http_version specified: ' + http_version)

    # NOTE(kgrifs): Normalize so that they conform to the standard
    #   protocol names with prefixed with "HTTP/"
    if http_version == '2.0':
        http_version = '2'
    elif http_version == '1':
        http_version = '1.0'

    return http_version


def _make_cookie_values(cookies: CookieArg) -> str:
    return '; '.join(
        [
            '{}={}'.format(key, cookie.value if hasattr(cookie, 'value') else cookie)
            for key, cookie in cookies.items()
        ]
    )