Accept any sequence of header pairs, not just list

CHANGELOG.rst

@@ -4,6 +4,8 @@ Release History
 1.4.0 (unreleased)
 ------------------
 
+- Accept any sequence of ``(name, value)`` header pairs, matching ``h11``.
+
 - Require ``event_hint`` when constructing ``RemoteProtocolError``.
   This is an API-breaking change.
 

src/wsproto/handshake.py

@@ -81,15 +81,17 @@ def initiate_upgrade_connection(
         This should be used if the request has already be received and
         parsed.
 
-        :param list headers: HTTP headers represented as a list of 2-tuples.
+        :param headers: HTTP headers as a sequence of ``(name, value)`` pairs.
         :param str path: A URL path.
         """
         if self.client:
             msg = "Cannot initiate an upgrade connection when acting as the client"
             raise LocalProtocolError(
                 msg,
             )
-        upgrade_request = h11.Request(method=b"GET", target=path, headers=headers)
+        upgrade_request = h11.Request(
+            method=b"GET", target=path, headers=list(headers),
+        )
         h11_client = h11.Connection(h11.CLIENT)
         self.receive_data(h11_client.send(upgrade_request))
 
@@ -204,7 +206,7 @@ def _process_connection_request(
         subprotocols: list[str] = []
         upgrade = b""
         version = None
-        headers: Headers = []
+        collected: list[tuple[bytes, bytes]] = []
         for name, value in event.headers:
             name = name.lower()
             if name == b"connection":
@@ -224,7 +226,8 @@ def _process_connection_request(
                 version = value
             elif name == b"upgrade":
                 upgrade = value
-            headers.append((name, value))
+            collected.append((name, value))
+        headers: Headers = collected
         if connection_tokens is None or not any(
             token.lower() == "upgrade" for token in connection_tokens
         ):
@@ -299,7 +302,7 @@ def _accept(self, event: AcceptConnection) -> bytes:
 
         response = h11.InformationalResponse(
             status_code=101,
-            headers=headers + event.extra_headers,
+            headers=headers + list(event.extra_headers),
             reason=b"Switching Protocols",
         )
         self._connection = Connection(
@@ -381,7 +384,7 @@ def _initiate_connection(self, request: Request) -> bytes:
         upgrade = h11.Request(
             method=b"GET",
             target=request.target.encode("ascii"),
-            headers=headers + request.extra_headers,
+            headers=headers + list(request.extra_headers),
         )
         return self._h11_connection.send(upgrade) or b""
 
@@ -397,7 +400,7 @@ def _establish_client_connection(
         accepts: list[str] = []
         subprotocol = None
         upgrade = b""
-        headers: Headers = []
+        collected: list[tuple[bytes, bytes]] = []
         for name, value in event.headers:
             name = name.lower()
             if name == b"connection":
@@ -415,7 +418,8 @@ def _establish_client_connection(
             if name == b"upgrade":
                 upgrade = value
                 continue  # Skip appending to headers
-            headers.append((name, value))
+            collected.append((name, value))
+        headers: Headers = collected
 
         if connection_tokens is None or not any(
             token.lower() == "upgrade" for token in connection_tokens

src/wsproto/typing.py

@@ -1,3 +1,3 @@
-from __future__ import annotations
+from collections.abc import Sequence
 
-Headers = list[tuple[bytes, bytes]]
+Headers = Sequence[tuple[bytes, bytes]]

tests/test_client.py

@@ -61,6 +61,21 @@ def test_connection_request_additional_headers() -> None:
     assert headers[b"x-bar"] == b"Foo"
 
 
+def test_connection_request_tuple_extra_headers() -> None:
+    # Headers accepts any sequence of pairs, not only list.
+    request = _make_connection_request(
+        Request(
+            host="localhost",
+            target="/",
+            extra_headers=((b"X-Foo", b"Bar"), (b"X-Bar", b"Foo")),
+        ),
+    )
+
+    headers = normed_header_dict(request.headers)
+    assert headers[b"x-foo"] == b"Bar"
+    assert headers[b"x-bar"] == b"Foo"
+
+
 def test_connection_request_simple_extension() -> None:
     extension = FakeExtension(offer_response=True)
     request = _make_connection_request(

tests/test_server.py

@@ -131,9 +131,9 @@ def test_connection_request_key_header() -> None:
     assert str(excinfo.value) == "Missing header, 'Sec-WebSocket-Key'"
 
 
-def test_upgrade_request() -> None:
-    server = WSConnection(SERVER)
-    server.initiate_upgrade_connection(
+@pytest.mark.parametrize(
+    "headers",
+    [
         [
             (b"Host", b"localhost"),
             (b"Connection", b"Keep-Alive, Upgrade"),
@@ -142,8 +142,21 @@ def test_upgrade_request() -> None:
             (b"Sec-WebSocket-Key", generate_nonce()),
             (b"X-Foo", b"bar"),
         ],
-        "/",
-    )
+        # Sequence that is not a list (the point of Headers = Sequence[...]).
+        (
+            (b"Host", b"localhost"),
+            (b"Connection", b"Keep-Alive, Upgrade"),
+            (b"Upgrade", b"websocket"),
+            (b"Sec-WebSocket-Version", b"13"),
+            (b"Sec-WebSocket-Key", generate_nonce()),
+            (b"X-Foo", b"bar"),
+        ),
+    ],
+    ids=["list", "tuple"],
+)
+def test_upgrade_request(headers: Headers) -> None:
+    server = WSConnection(SERVER)
+    server.initiate_upgrade_connection(headers, "/")
     event = next(server.events())
     event = cast("Request", event)
 
@@ -222,6 +235,18 @@ def test_handshake_extra_headers() -> None:
     ]
 
 
+def test_handshake_tuple_extra_headers() -> None:
+    response, nonce = _make_handshake([], accept_headers=((b"X-Foo", b"bar"),))
+
+    assert response.status_code == 101
+    assert sorted(response.headers) == [
+        (b"connection", b"Upgrade"),
+        (b"sec-websocket-accept", generate_accept_token(nonce)),
+        (b"upgrade", b"websocket"),
+        (b"x-foo", b"bar"),
+    ]
+
+
 @pytest.mark.parametrize("accept_subprotocol", ["one", "two"])
 def test_handshake_with_subprotocol(accept_subprotocol: str) -> None:
     response, _ = _make_handshake(