Repository: aio-libs/aiohttp
Pull request: Refactor WSMessage to use tagged unions
This reworks WSMessage to be a union of dataclasses which can provide much better type safety using tagged unions (on the `type` attribute).

I need to go over a couple of details (like what should be exported and from where, imports are a little messy currently). But, my main question before I finish it off, is whether it makes sense to have `.json()` only on the TEXT/BINARY messages? If we do this, it becomes easy for mypy to validate the code, requiring a `msg.type is WSMsgType.TEXT` check before calling `.json()`.

Fixes #7313 

Problem statement:
websocket: message.json() fails with a TypeError when the message exceeds the size limit
### Describe the bug

When a websocket message exceeds the size limit, calling message.json() raises a TypeError instead of an informative exception


### To Reproduce

Send a large websocket message to an aiohttp server.

In the handler, calling message.json() raises TypeError:

    TypeError: the JSON object must be str, bytes or bytearray, not WebSocketError

`message.data` is set to a WebSocketError instance:

    aiohttp.http_websocket.WebSocketError: Message size 29818 exceeds limit 8192

(I lowered the limit to more easily trigger the problem)

### Expected behavior

I would expect calling `message.json()` to raise the WebSocketError.

### Logs/tracebacks

```python-traceback
Traceback (most recent call last):
  File "/Users/just/code/git/BlackFoundry/fontra/src/fontra/core/remote.py", line 47, in _handleConnection
    message = message.json()
              ^^^^^^^^^^^^^^
  File "/Users/just/code/git/BlackFoundry/fontra/venv/lib/python3.11/site-packages/aiohttp/http_websocket.py", line 98, in json
    return loads(self.data)
           ^^^^^^^^^^^^^^^^
  File "/Library/Frameworks/Python.framework/Versions/3.11/lib/python3.11/json/__init__.py", line 339, in loads
    raise TypeError(f'the JSON object must be str, bytes or bytearray, '
TypeError: the JSON object must be str, bytes or bytearray, not WebSocketError
```


### Python Version

```console
Python 3.11.2
```


### aiohttp Version

```console
Name: aiohttp
Version: 3.8.4
```


### multidict Version

```console
Name: multidict
Version: 6.0.4
```


### yarl Version

```console
Name: yarl
Version: 1.8.2
```


### OS

macOS 10.15

### Related component

Server

### Additional context

_No response_

### Code of Conduct

- [X] I agree to follow the aio-libs Code of Conduct

Hints from issue discussion:
OK, I've been working through a PR to try and make this more type safe.

2 options we can do. First is to raise the exception from `.json()`, but I'm not sure how much sense this makes, and it would still fail on other message types (e.g. CLOSE has an int, which would still produce a TypeError). This would not affect type safety.

Second is to have `.json()` only on TEXT and BINARY messages. This means the code would need to do something like `if msg.type in (WSMsgType.TEXT, WSMsgType.BINARY)`. But, this could be checked with mypy etc. If you don't check the type it'll complain that there's no `.json()` method.
Note that you can implement the second approach in your code now. My changes would just make it a requirement and flag up any issues when checked with mypy.
I think I am puzzled (as a newbie aiohttp user) why the error is delivered as a message instead of a raised exception.
I'm not too sure what the reasoning is for this, but you'll note from the websockets example that the expected behaviour is to check for WSMsgType.ERROR and then break (because the connection will be closed on an error):
https://docs.aiohttp.org/en/stable/client_quickstart.html#websockets

The data will only be an exception if the type is ERROR.

Maybe @asvetlov can provide some reasoning why we don't raise the exception.
> The data will only be an exception if the type is ERROR.

Thanks, I've adjusted my code to check for `WSMsgType.ERROR`.

Commit message: [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Patch to review:
diff --git a/aiohttp/__init__.py b/aiohttp/__init__.py
index dc11fefc949..a0e7cfd1462 100644
--- a/aiohttp/__init__.py
+++ b/aiohttp/__init__.py
@@ -55,7 +55,7 @@
     HttpVersion11,
     WebSocketError,
     WSCloseCode,
-    WSMessage,
+    WSMessageType,
     WSMsgType,
 )
 from .multipart import (
@@ -173,7 +173,7 @@
     "HttpVersion11",
     "WSMsgType",
     "WSCloseCode",
-    "WSMessage",
+    "WSMessageType",
     "WebSocketError",
     # multipart
     "BadContentDispositionHeader",
diff --git a/aiohttp/client.py b/aiohttp/client.py
index 343d20436e7..a27110a8fa4 100644
--- a/aiohttp/client.py
+++ b/aiohttp/client.py
@@ -101,7 +101,7 @@
     strip_auth_from_url,
 )
 from .http import WS_KEY, HttpVersion, WebSocketReader, WebSocketWriter
-from .http_websocket import WSHandshakeError, WSMessage, ws_ext_gen, ws_ext_parse
+from .http_websocket import WSHandshakeError, WSMessageType, ws_ext_gen, ws_ext_parse
 from .streams import FlowControlDataQueue
 from .tracing import Trace, TraceConfig
 from .typedefs import JSONEncoder, LooseCookies, LooseHeaders, Query, StrOrURL
@@ -1033,7 +1033,7 @@ async def _ws_connect(
 
             transport = conn.transport
             assert transport is not None
-            reader: FlowControlDataQueue[WSMessage] = FlowControlDataQueue(
+            reader: FlowControlDataQueue[WSMessageType] = FlowControlDataQueue(
                 conn_proto, 2**16, loop=self._loop
             )
             conn_proto.set_parser(WebSocketReader(reader, max_msg_size), reader)
diff --git a/aiohttp/client_ws.py b/aiohttp/client_ws.py
index 850dd8ca8a8..4756f544d20 100644
--- a/aiohttp/client_ws.py
+++ b/aiohttp/client_ws.py
@@ -4,20 +4,19 @@
 import dataclasses
 import sys
 from types import TracebackType
-from typing import Any, Final, Optional, Type, cast
+from typing import Any, Final, Optional, Type
 
 from .client_exceptions import ClientError, ServerTimeoutError
 from .client_reqrep import ClientResponse
 from .helpers import calculate_timeout_when, set_result
 from .http import (
-    WS_CLOSED_MESSAGE,
-    WS_CLOSING_MESSAGE,
     WebSocketError,
     WSCloseCode,
-    WSMessage,
+    WSMessageClosed,
+    WSMessageClosing,
     WSMsgType,
 )
-from .http_websocket import WebSocketWriter  # WSMessage
+from .http_websocket import WebSocketWriter, WSMessageError, WSMessageType
 from .streams import EofStream, FlowControlDataQueue
 from .typedefs import (
     DEFAULT_JSON_DECODER,
@@ -46,7 +45,7 @@ class ClientWSTimeout:
 class ClientWebSocketResponse:
     def __init__(
         self,
-        reader: "FlowControlDataQueue[WSMessage]",
+        reader: "FlowControlDataQueue[WSMessageType]",
         writer: WebSocketWriter,
         protocol: Optional[str],
         response: ClientResponse,
@@ -173,7 +172,7 @@ def _handle_ping_pong_exception(self, exc: BaseException) -> None:
         self._exception = exc
         self._response.close()
         if self._waiting and not self._closing:
-            self._reader.feed_data(WSMessage(WSMsgType.ERROR, exc, None))
+            self._reader.feed_data(WSMessageError(data=exc, extra=None))
 
     def _set_closed(self) -> None:
         """Set the connection to closed.
@@ -264,7 +263,7 @@ async def close(self, *, code: int = WSCloseCode.OK, message: bytes = b"") -> bo
             assert self._loop is not None
             self._close_wait = self._loop.create_future()
             self._set_closing()
-            self._reader.feed_data(WS_CLOSING_MESSAGE)
+            self._reader.feed_data(WSMessageClosing())
             await self._close_wait
 
         if not self._closed:
@@ -306,7 +305,7 @@ async def close(self, *, code: int = WSCloseCode.OK, message: bytes = b"") -> bo
         else:
             return False
 
-    async def receive(self, timeout: Optional[float] = None) -> WSMessage:
+    async def receive(self, timeout: Optional[float] = None) -> WSMessageType:
         receive_timeout = timeout or self._timeout.ws_receive
 
         while True:
@@ -314,10 +313,10 @@ async def receive(self, timeout: Optional[float] = None) -> WSMessage:
                 raise RuntimeError("Concurrent call to receive() is not allowed")
 
             if self._closed:
-                return WS_CLOSED_MESSAGE
+                return WSMessageClosed()
             elif self._closing:
                 await self.close()
-                return WS_CLOSED_MESSAGE
+                return WSMessageClosed()
 
             try:
                 self._waiting = True
@@ -342,22 +341,22 @@ async def receive(self, timeout: Optional[float] = None) -> WSMessage:
             except EofStream:
                 self._close_code = WSCloseCode.OK
                 await self.close()
-                return WSMessage(WSMsgType.CLOSED, None, None)
+                return WSMessageClosed()
             except ClientError:
                 # Likely ServerDisconnectedError when connection is lost
                 self._set_closed()
                 self._close_code = WSCloseCode.ABNORMAL_CLOSURE
-                return WS_CLOSED_MESSAGE
+                return WSMessageClosed()
             except WebSocketError as exc:
                 self._close_code = exc.code
                 await self.close(code=exc.code)
-                return WSMessage(WSMsgType.ERROR, exc, None)
+                return WSMessageError(data=exc)
             except Exception as exc:
                 self._exception = exc
                 self._set_closing()
                 self._close_code = WSCloseCode.ABNORMAL_CLOSURE
                 await self.close()
-                return WSMessage(WSMsgType.ERROR, exc, None)
+                return WSMessageError(data=exc)
 
             if msg.type is WSMsgType.CLOSE:
                 self._set_closing()
@@ -379,13 +378,13 @@ async def receive_str(self, *, timeout: Optional[float] = None) -> str:
         msg = await self.receive(timeout)
         if msg.type is not WSMsgType.TEXT:
             raise TypeError(f"Received message {msg.type}:{msg.data!r} is not str")
-        return cast(str, msg.data)
+        return msg.data
 
     async def receive_bytes(self, *, timeout: Optional[float] = None) -> bytes:
         msg = await self.receive(timeout)
         if msg.type is not WSMsgType.BINARY:
             raise TypeError(f"Received message {msg.type}:{msg.data!r} is not bytes")
-        return cast(bytes, msg.data)
+        return msg.data
 
     async def receive_json(
         self,
@@ -399,7 +398,7 @@ async def receive_json(
     def __aiter__(self) -> "ClientWebSocketResponse":
         return self
 
-    async def __anext__(self) -> WSMessage:
+    async def __anext__(self) -> WSMessageType:
         msg = await self.receive()
         if msg.type in (WSMsgType.CLOSE, WSMsgType.CLOSING, WSMsgType.CLOSED):
             raise StopAsyncIteration
diff --git a/aiohttp/http.py b/aiohttp/http.py
index 244d71c4197..f3e2203623c 100644
--- a/aiohttp/http.py
+++ b/aiohttp/http.py
@@ -11,14 +11,14 @@
     RawResponseMessage,
 )
 from .http_websocket import (
-    WS_CLOSED_MESSAGE,
-    WS_CLOSING_MESSAGE,
     WS_KEY,
     WebSocketError,
     WebSocketReader,
     WebSocketWriter,
     WSCloseCode,
-    WSMessage,
+    WSMessageClosed,
+    WSMessageClosing,
+    WSMessageType,
     WSMsgType,
     ws_ext_gen,
     ws_ext_parse,
@@ -41,14 +41,14 @@
     "RawRequestMessage",
     "RawResponseMessage",
     # .http_websocket
-    "WS_CLOSED_MESSAGE",
-    "WS_CLOSING_MESSAGE",
+    "WSMessageClosed",
+    "WSMessageClosing",
     "WS_KEY",
     "WebSocketReader",
     "WebSocketWriter",
     "ws_ext_gen",
     "ws_ext_parse",
-    "WSMessage",
+    "WSMessageType",
     "WebSocketError",
     "WSMsgType",
     "WSCloseCode",
diff --git a/aiohttp/http_websocket.py b/aiohttp/http_websocket.py
index 3fe7af4e32c..0a3ea1fda8d 100644
--- a/aiohttp/http_websocket.py
+++ b/aiohttp/http_websocket.py
@@ -7,6 +7,7 @@
 import re
 import sys
 import zlib
+from dataclasses import dataclass
 from enum import IntEnum
 from functools import partial
 from struct import Struct
@@ -14,14 +15,18 @@
     Any,
     Callable,
     Final,
+    Generic,
     List,
+    Literal,
     NamedTuple,
     Optional,
     Pattern,
     Set,
     Tuple,
+    TypeVar,
     Union,
     cast,
+    overload,
 )
 
 from .base_protocol import BaseProtocol
@@ -110,26 +115,90 @@ class WSMsgType(IntEnum):
 MASK_LEN: Final[int] = 4
 
 
-class WSMessage(NamedTuple):
+@dataclass
+class _WSMessage:
+    data: object
     type: WSMsgType
-    # To type correctly, this would need some kind of tagged union for each type.
-    data: Any
-    extra: Optional[str]
+    extra: Optional[str] = None
 
-    def json(self, *, loads: Callable[[Any], Any] = json.loads) -> Any:
-        """Return parsed JSON data.
 
-        .. versionadded:: 0.22
-        """
+@dataclass
+class WSMessageContinuation(_WSMessage):
+    data: bytes
+    type: Literal[WSMsgType.CONTINUATION] = WSMsgType.CONTINUATION
+
+
+@dataclass
+class WSMessageText(_WSMessage):
+    data: str
+    type: Literal[WSMsgType.TEXT] = WSMsgType.TEXT
+
+    def json(
+        self, *, loads: Callable[[Union[str, bytes, bytearray]], Any] = json.loads
+    ) -> Any:
+        """Return parsed JSON data."""
+        return loads(self.data)
+
+
+@dataclass
+class WSMessageBinary(_WSMessage):
+    data: bytes
+    type: Literal[WSMsgType.BINARY] = WSMsgType.BINARY
+
+    def json(
+        self, *, loads: Callable[[Union[str, bytes, bytearray]], Any] = json.loads
+    ) -> Any:
+        """Return parsed JSON data."""
         return loads(self.data)
 
 
-# Constructing the tuple directly to avoid the overhead of
-# the lambda and arg processing since NamedTuples are constructed
-# with a run time built lambda
-# https://github.com/python/cpython/blob/d83fcf8371f2f33c7797bc8f5423a8bca8c46e5c/Lib/collections/__init__.py#L441
-WS_CLOSED_MESSAGE = tuple.__new__(WSMessage, (WSMsgType.CLOSED, None, None))
-WS_CLOSING_MESSAGE = tuple.__new__(WSMessage, (WSMsgType.CLOSING, None, None))
+@dataclass
+class WSMessagePing(_WSMessage):
+    data: bytes
+    type: Literal[WSMsgType.PING] = WSMsgType.PING
+
+
+@dataclass
+class WSMessagePong(_WSMessage):
+    data: bytes
+    type: Literal[WSMsgType.PONG] = WSMsgType.PONG
+
+
+@dataclass
+class WSMessageClose(_WSMessage):
+    data: int
+    type: Literal[WSMsgType.CLOSE] = WSMsgType.CLOSE
+
+
+@dataclass
+class WSMessageClosing(_WSMessage):
+    data: None = None
+    type: Literal[WSMsgType.CLOSING] = WSMsgType.CLOSING
+
+
+@dataclass
+class WSMessageClosed(_WSMessage):
+    data: None = None
+    type: Literal[WSMsgType.CLOSED] = WSMsgType.CLOSED
+
+
+@dataclass
+class WSMessageError(_WSMessage):
+    data: BaseException
+    type: Literal[WSMsgType.ERROR] = WSMsgType.ERROR
+
+
+WSMessageType = Union[
+    WSMessageContinuation,
+    WSMessageText,
+    WSMessageBinary,
+    WSMessagePing,
+    WSMessagePong,
+    WSMessageClose,
+    WSMessageClosing,
+    WSMessageClosed,
+    WSMessageError,
+]
 
 
 class WebSocketError(Exception):
@@ -287,7 +356,7 @@ class WSParserState(IntEnum):
 
 class WebSocketReader:
     def __init__(
-        self, queue: DataQueue[WSMessage], max_msg_size: int, compress: bool = True
+        self, queue: DataQueue[WSMessageType], max_msg_size: int, compress: bool = True
     ) -> None:
         self.queue = queue
         self._max_msg_size = max_msg_size
@@ -327,6 +396,7 @@ def feed_data(self, data: bytes) -> Tuple[bool, bytes]:
         return False, b""
 
     def _feed_data(self, data: bytes) -> None:
+        msg: WSMessageType
         for fin, opcode, payload, compressed in self.parse_frame(data):
             if opcode in MESSAGE_TYPES_WITH_CONTENT:
                 # load text/binary
@@ -406,13 +476,11 @@ def _feed_data(self, data: bytes) -> None:
                             WSCloseCode.INVALID_TEXT, "Invalid UTF-8 text message"
                         ) from exc
 
-                    # tuple.__new__ is used to avoid the overhead of the lambda
-                    msg = tuple.__new__(WSMessage, (WSMsgType.TEXT, text, ""))
+                    msg = WSMessageText(data=text, extra="")
                     self.queue.feed_data(msg)
                     continue
 
-                # tuple.__new__ is used to avoid the overhead of the lambda
-                msg = tuple.__new__(WSMessage, (WSMsgType.BINARY, payload_merged, ""))
+                msg = WSMessageBinary(data=payload_merged, extra="")
                 self.queue.feed_data(msg)
             elif opcode == WSMsgType.CLOSE:
                 if len(payload) >= 2:
@@ -428,25 +496,23 @@ def _feed_data(self, data: bytes) -> None:
                         raise WebSocketError(
                             WSCloseCode.INVALID_TEXT, "Invalid UTF-8 text message"
                         ) from exc
-                    msg = tuple.__new__(
-                        WSMessage, (WSMsgType.CLOSE, close_code, close_message)
-                    )
+                    msg = WSMessageClose(data=close_code, extra=close_message)
                 elif payload:
                     raise WebSocketError(
                         WSCloseCode.PROTOCOL_ERROR,
                         f"Invalid close frame: {fin} {opcode} {payload!r}",
                     )
                 else:
-                    msg = tuple.__new__(WSMessage, (WSMsgType.CLOSE, 0, ""))
+                    msg = WSMessageClose(data=0, extra="")
 
                 self.queue.feed_data(msg)
 
             elif opcode == WSMsgType.PING:
-                msg = tuple.__new__(WSMessage, (WSMsgType.PING, payload, ""))
+                msg = WSMessagePing(data=payload, extra="")
                 self.queue.feed_data(msg)
 
             elif opcode == WSMsgType.PONG:
-                msg = tuple.__new__(WSMessage, (WSMsgType.PONG, payload, ""))
+                msg = WSMessagePong(data=payload, extra="")
                 self.queue.feed_data(msg)
 
             else:
diff --git a/aiohttp/web_ws.py b/aiohttp/web_ws.py
index 56e4ad1cf8e..74b4ce5f954 100644
--- a/aiohttp/web_ws.py
+++ b/aiohttp/web_ws.py
@@ -13,18 +13,19 @@
 from .abc import AbstractStreamWriter
 from .helpers import calculate_timeout_when, set_exception, set_result
 from .http import (
-    WS_CLOSED_MESSAGE,
-    WS_CLOSING_MESSAGE,
     WS_KEY,
     WebSocketError,
     WebSocketReader,
     WebSocketWriter,
     WSCloseCode,
-    WSMessage,
-    WSMsgType as WSMsgType,
+    WSMessageClosed,
+    WSMessageClosing,
+    WSMessageType,
+    WSMsgType,
     ws_ext_gen,
     ws_ext_parse,
 )
+from .http_websocket import WSMessageError
 from .log import ws_logger
 from .streams import EofStream, FlowControlDataQueue
 from .typedefs import JSONDecoder, JSONEncoder
@@ -100,7 +101,7 @@ def __init__(
         self._protocols = protocols
         self._ws_protocol: Optional[str] = None
         self._writer: Optional[WebSocketWriter] = None
-        self._reader: Optional[FlowControlDataQueue[WSMessage]] = None
+        self._reader: Optional[FlowControlDataQueue[WSMessageType]] = None
         self._closed = False
         self._closing = False
         self._conn_lost = 0
@@ -210,7 +211,7 @@ def _handle_ping_pong_exception(self, exc: BaseException) -> None:
         self._set_code_close_transport(WSCloseCode.ABNORMAL_CLOSURE)
         self._exception = exc
         if self._waiting and not self._closing and self._reader is not None:
-            self._reader.feed_data(WSMessage(WSMsgType.ERROR, exc, None))
+            self._reader.feed_data(WSMessageError(data=exc, extra=None))
 
     def _set_closed(self) -> None:
         """Set the connection to closed.
@@ -479,7 +480,7 @@ async def close(
             assert self._loop is not None
             assert self._close_wait is None
             self._close_wait = self._loop.create_future()
-            reader.feed_data(WS_CLOSING_MESSAGE)
+            reader.feed_data(WSMessageClosing())
             await self._close_wait
 
         if self._closing:
@@ -505,13 +506,13 @@ async def close(
         self._exception = asyncio.TimeoutError()
         return True
 
-    def _set_closing(self, code: WSCloseCode) -> None:
+    def _set_closing(self, code: int) -> None:
         """Set the close code and mark the connection as closing."""
         self._closing = True
         self._close_code = code
         self._cancel_heartbeat()
 
-    def _set_code_close_transport(self, code: WSCloseCode) -> None:
+    def _set_code_close_transport(self, code: int) -> None:
         """Set the close code and close the transport."""
         self._close_code = code
         self._close_transport()
@@ -521,7 +522,7 @@ def _close_transport(self) -> None:
         if self._req is not None and self._req.transport is not None:
             self._req.transport.close()
 
-    async def receive(self, timeout: Optional[float] = None) -> WSMessage:
+    async def receive(self, timeout: Optional[float] = None) -> WSMessageType:
         if self._reader is None:
             raise RuntimeError("Call .prepare() first")
 
@@ -536,9 +537,9 @@ async def receive(self, timeout: Optional[float] = None) -> WSMessage:
                 self._conn_lost += 1
                 if self._conn_lost >= THRESHOLD_CONNLOST_ACCESS:
                     raise RuntimeError("WebSocket connection is closed.")
-                return WS_CLOSED_MESSAGE
+                return WSMessageClosed()
             elif self._closing:
-                return WS_CLOSING_MESSAGE
+                return WSMessageClosing()
 
             try:
                 self._waiting = True
@@ -562,16 +563,16 @@ async def receive(self, timeout: Optional[float] = None) -> WSMessage:
             except EofStream:
                 self._close_code = WSCloseCode.OK
                 await self.close()
-                return WSMessage(WSMsgType.CLOSED, None, None)
+                return WSMessageClosed()
             except WebSocketError as exc:
                 self._close_code = exc.code
                 await self.close(code=exc.code)
-                return WSMessage(WSMsgType.ERROR, exc, None)
+                return WSMessageError(data=exc)
             except Exception as exc:
                 self._exception = exc
                 self._set_closing(WSCloseCode.ABNORMAL_CLOSURE)
                 await self.close()
-                return WSMessage(WSMsgType.ERROR, exc, None)
+                return WSMessageError(data=exc)
 
             if msg.type is WSMsgType.CLOSE:
                 self._set_closing(msg.data)
@@ -600,13 +601,13 @@ async def receive_str(self, *, timeout: Optional[float] = None) -> str:
                     msg.type, msg.data
                 )
             )
-        return cast(str, msg.data)
+        return msg.data
 
     async def receive_bytes(self, *, timeout: Optional[float] = None) -> bytes:
         msg = await self.receive(timeout)
         if msg.type is not WSMsgType.BINARY:
             raise TypeError(f"Received message {msg.type}:{msg.data!r} is not bytes")
-        return cast(bytes, msg.data)
+        return msg.data
 
     async def receive_json(
         self, *, loads: JSONDecoder = json.loads, timeout: Optional[float] = None
@@ -620,7 +621,7 @@ async def write(self, data: bytes) -> None:
     def __aiter__(self) -> "WebSocketResponse":
         return self
 
-    async def __anext__(self) -> WSMessage:
+    async def __anext__(self) -> WSMessageType:
         msg = await self.receive()
         if msg.type in (WSMsgType.CLOSE, WSMsgType.CLOSING, WSMsgType.CLOSED):
             raise StopAsyncIteration
diff --git a/examples/client_ws.py b/examples/client_ws.py
index 6f9a2c03cf3..e6981886849 100755
--- a/examples/client_ws.py
+++ b/examples/client_ws.py
@@ -16,20 +16,20 @@ async def dispatch(ws: aiohttp.ClientWebSocketResponse) -> None:
         while True:
             msg = await ws.receive()
 
-            if msg.type == aiohttp.WSMsgType.TEXT:
+            if msg.type is aiohttp.WSMsgType.TEXT:
                 print("Text: ", msg.data.strip())
-            elif msg.type == aiohttp.WSMsgType.BINARY:
+            elif msg.type is aiohttp.WSMsgType.BINARY:
                 print("Binary: ", msg.data)
-            elif msg.type == aiohttp.WSMsgType.PING:
+            elif msg.type is aiohttp.WSMsgType.PING:
                 await ws.pong()
-            elif msg.type == aiohttp.WSMsgType.PONG:
+            elif msg.type is aiohttp.WSMsgType.PONG:
                 print("Pong received")
             else:
-                if msg.type == aiohttp.WSMsgType.CLOSE:
+                if msg.type is aiohttp.WSMsgType.CLOSE:
                     await ws.close()
-                elif msg.type == aiohttp.WSMsgType.ERROR:
+                elif msg.type is aiohttp.WSMsgType.ERROR:
                     print("Error during receive %s" % ws.exception())
-                elif msg.type == aiohttp.WSMsgType.CLOSED:
+                elif msg.type is aiohttp.WSMsgType.CLOSED:
                     pass
 
                 break
diff --git a/examples/server_simple.py b/examples/server_simple.py
index c68e141b33e..c602bff426d 100644
--- a/examples/server_simple.py
+++ b/examples/server_simple.py
@@ -13,11 +13,11 @@ async def wshandle(request: web.Request) -> web.StreamResponse:
     await ws.prepare(request)
 
     async for msg in ws:
-        if msg.type == web.WSMsgType.TEXT:
+        if msg.type is web.WSMsgType.TEXT:
             await ws.send_str(f"Hello, {msg.data}")
-        elif msg.type == web.WSMsgType.BINARY:
+        elif msg.type is web.WSMsgType.BINARY:
             await ws.send_bytes(msg.data)
-        elif msg.type == web.WSMsgType.CLOSE:
+        elif msg.type is web.WSMsgType.CLOSE:
             break
 
     return ws
diff --git a/examples/web_ws.py b/examples/web_ws.py
index ec817d501b9..a5626ac90c9 100755
--- a/examples/web_ws.py
+++ b/examples/web_ws.py
@@ -31,11 +31,11 @@ async def wshandler(request: web.Request) -> Union[web.WebSocketResponse, web.Re
             await ws.send_str("Someone joined")
         request.app[sockets].append(resp)
 
-        async for msg in resp:  # type: ignore[misc]
-            if msg.type == web.WSMsgType.TEXT:  # type: ignore[misc]
+        async for msg in resp:
+            if msg.type is web.WSMsgType.TEXT:
                 for ws in request.app[sockets]:
                     if ws is not resp:
-                        await ws.send_str(msg.data)  # type: ignore[misc]
+                        await ws.send_str(msg.data)
             else:
                 return resp
         return resp
diff --git a/tests/autobahn/client/client.py b/tests/autobahn/client/client.py
index dfca77d12b2..5dbcd6ff654 100644
--- a/tests/autobahn/client/client.py
+++ b/tests/autobahn/client/client.py
@@ -8,7 +8,9 @@
 async def client(url: str, name: str) -> None:
     async with aiohttp.ClientSession() as session:
         async with session.ws_connect(url + "/getCaseCount") as ws:
-            num_tests = int((await ws.receive()).data)
+            msg = await ws.receive()
+            assert msg.type is aiohttp.WSMsgType.TEXT
+            num_tests = int(msg.data)
             print("running %d cases" % num_tests)
 
         for i in range(1, num_tests + 1):
@@ -16,9 +18,9 @@ async def client(url: str, name: str) -> None:
             text_url = url + "/runCase?case=%d&agent=%s" % (i, name)
             async with session.ws_connect(text_url) as ws:
                 async for msg in ws:
-                    if msg.type == aiohttp.WSMsgType.TEXT:
+                    if msg.type is aiohttp.WSMsgType.TEXT:
                         await ws.send_str(msg.data)
-                    elif msg.type == aiohttp.WSMsgType.BINARY:
+                    elif msg.type is aiohttp.WSMsgType.BINARY:
                         await ws.send_bytes(msg.data)
                     else:
                         break
diff --git a/tests/autobahn/server/server.py b/tests/autobahn/server/server.py
index 580d4d29343..d2801e5ef3d 100644
--- a/tests/autobahn/server/server.py
+++ b/tests/autobahn/server/server.py
@@ -21,11 +21,11 @@ async def wshandler(request: web.Request) -> web.WebSocketResponse:
     while True:
         msg = await ws.receive()
 
-        if msg.type == web.WSMsgType.TEXT:
+        if msg.type is web.WSMsgType.TEXT:
             await ws.send_str(msg.data)
-        elif msg.type == web.WSMsgType.BINARY:
+        elif msg.type is web.WSMsgType.BINARY:
             await ws.send_bytes(msg.data)
-        elif msg.type == web.WSMsgType.CLOSE:
+        elif msg.type is web.WSMsgType.CLOSE:
             await ws.close()
             break
         else:
diff --git a/tests/test_client_ws.py b/tests/test_client_ws.py
index 18f6222dbda..a3815eac272 100644
--- a/tests/test_client_ws.py
+++ b/tests/test_client_ws.py
@@ -16,6 +16,7 @@
     hdrs,
 )
 from aiohttp.http import WS_KEY
+from aiohttp.http_websocket import WSMessageClose
 from aiohttp.streams import EofStream
 from aiohttp.test_utils import make_mocked_coro
 
@@ -385,9 +386,7 @@ async def test_close(
                 resp = await session.ws_connect("http://test.org")
                 assert not resp.closed
 
-                resp._reader.feed_data(
-                    aiohttp.WSMessage(aiohttp.WSMsgType.CLOSE, b"", "")
-                )
+                resp._reader.feed_data(WSMessageClose(data=0, extra=""))
 
                 res = await resp.close()
                 writer.close.assert_called_with(1000, b"")
diff --git a/tests/test_client_ws_functional.py b/tests/test_client_ws_functional.py
index bc70bbd2124..fb381b1f643 100644
--- a/tests/test_client_ws_functional.py
+++ b/tests/test_client_ws_functional.py
@@ -136,8 +136,9 @@ async def handler(request: web.Request) -> web.WebSocketResponse:
         ws = web.WebSocketResponse()
         await ws.prepare(request)
 
-        data = await ws.receive()
-        await ws.send_frame(data.data, data.type)
+        msg = await ws.receive()
+        assert msg.type is WSMsgType.BINARY
+        await ws.send_frame(msg.data, msg.type)
         await ws.close()
         return ws
 
@@ -840,7 +841,6 @@ async def handler(request: web.Request) -> NoReturn:
         msg = await resp.receive(5.0)
         assert ping_received
         assert resp.close_code is WSCloseCode.ABNORMAL_CLOSURE
-        assert msg
         assert msg.type is WSMsgType.ERROR
         assert isinstance(msg.data, ServerTimeoutError)
 
@@ -990,6 +990,7 @@ async def handler(request: web.Request) -> web.WebSocketResponse:
         ws = web.WebSocketResponse()
         await ws.prepare(request)
         msg = await ws.receive()
+        assert msg.type is WSMsgType.TEXT
         await ws.send_str(msg.data + "/answer")
         await ws.close()
         return ws
@@ -1015,6 +1016,7 @@ async def handler(request: web.Request) -> web.WebSocketResponse:
         ws = web.WebSocketResponse()
         await ws.prepare(request)
         msg = await ws.receive()
+        assert msg.type is WSMsgType.TEXT
         await ws.send_str(msg.data + "/answer")
         await ws.close()
         return ws
@@ -1038,6 +1040,7 @@ async def handler(request: web.Request) -> web.WebSocketResponse:
         ws = web.WebSocketResponse()
         await ws.prepare(request)
         msg = await ws.receive()
+        assert msg.type is WSMsgType.TEXT
         await ws.send_str(msg.data + "/answer")
         await ws.close()
         return ws
diff --git a/tests/test_web_websocket.py b/tests/test_web_websocket.py
index 76656b2ab7a..6bdaee5e76b 100644
--- a/tests/test_web_websocket.py
+++ b/tests/test_web_websocket.py
@@ -9,7 +9,8 @@
 from pytest_mock import MockerFixture
 
 from aiohttp import WSMsgType, web
-from aiohttp.http import WS_CLOSED_MESSAGE, WSMessage
+from aiohttp.http import WSMessageType
+from aiohttp.http_websocket import WSMessageClose, WSMessageClosed, WSMessageClosing
 from aiohttp.streams import EofStream
 from aiohttp.test_utils import make_mocked_coro, make_mocked_request
 from aiohttp.web_ws import WebSocketReady
@@ -241,7 +242,7 @@ async def test_send_str_closed(make_request: _RequestMaker) -> None:
     ws = web.WebSocketResponse()
     await ws.prepare(req)
     assert ws._reader is not None
-    ws._reader.feed_data(WS_CLOSED_MESSAGE)
+    ws._reader.feed_data(WSMessageClosed())
     await ws.close()
     assert req.transport is not None
     assert len(req.transport.close.mock_calls) == 1  # type: ignore[attr-defined]
@@ -255,7 +256,7 @@ async def test_send_bytes_closed(make_request: _RequestMaker) -> None:
     ws = web.WebSocketResponse()
     await ws.prepare(req)
     assert ws._reader is not None
-    ws._reader.feed_data(WS_CLOSED_MESSAGE)
+    ws._reader.feed_data(WSMessageClosed())
     await ws.close()
 
     with pytest.raises(ConnectionError):
@@ -267,7 +268,7 @@ async def test_send_json_closed(make_request: _RequestMaker) -> None:
     ws = web.WebSocketResponse()
     await ws.prepare(req)
     assert ws._reader is not None
-    ws._reader.feed_data(WS_CLOSED_MESSAGE)
+    ws._reader.feed_data(WSMessageClosed())
     await ws.close()
 
     with pytest.raises(ConnectionError):
@@ -279,7 +280,7 @@ async def test_send_frame_closed(make_request: _RequestMaker) -> None:
     ws = web.WebSocketResponse()
     await ws.prepare(req)
     assert ws._reader is not None
-    ws._reader.feed_data(WS_CLOSED_MESSAGE)
+    ws._reader.feed_data(WSMessageClosed())
     await ws.close()
 
     with pytest.raises(ConnectionError):
@@ -291,7 +292,7 @@ async def test_ping_closed(make_request: _RequestMaker) -> None:
     ws = web.WebSocketResponse()
     await ws.prepare(req)
     assert ws._reader is not None
-    ws._reader.feed_data(WS_CLOSED_MESSAGE)
+    ws._reader.feed_data(WSMessageClosed())
     await ws.close()
 
     with pytest.raises(ConnectionError):
@@ -303,7 +304,7 @@ async def test_pong_closed(make_request: _RequestMaker, mocker: MockerFixture) -
     ws = web.WebSocketResponse()
     await ws.prepare(req)
     assert ws._reader is not None
-    ws._reader.feed_data(WS_CLOSED_MESSAGE)
+    ws._reader.feed_data(WSMessageClosed())
     await ws.close()
 
     with pytest.raises(ConnectionError):
@@ -315,7 +316,7 @@ async def test_close_idempotent(make_request: _RequestMaker) -> None:
     ws = web.WebSocketResponse()
     await ws.prepare(req)
     assert ws._reader is not None
-    ws._reader.feed_data(WS_CLOSED_MESSAGE)
+    ws._reader.feed_data(WSMessageClosed())
     close_code = await ws.close(code=1, message=b"message1")
     assert close_code == 1
     assert ws.closed
@@ -360,7 +361,7 @@ async def test_write_eof_idempotent(make_request: _RequestMaker) -> None:
     assert len(req.transport.close.mock_calls) == 0  # type: ignore[attr-defined]
 
     assert ws._reader is not None
-    ws._reader.feed_data(WS_CLOSED_MESSAGE)
+    ws._reader.feed_data(WSMessageClosed())
     await ws.close()
 
     await ws.write_eof()
@@ -420,7 +421,7 @@ async def test_receive_close_but_left_open(
     req = make_request("GET", "/")
     ws = web.WebSocketResponse()
     await ws.prepare(req)
-    close_message = WSMessage(WSMsgType.CLOSE, 1000, "close")
+    close_message = WSMessageClose(data=1000, extra="close")
 
     ws._reader = mock.Mock()
     ws._reader.read = mock.AsyncMock(return_value=close_message)
@@ -442,7 +443,7 @@ async def test_receive_closing(
     req = make_request("GET", "/")
     ws = web.WebSocketResponse()
     await ws.prepare(req)
-    closing_message = WSMessage(WSMsgType.CLOSING, 1000, "closing")
+    closing_message = WSMessageClosing()
 
     ws._reader = mock.Mock()
     read_mock = mock.AsyncMock(return_value=closing_message)
@@ -472,7 +473,7 @@ async def test_close_after_closing(
     req = make_request("GET", "/")
     ws = web.WebSocketResponse()
     await ws.prepare(req)
-    closing_message = WSMessage(WSMsgType.CLOSING, 1000, "closing")
+    closing_message = WSMessageClosing()
 
     ws._reader = mock.Mock()
     ws._reader.read = mock.AsyncMock(return_value=closing_message)
@@ -520,7 +521,7 @@ async def test_multiple_receive_on_close_connection(
     ws = web.WebSocketResponse()
     await ws.prepare(req)
     assert ws._reader is not None
-    ws._reader.feed_data(WS_CLOSED_MESSAGE)
+    ws._reader.feed_data(WSMessageClosed())
     await ws.close()
 
     await ws.receive()
diff --git a/tests/test_web_websocket_functional.py b/tests/test_web_websocket_functional.py
index 2f20cfa8ac1..6b04dc7c110 100644
--- a/tests/test_web_websocket_functional.py
+++ b/tests/test_web_websocket_functional.py
@@ -41,6 +41,7 @@ async def handler(request: web.Request) -> web.WebSocketResponse:
         await ws.prepare(request)
         msg = await ws.receive()
 
+        assert msg.type is WSMsgType.TEXT
         msg_json = msg.json()
         answer = msg_json["test"]
         await ws.send_str(answer)
@@ -234,6 +235,7 @@ async def handler(request: web.Request) -> web.WebSocketResponse:
 
     await ws.send_str('{"request": "test"}')
     msg = await ws.receive()
+    assert msg.type is WSMsgType.TEXT
     data = msg.json()
     assert msg.type == aiohttp.WSMsgType.TEXT
     assert data["response"] == "test"
@@ -1173,6 +1175,7 @@ async def websocket_handler(request: web.Request) -> web.WebSocketResponse:
 
         try:
             async for message in websocket:
+                assert message.type is WSMsgType.TEXT
                 await websocket.send_json({"ok": True, "message": message.json()})
         finally:
             request.app[websockets].discard(websocket)
diff --git a/tests/test_websocket_parser.py b/tests/test_websocket_parser.py
index d508537b716..b30968729d5 100644
--- a/tests/test_websocket_parser.py
+++ b/tests/test_websocket_parser.py
@@ -9,7 +9,7 @@
 
 import aiohttp
 from aiohttp import http_websocket
-from aiohttp.http import WebSocketError, WSCloseCode, WSMessage, WSMsgType
+from aiohttp.http import WebSocketError, WSCloseCode, WSMessageType, WSMsgType
 from aiohttp.http_websocket import (
     _WS_DEFLATE_TRAILING,
     PACK_CLOSE_CODE,
@@ -17,6 +17,11 @@
     PACK_LEN2,
     PACK_LEN3,
     WebSocketReader,
+    WSMessageBinary,
+    WSMessageClose,
+    WSMessagePing,
+    WSMessagePong,
+    WSMessageText,
     _websocket_mask,
 )
 
@@ -85,12 +90,12 @@ def build_close_frame(
 
 
 @pytest.fixture()
-def out(loop: asyncio.AbstractEventLoop) -> aiohttp.DataQueue[WSMessage]:
+def out(loop: asyncio.AbstractEventLoop) -> aiohttp.DataQueue[WSMessageType]:
     return aiohttp.DataQueue(loop)
 
 
 @pytest.fixture()
-def parser(out: aiohttp.DataQueue[WSMessage]) -> WebSocketReader:
+def parser(out: aiohttp.DataQueue[WSMessageType]) -> WebSocketReader:
     return WebSocketReader(out, 4 * 1024 * 1024)
 
 
@@ -136,14 +141,14 @@ def test_parse_frame_mask(parser: WebSocketReader) -> None:
 
 
 def test_parse_frame_header_reversed_bits(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     with pytest.raises(WebSocketError):
         parser.parse_frame(struct.pack("!BB", 0b01100000, 0b00000000))
 
 
 def test_parse_frame_header_control_frame(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     with pytest.raises(WebSocketError):
         parser.parse_frame(struct.pack("!BB", 0b00001000, 0b00000000))
@@ -151,61 +156,65 @@ def test_parse_frame_header_control_frame(
 
 @pytest.mark.xfail()
 def test_parse_frame_header_new_data_err(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     with pytest.raises(WebSocketError):
         parser.parse_frame(struct.pack("!BB", 0b000000000, 0b00000000))
 
 
 def test_parse_frame_header_payload_size(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     with pytest.raises(WebSocketError):
         parser.parse_frame(struct.pack("!BB", 0b10001000, 0b01111110))
 
 
-def test_ping_frame(out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader) -> None:
+def test_ping_frame(
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
+) -> None:
     with mock.patch.object(parser, "parse_frame", autospec=True) as m:
         m.return_value = [(1, WSMsgType.PING, b"data", False)]
 
         parser.feed_data(b"")
         res = out._buffer[0]
-        assert res == (WSMsgType.PING, b"data", "")
+        assert res == WSMessagePing(data=b"data", extra="")
 
 
-def test_pong_frame(out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader) -> None:
+def test_pong_frame(
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
+) -> None:
     with mock.patch.object(parser, "parse_frame", autospec=True) as m:
         m.return_value = [(1, WSMsgType.PONG, b"data", False)]
 
         parser.feed_data(b"")
         res = out._buffer[0]
-        assert res == (WSMsgType.PONG, b"data", "")
+        assert res == WSMessagePong(data=b"data", extra="")
 
 
 def test_close_frame(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     with mock.patch.object(parser, "parse_frame", autospec=True) as m:
         m.return_value = [(1, WSMsgType.CLOSE, b"", False)]
 
         parser.feed_data(b"")
         res = out._buffer[0]
-        assert res == (WSMsgType.CLOSE, 0, "")
+        assert res == WSMessageClose(data=0, extra="")
 
 
 def test_close_frame_info(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     with mock.patch.object(parser, "parse_frame", autospec=True) as m:
         m.return_value = [(1, WSMsgType.CLOSE, b"0112345", False)]
 
         parser.feed_data(b"")
         res = out._buffer[0]
-        assert res == WSMessage(WSMsgType.CLOSE, 12337, "12345")
+        assert res == WSMessageClose(data=12337, extra="12345")
 
 
 def test_close_frame_invalid(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     with mock.patch.object(parser, "parse_frame", autospec=True) as m:
         m.return_value = [(1, WSMsgType.CLOSE, b"1", False)]
@@ -217,7 +226,7 @@ def test_close_frame_invalid(
 
 
 def test_close_frame_invalid_2(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     data = build_close_frame(code=1)
 
@@ -237,7 +246,7 @@ def test_close_frame_unicode_err(parser: WebSocketReader) -> None:
 
 
 def test_unknown_frame(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     with mock.patch.object(parser, "parse_frame", autospec=True) as m:
         m.return_value = [(1, WSMsgType.CONTINUATION, b"", False)]
@@ -247,12 +256,12 @@ def test_unknown_frame(
 
 
 def test_simple_text(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     data = build_frame(b"text", WSMsgType.TEXT)
     parser._feed_data(data)
     res = out._buffer[0]
-    assert res == (WSMsgType.TEXT, "text", "")
+    assert res == WSMessageText(data="text", extra="")
 
 
 def test_simple_text_unicode_err(parser: WebSocketReader) -> None:
@@ -265,29 +274,29 @@ def test_simple_text_unicode_err(parser: WebSocketReader) -> None:
 
 
 def test_simple_binary(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     with mock.patch.object(parser, "parse_frame", autospec=True) as m:
         m.return_value = [(1, WSMsgType.BINARY, b"binary", False)]
 
         parser.feed_data(b"")
         res = out._buffer[0]
-        assert res == (WSMsgType.BINARY, b"binary", "")
+        assert res == WSMessageBinary(data=b"binary", extra="")
 
 
 def test_fragmentation_header(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     data = build_frame(b"a", WSMsgType.TEXT)
     parser._feed_data(data[:1])
     parser._feed_data(data[1:])
 
     res = out._buffer[0]
-    assert res == (WSMessage(WSMsgType.TEXT, "a", ""))
+    assert res == WSMessageText(data="a", extra="")
 
 
 def test_continuation(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     data1 = build_frame(b"line1", WSMsgType.TEXT, is_fin=False)
     parser._feed_data(data1)
@@ -296,11 +305,11 @@ def test_continuation(
     parser._feed_data(data2)
 
     res = out._buffer[0]
-    assert res == (WSMessage(WSMsgType.TEXT, "line1line2", ""))
+    assert res == WSMessageText(data="line1line2", extra="")
 
 
 def test_continuation_with_ping(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     with mock.patch.object(parser, "parse_frame", autospec=True) as m:
         m.return_value = [
@@ -319,13 +328,13 @@ def test_continuation_with_ping(
         parser._feed_data(data3)
 
         res = out._buffer[0]
-        assert res == WSMessage(WSMsgType.PING, b"", "")
+        assert res == WSMessagePing(data=b"", extra="")
         res = out._buffer[1]
-        assert res == WSMessage(WSMsgType.TEXT, "line1line2", "")
+        assert res == WSMessageText(data="line1line2", extra="")
 
 
 def test_continuation_err(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     with mock.patch.object(parser, "parse_frame", autospec=True) as m:
         m.return_value = [
@@ -338,7 +347,7 @@ def test_continuation_err(
 
 
 def test_continuation_with_close(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     with mock.patch.object(parser, "parse_frame", autospec=True) as m:
         m.return_value = [
@@ -354,13 +363,13 @@ def test_continuation_with_close(
 
         parser.feed_data(b"")
         res = out._buffer[0]
-        assert res, WSMessage(WSMsgType.CLOSE, 1002, "test")
+        assert res == WSMessageClose(data=1002, extra="test")
         res = out._buffer[1]
-        assert res == WSMessage(WSMsgType.TEXT, "line1line2", "")
+        assert res == WSMessageText(data="line1line2", extra="")
 
 
 def test_continuation_with_close_unicode_err(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     with mock.patch.object(parser, "parse_frame", autospec=True) as m:
         m.return_value = [
@@ -381,7 +390,7 @@ def test_continuation_with_close_unicode_err(
 
 
 def test_continuation_with_close_bad_code(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     with mock.patch.object(parser, "parse_frame", autospec=True) as m:
         m.return_value = [
@@ -397,7 +406,7 @@ def test_continuation_with_close_bad_code(
 
 
 def test_continuation_with_close_bad_payload(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     with mock.patch.object(parser, "parse_frame", autospec=True) as m:
         m.return_value = [
@@ -413,7 +422,7 @@ def test_continuation_with_close_bad_payload(
 
 
 def test_continuation_with_close_empty(
-    out: aiohttp.DataQueue[WSMessage], parser: WebSocketReader
+    out: aiohttp.DataQueue[WSMessageType], parser: WebSocketReader
 ) -> None:
     with mock.patch.object(parser, "parse_frame", autospec=True) as m:
         m.return_value = [
@@ -424,9 +433,9 @@ def test_continuation_with_close_empty(
 
         parser.feed_data(b"")
         res = out._buffer[0]
-        assert res, WSMessage(WSMsgType.CLOSE, 0, "")
+        assert res == WSMessageClose(data=0, extra="")
         res = out._buffer[1]
-        assert res == WSMessage(WSMsgType.TEXT, "line1line2", "")
+        assert res == WSMessageText(data="line1line2", extra="")
 
 
 websocket_mask_data: bytes = b"some very long data for masking by websocket"
@@ -505,7 +514,7 @@ def test_parse_compress_error_frame(parser: WebSocketReader) -> None:
     assert ctx.value.code == WSCloseCode.PROTOCOL_ERROR
 
 
-def test_parse_no_compress_frame_single(out: aiohttp.DataQueue[WSMessage]) -> None:
+def test_parse_no_compress_frame_single(out: aiohttp.DataQueue[WSMessageType]) -> None:
     parser_no_compress = WebSocketReader(out, 0, compress=False)
     with pytest.raises(WebSocketError) as ctx:
         parser_no_compress.parse_frame(struct.pack("!BB", 0b11000001, 0b00000001))
@@ -514,7 +523,7 @@ def test_parse_no_compress_frame_single(out: aiohttp.DataQueue[WSMessage]) -> No
     assert ctx.value.code == WSCloseCode.PROTOCOL_ERROR
 
 
-def test_msg_too_large(out: aiohttp.DataQueue[WSMessage]) -> None:
+def test_msg_too_large(out: aiohttp.DataQueue[WSMessageType]) -> None:
     parser = WebSocketReader(out, 256, compress=False)
     data = build_frame(b"text" * 256, WSMsgType.TEXT)
     with pytest.raises(WebSocketError) as ctx:
@@ -522,7 +531,7 @@ def test_msg_too_large(out: aiohttp.DataQueue[WSMessage]) -> None:
     assert ctx.value.code == WSCloseCode.MESSAGE_TOO_BIG
 
 
-def test_msg_too_large_not_fin(out: aiohttp.DataQueue[WSMessage]) -> None:
+def test_msg_too_large_not_fin(out: aiohttp.DataQueue[WSMessageType]) -> None:
     parser = WebSocketReader(out, 256, compress=False)
     data = build_frame(b"text" * 256, WSMsgType.TEXT, is_fin=False)
     with pytest.raises(WebSocketError) as ctx:
@@ -530,7 +539,7 @@ def test_msg_too_large_not_fin(out: aiohttp.DataQueue[WSMessage]) -> None:
     assert ctx.value.code == WSCloseCode.MESSAGE_TOO_BIG
 
 
-def test_compressed_msg_too_large(out: aiohttp.DataQueue[WSMessage]) -> None:
+def test_compressed_msg_too_large(out: aiohttp.DataQueue[WSMessageType]) -> None:
     parser = WebSocketReader(out, 256, compress=True)
     data = build_frame(b"aaa" * 256, WSMsgType.TEXT, compress=True)
     with pytest.raises(WebSocketError) as ctx:
diff --git a/tests/test_websocket_writer.py b/tests/test_websocket_writer.py
index 57745d237d8..ae6ef588fc3 100644
--- a/tests/test_websocket_writer.py
+++ b/tests/test_websocket_writer.py
@@ -5,7 +5,7 @@
 
 import pytest
 
-from aiohttp import DataQueue, WSMessage, WSMsgType
+from aiohttp import DataQueue, WSMessageType, WSMsgType
 from aiohttp.base_protocol import BaseProtocol
 from aiohttp.http import WebSocketReader, WebSocketWriter
 from aiohttp.test_utils import make_mocked_coro
@@ -153,7 +153,7 @@ async def test_concurrent_messages(
         "aiohttp.http_websocket.WEBSOCKET_MAX_SYNC_CHUNK_SIZE", max_sync_chunk_size
     ):
         writer = WebSocketWriter(protocol, transport, compress=15)
-        queue: DataQueue[WSMessage] = DataQueue(asyncio.get_running_loop())
+        queue: DataQueue[WSMessageType] = DataQueue(asyncio.get_running_loop())
         reader = WebSocketReader(queue, 50000)
         writers = []
         payloads = []
@@ -169,7 +169,8 @@ async def test_concurrent_messages(
         result, _ = reader.feed_data(call_bytes)
         assert result is False
         msg = await queue.read()
-        bytes_data: bytes = msg.data
+        assert msg.type is WSMsgType.BINARY
+        bytes_data = msg.data
         first_char = bytes_data[0:1]
         char_val = ord(first_char)
         assert len(bytes_data) == char_val


Code review comments:
diff_hunk: @@ -9,7 +9,8 @@
 from pytest_mock import MockerFixture
 
 from aiohttp import WSMsgType, web
-from aiohttp.http import WS_CLOSED_MESSAGE, WSMessage
+from aiohttp.http import WSMessageType
path: tests/test_web_websocket.py
text: @user1:
## Unused import

Import of 'WSMessageType' is not used.

[Show more details](https://github.com/aio-libs/aiohttp/security/code-scanning/1014)