EVOLUTION-MANAGER

Edit File: test_receivebuffer.py

import re

import pytest

from .._receivebuffer import ReceiveBuffer

def test_receivebuffer():
    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):
    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"