Gecko [ tsuru-no-tsurugi.tokyo ]

Name	Size	Permission	Date
__pycache__	[ DIR ]	drwxr-xr-x	2024-08-30 00:22
data	[ DIR ]	drwxr-xr-x	2024-08-30 00:22
__init__.py	0 B	-rw-r--r--	2022-01-20 05:32
helpers.py	3.28 KB	-rw-r--r--	2022-01-20 05:32
test_against_stdlib_http.py	3.9 KB	-rw-r--r--	2022-01-20 05:32
test_connection.py	37.81 KB	-rw-r--r--	2022-01-20 05:32
test_events.py	4.55 KB	-rw-r--r--	2022-01-20 05:32
test_headers.py	5.48 KB	-rw-r--r--	2022-01-20 05:32
test_helpers.py	794 B	-rw-r--r--	2022-01-20 05:32
test_io.py	15.86 KB	-rw-r--r--	2022-01-20 05:32
test_receivebuffer.py	3.37 KB	-rw-r--r--	2022-01-20 05:32
test_state.py	8.72 KB	-rw-r--r--	2022-01-20 05:32
test_util.py	2.9 KB	-rw-r--r--	2022-01-20 05:32

Rename

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"