venv

1 files changed, 135 insertions, 0 deletions

diff --git a/venv/lib/python3.11/site-packages/h11/tests/test_receivebuffer.py b/venv/lib/python3.11/site-packages/h11/tests/test_receivebuffer.py
new file mode 100644
index 0000000..21a3870
--- /dev/null
+++ b/venv/lib/python3.11/site-packages/h11/tests/test_receivebuffer.py
@@ -0,0 +1,135 @@
+import re
+from typing import Tuple
+
+import pytest
+
+from .._receivebuffer import ReceiveBuffer
+
+
+def test_receivebuffer() -> None:
+    b = ReceiveBuffer()
+    assert not b
+    assert len(b) == 0
+    assert bytes(b) == b""
+
+    b += b"123"
+    assert b
+    assert len(b) == 3
+    assert bytes(b) == b"123"
+
+    assert bytes(b) == b"123"
+
+    assert b.maybe_extract_at_most(2) == b"12"
+    assert b
+    assert len(b) == 1
+    assert bytes(b) == b"3"
+
+    assert bytes(b) == b"3"
+
+    assert b.maybe_extract_at_most(10) == b"3"
+    assert bytes(b) == b""
+
+    assert b.maybe_extract_at_most(10) is None
+    assert not b
+
+    ################################################################
+    # maybe_extract_until_next
+    ################################################################
+
+    b += b"123\n456\r\n789\r\n"
+
+    assert b.maybe_extract_next_line() == b"123\n456\r\n"
+    assert bytes(b) == b"789\r\n"
+
+    assert b.maybe_extract_next_line() == b"789\r\n"
+    assert bytes(b) == b""
+
+    b += b"12\r"
+    assert b.maybe_extract_next_line() is None
+    assert bytes(b) == b"12\r"
+
+    b += b"345\n\r"
+    assert b.maybe_extract_next_line() is None
+    assert bytes(b) == b"12\r345\n\r"
+
+    # here we stopped at the middle of b"\r\n" delimiter
+
+    b += b"\n6789aaa123\r\n"
+    assert b.maybe_extract_next_line() == b"12\r345\n\r\n"
+    assert b.maybe_extract_next_line() == b"6789aaa123\r\n"
+    assert b.maybe_extract_next_line() is None
+    assert bytes(b) == b""
+
+    ################################################################
+    # maybe_extract_lines
+    ################################################################
+
+    b += b"123\r\na: b\r\nfoo:bar\r\n\r\ntrailing"
+    lines = b.maybe_extract_lines()
+    assert lines == [b"123", b"a: b", b"foo:bar"]
+    assert bytes(b) == b"trailing"
+
+    assert b.maybe_extract_lines() is None
+
+    b += b"\r\n\r"
+    assert b.maybe_extract_lines() is None
+
+    assert b.maybe_extract_at_most(100) == b"trailing\r\n\r"
+    assert not b
+
+    # Empty body case (as happens at the end of chunked encoding if there are
+    # no trailing headers, e.g.)
+    b += b"\r\ntrailing"
+    assert b.maybe_extract_lines() == []
+    assert bytes(b) == b"trailing"
+
+
+@pytest.mark.parametrize(
+    "data",
+    [
+        pytest.param(
+            (
+                b"HTTP/1.1 200 OK\r\n",
+                b"Content-type: text/plain\r\n",
+                b"Connection: close\r\n",
+                b"\r\n",
+                b"Some body",
+            ),
+            id="with_crlf_delimiter",
+        ),
+        pytest.param(
+            (
+                b"HTTP/1.1 200 OK\n",
+                b"Content-type: text/plain\n",
+                b"Connection: close\n",
+                b"\n",
+                b"Some body",
+            ),
+            id="with_lf_only_delimiter",
+        ),
+        pytest.param(
+            (
+                b"HTTP/1.1 200 OK\n",
+                b"Content-type: text/plain\r\n",
+                b"Connection: close\n",
+                b"\n",
+                b"Some body",
+            ),
+            id="with_mixed_crlf_and_lf",
+        ),
+    ],
+)
+def test_receivebuffer_for_invalid_delimiter(data: Tuple[bytes]) -> None:
+    b = ReceiveBuffer()
+
+    for line in data:
+        b += line
+
+    lines = b.maybe_extract_lines()
+
+    assert lines == [
+        b"HTTP/1.1 200 OK",
+        b"Content-type: text/plain",
+        b"Connection: close",
+    ]
+    assert bytes(b) == b"Some body"