<commit_before>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"

    b.compress()
    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"

    b.compress()
    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"12345a6789aa"

    assert b.maybe_extract_until_next(re.compile(b"a"), 1) == b"12345a"
    assert bytes(b) == b"6789aa"

    assert b.maybe_extract_until_next(re.compile(b"aaa"), 3) is None
    assert bytes(b) == b"6789aa"

    b += b"a12"
    assert b.maybe_extract_until_next(re.compile(b"aaa"), 3) == b"6789aaa"
    assert bytes(b) == b"12"

    # check repeated searches for the same needle, triggering the
    # pickup-where-we-left-off logic
    b += b"345"
    assert b.maybe_extract_until_next(re.compile(b"aaa"), 3) is None

    b += b"6789aaa123"
    assert b.maybe_extract_until_next(re.compile(b"aaa"), 3) == b"123456789aaa"
    assert bytes(b) == b"123"

    ################################################################
    # maybe_extract_lines
    ################################################################

    b += b"\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",
    [
        (
            b"HTTP/1.1 200 OK\r\n",
            b"Content-type: text/plain\r\n",
            b"\r\n",
            b"Some body",
        ),
        (b"HTTP/1.1 200 OK\n", b"Content-type: text/plain\n", b"\n", b"Some body"),
        (b"HTTP/1.1 200 OK\r\n", b"Content-type: text/plain\n", b"\n", b"Some body"),
    ],
)
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"]
    assert bytes(b) == b"Some body"
<commit_msg>Small rfg for tests - added pytest param names to test_receivebuffer_for_invalid_delimiter<commit_after>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"

    b.compress()
    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"

    b.compress()
    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"12345a6789aa"

    assert b.maybe_extract_until_next(re.compile(b"a"), 1) == b"12345a"
    assert bytes(b) == b"6789aa"

    assert b.maybe_extract_until_next(re.compile(b"aaa"), 3) is None
    assert bytes(b) == b"6789aa"

    b += b"a12"
    assert b.maybe_extract_until_next(re.compile(b"aaa"), 3) == b"6789aaa"
    assert bytes(b) == b"12"

    # check repeated searches for the same needle, triggering the
    # pickup-where-we-left-off logic
    b += b"345"
    assert b.maybe_extract_until_next(re.compile(b"aaa"), 3) is None

    b += b"6789aaa123"
    assert b.maybe_extract_until_next(re.compile(b"aaa"), 3) == b"123456789aaa"
    assert bytes(b) == b"123"

    ################################################################
    # maybe_extract_lines
    ################################################################

    b += b"\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"\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"\n", b"Some body"),
            id="with_lf_only_delimiter",
        ),
        pytest.param(
            (
                b"HTTP/1.1 200 OK\r\n",
                b"Content-type: text/plain\n",
                b"\n",
                b"Some body",
            ),
            id="with_double_lf_before_body",
        ),
        pytest.param(
            (
                b"HTTP/1.1 200 OK\r\n",
                b"Content-type: text/plain\r\n",
                b"\n",
                b"Some body",
            ),
            id="with_mixed_crlf",
        ),
    ],
)
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"]
    assert bytes(b) == b"Some body"

{'fraction_non_alphanumeric': 0.2132433704691668, 'fraction_numerical': 0.033736074062450966, 'mean_word_length': 2.547022815804118, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4602671', 'n_tokens_mistral': 2400, 'n_tokens_neox': 2129, 'n_words': 469}