ronantakizawa/codereview-bench huggingface dataset:
Repository: httpie/httpie
File: httpie/models.py (Python)

Code before:

    def iter_lines(self, chunk_size):
        return ((line, b'\n') for line in self._orig.iter_lines(chunk_size))

    # noinspection PyProtectedMember
    @property
    def headers(self):
        try:
            raw_version = self._orig.raw._original_response.version
        except AttributeError:
            # Assume HTTP/1.1
            raw_version = 11
        version = {
            9: '0.9',
            10: '1.0',
            11: '1.1',
            20: '2',
        }[raw_version]

        original = self._orig
        status_line = f'HTTP/{version} {original.status_code} {original.reason}'
        headers = [status_line]
        headers.extend(
            ': '.join(header)
            for header in original.headers.items()
            if header[0] != "Set-Cookie"
        )
        headers.extend(
            f'Set-Cookie: {cookie}'
            for cookie in split_cookies(original.headers.get('Set-Cookie'))
        )
        return '\r\n'.join(headers)

    @property
    def encoding(self):
        return self._orig.encoding or 'utf8'

    @property
    def body(self):
        # Only now the response body is fetched.
        # Shouldn't be touched unless the body is actually needed.
        return self._orig.content


class HTTPRequest(HTTPMessage):
    """A :class:`requests.models.Request` wrapper."""

    def iter_body(self, chunk_size):
        yield self.body

    def iter_lines(self, chunk_size):

Review comment (suggestion):
```suggestion
            if header[0] != 'Set-Cookie'
```

Code after:

    def iter_lines(self, chunk_size):
        return ((line, b'\n') for line in self._orig.iter_lines(chunk_size))

    # noinspection PyProtectedMember
    @property
    def headers(self):
        try:
            raw_version = self._orig.raw._original_response.version
        except AttributeError:
            # Assume HTTP/1.1
            raw_version = 11
        version = {
            9: '0.9',
            10: '1.0',
            11: '1.1',
            20: '2',
        }[raw_version]

        original = self._orig
        status_line = f'HTTP/{version} {original.status_code} {original.reason}'
        headers = [status_line]
        headers.extend(
            ': '.join(header)
            for header in original.headers.items()
            if header[0] != 'Set-Cookie'
        )
        headers.extend(
            f'Set-Cookie: {cookie}'
            for cookie in split_cookies(original.headers.get('Set-Cookie'))
        )
        return '\r\n'.join(headers)

    @property
    def encoding(self):
        return self._orig.encoding or 'utf8'

    @property
    def body(self):
        # Only now the response body is fetched.
        # Shouldn't be touched unless the body is actually needed.
        return self._orig.content


class HTTPRequest(HTTPMessage):
    """A :class:`requests.models.Request` wrapper."""

    def iter_body(self, chunk_size):
        yield self.body

    def iter_lines(self, chunk_size):
Repository: httpie/httpie
File: tests/test_cookie.py (Python)

Code before:
from flask import Flask, make_response

from .utils import http

app = Flask(__name__)


@app.route('/')
def main():
    response = make_response()
    response.set_cookie('hello', value='world')
    response.set_cookie('oatmeal_raisin', value='is the best')
    return response


def available_port():
    conn = socket()
    conn.bind(('', 0))
    port = conn.getsockname()[1]
    conn.close()
    return port


def test_cookie_parser():
    port = available_port()
    server = Process(target=app.run, kwargs={'port': port})
    try:
        server.start()
        response = http(f'http://localhost:{port}/')
        assert 'Set-Cookie: hello=world; Path=/' in response
        assert 'Set-Cookie: oatmeal_raisin="is the best"; Path=/' in response
    finally:
        server.terminate()
        server.join()


Review comment (suggestion):
```suggestion
    server = Process(target=app.run, kwargs={'port': 8888})
```

Code after:

from .utils import http


class TestIntegration:

    def setup_mock_server(self, handler):
        """Configure mock server."""
        # Passing 0 as the port will cause a random free port to be chosen.
        self.mock_server = HTTPServer(('localhost', 0), handler)
        _, self.mock_server_port = self.mock_server.server_address

        # Start running mock server in a separate thread.
        # Daemon threads automatically shut down when the main process exits.
        self.mock_server_thread = Thread(target=self.mock_server.serve_forever)
        self.mock_server_thread.setDaemon(True)
        self.mock_server_thread.start()

    def test_cookie_parser(self):
        """Not directly testing HTTPie but `requests` to ensure their cookies handling
        is still as expected by `get_expired_cookies()`.
        """

        class MockServerRequestHandler(BaseHTTPRequestHandler):
            """"HTTP request handler."""

            def do_GET(self):
                """Handle GET requests."""
                # Craft multiple cookies
                cookie = SimpleCookie()
                cookie['hello'] = 'world'
                cookie['hello']['path'] = self.path
                cookie['oatmeal_raisin'] = 'is the best'
                cookie['oatmeal_raisin']['path'] = self.path

                # Send HTTP headers
                self.send_response(200)
                self.send_header('Set-Cookie', cookie.output())
                self.end_headers()

        self.setup_mock_server(MockServerRequestHandler)
        response = http(f'http://localhost:{self.mock_server_port}/')
        assert 'Set-Cookie: hello=world; Path=/' in response
        assert 'Set-Cookie: oatmeal_raisin="is the best"; Path=/' in response
Repository: httpie/httpie
File: tests/test_cookie.py (Python)

Code before:
from flask import Flask, make_response

from .utils import http

app = Flask(__name__)


@app.route('/')
def main():
    response = make_response()
    response.set_cookie('hello', value='world')
    response.set_cookie('oatmeal_raisin', value='is the best')
    return response


def available_port():
    conn = socket()
    conn.bind(('', 0))
    port = conn.getsockname()[1]
    conn.close()
    return port


def test_cookie_parser():
    port = available_port()
    server = Process(target=app.run, kwargs={'port': 8888})
    try:
        server.start()
        response = http(f'http://localhost:{port}/')
        assert 'Set-Cookie: hello=world; Path=/' in response
        assert 'Set-Cookie: oatmeal_raisin="is the best"; Path=/' in response
    finally:
        server.terminate()
        server.join()


Review comment (suggestion):
```suggestion
    port = 8080
    server = Process(target=app.run, kwargs={'port': port})
```

Code after:

from .utils import http


class TestIntegration:

    def setup_mock_server(self, handler):
        """Configure mock server."""
        # Passing 0 as the port will cause a random free port to be chosen.
        self.mock_server = HTTPServer(('localhost', 0), handler)
        _, self.mock_server_port = self.mock_server.server_address

        # Start running mock server in a separate thread.
        # Daemon threads automatically shut down when the main process exits.
        self.mock_server_thread = Thread(target=self.mock_server.serve_forever)
        self.mock_server_thread.setDaemon(True)
        self.mock_server_thread.start()

    def test_cookie_parser(self):
        """Not directly testing HTTPie but `requests` to ensure their cookies handling
        is still as expected by `get_expired_cookies()`.
        """

        class MockServerRequestHandler(BaseHTTPRequestHandler):
            """"HTTP request handler."""

            def do_GET(self):
                """Handle GET requests."""
                # Craft multiple cookies
                cookie = SimpleCookie()
                cookie['hello'] = 'world'
                cookie['hello']['path'] = self.path
                cookie['oatmeal_raisin'] = 'is the best'
                cookie['oatmeal_raisin']['path'] = self.path

                # Send HTTP headers
                self.send_response(200)
                self.send_header('Set-Cookie', cookie.output())
                self.end_headers()

        self.setup_mock_server(MockServerRequestHandler)
        response = http(f'http://localhost:{self.mock_server_port}/')
        assert 'Set-Cookie: hello=world; Path=/' in response
        assert 'Set-Cookie: oatmeal_raisin="is the best"; Path=/' in response
Repository: httpie/httpie
File: httpie/models.py (Python)

Code before:
from typing import Iterable, Optional
from urllib.parse import urlsplit

from httpie.utils import split_cookies


class HTTPMessage:
    """Abstract class for HTTP messages."""

    def __init__(self, orig):
        self._orig = orig

    def iter_body(self, chunk_size: int) -> Iterable[bytes]:
        """Return an iterator over the body."""
        raise NotImplementedError()

    def iter_lines(self, chunk_size: int) -> Iterable[bytes]:
        """Return an iterator over the body yielding (`line`, `line_feed`)."""
        raise NotImplementedError()

    @property
    def headers(self) -> str:
        """Return a `str` with the message's headers."""
        raise NotImplementedError()

    @property
    def encoding(self) -> Optional[str]:
        """Return a `str` with the message's encoding, if known."""
        raise NotImplementedError()

Review comment (suggestion):
```suggestion
from .utils import split_cookies
```

Code after:
from typing import Iterable, Optional
from urllib.parse import urlsplit

from .utils import split_cookies


class HTTPMessage:
    """Abstract class for HTTP messages."""

    def __init__(self, orig):
        self._orig = orig

    def iter_body(self, chunk_size: int) -> Iterable[bytes]:
        """Return an iterator over the body."""
        raise NotImplementedError()

    def iter_lines(self, chunk_size: int) -> Iterable[bytes]:
        """Return an iterator over the body yielding (`line`, `line_feed`)."""
        raise NotImplementedError()

    @property
    def headers(self) -> str:
        """Return a `str` with the message's headers."""
        raise NotImplementedError()

    @property
    def encoding(self) -> Optional[str]:
        """Return a `str` with the message's encoding, if known."""
        raise NotImplementedError()
Repository: httpie/httpie
File: httpie/utils.py (Python)

Code before:
    <https://github.com/psf/requests/issues/2773#issuecomment-174312831>
    """

    def __call__(self, r):
        return r


def get_content_type(filename):
    """
    Return the content type for ``filename`` in format appropriate
    for Content-Type headers, or ``None`` if the file type is unknown
    to ``mimetypes``.

    """
    mime, encoding = mimetypes.guess_type(filename, strict=False)
    if mime:
        content_type = mime
        if encoding:
            content_type = f'{mime}; charset={encoding}'
        return content_type


def split_cookies(cookies):
    """
    When Requests stores cookies in ``response.headers['Set-Cookie']``
    it concatenates all of them through ``, ``

    This function splits cookies apart being careful to not to
    split on ``, `` which may be part of cookie value.
    """
    if not cookies:
        return []
    return RE_COOKIE_SPLIT.split(cookies)


def get_expired_cookies(
    cookies: str,
    now: float = None
) -> List[dict]:

    now = now or time.time()

    def is_expired(expires: Optional[float]) -> bool:
        return expires is not None and expires <= now

    attr_sets: List[Tuple[str, str]] = parse_ns_headers(
        split_cookies(cookies)
    )

    cookies = [
        # The first attr name is the cookie name.

Review comment (suggestion):
```suggestion
    When ``requests`` stores cookies in ``response.headers['Set-Cookie']``
    it concatenates all of them through ``, ``.
```

Code after:
    <https://github.com/psf/requests/issues/2773#issuecomment-174312831>
    """

    def __call__(self, r):
        return r


def get_content_type(filename):
    """
    Return the content type for ``filename`` in format appropriate
    for Content-Type headers, or ``None`` if the file type is unknown
    to ``mimetypes``.

    """
    mime, encoding = mimetypes.guess_type(filename, strict=False)
    if mime:
        content_type = mime
        if encoding:
            content_type = f'{mime}; charset={encoding}'
        return content_type


def split_cookies(cookies):
    """
    When ``requests`` stores cookies in ``response.headers['Set-Cookie']``
    it concatenates all of them through ``, ``.

    This function splits cookies apart being careful to not to
    split on ``, `` which may be part of cookie value.
    """
    if not cookies:
        return []
    return RE_COOKIE_SPLIT.split(cookies)


def get_expired_cookies(
    cookies: str,
    now: float = None
) -> List[dict]:

    now = now or time.time()

    def is_expired(expires: Optional[float]) -> bool:
        return expires is not None and expires <= now

    attr_sets: List[Tuple[str, str]] = parse_ns_headers(
        split_cookies(cookies)
    )

    cookies = [
        # The first attr name is the cookie name.
Repository: httpie/httpie
File: httpie/output/formatters/json.py (Python)

Code before:
import json
from typing import Generator

from ...plugins import FormatterPlugin

XXSI_PREFIXES = (")]}',\n", ")]}',", 'while(1);', 'for(;;)')


class JSONFormatter(FormatterPlugin):

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.enabled = self.format_options['json']['format']

    def _find_potential_json_bodies(self, data: str) -> Generator[str, None, None]:
        """Yield possible JSON bodies derived from `data`.

        Sometimes the JSON body is prefixed with a XSSI token, specific to the server.
        In order to be able to display well-formatted JSON, that method will try
        several tricks to find the actual body of the JSON.

        Here are yielded bodies (if one is successfully parsed by :meth:`._load_json()`,
        then the process is stopped early):

            1. First, the full `data`.
            2. Then, strip off eventual :data::`XSSI_PREFIXES`.
            3. Then, try to find the start of the actual body by applying some heuristics.
            4. Finally, iterate over `data` one character at a time.
               Note: that path should be rarely taken, it likely means
                     `data` is a simple JSON string prefixed with a
                     custom XSSI prefix, or not JSON at all.

Review comment (security):
I added the constant to ease future prefix add.

For example, I [read](http://patorjk.com/blog/2013/02/05/crafty-tricks-for-avoiding-xssi/) that some service was using `123\n` as prefix. Such prefix would break the formatting as no heuristics could handle it: it is only valid JSON values. For instance, that response's body would be problematic without the possibility to add `123\n` prefix later:
```json
123\n"some string"
```

For more context, have a look at the doctring of `JSONFormatter._find_potential_json_bodies()`.

Code after:
import json

from ...plugins import FormatterPlugin


class JSONFormatter(FormatterPlugin):

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.enabled = self.format_options['json']['format']

    def format_body(self, body: str, mime: str) -> str:
        maybe_json = [
            'json',
            'javascript',
            'text',
        ]
        if (self.kwargs['explicit_json']
                or any(token in mime for token in maybe_json)):
            from ..utils import load_prefixed_json
            try:
                data_prefix, json_obj = load_prefixed_json(body)
            except ValueError:
                pass  # Invalid JSON, ignore.
            else:
                # Indent, sort keys by name, and avoid
                # unicode escapes to improve readability.
                body = data_prefix + json.dumps(
                    obj=json_obj,
                    sort_keys=self.format_options['json']['sort_keys'],
                    ensure_ascii=False,
Repository: httpie/httpie
File: httpie/output/formatters/json.py (Python)

Code before:
               Note: that path should be rarely taken, it likely means
                     `data` is a simple JSON string prefixed with a
                     custom XSSI prefix, or not JSON at all.

        """
        # First, the full data.
        yield data

        # Then, strip off eventual XSSI prefix.
        for prefix in XXSI_PREFIXES:
            if data.startswith(prefix):
                yield data[len(prefix)]

        # Then, try to find the start of the actual body.
        token_position = {data.find('{'), data.find('[')} - {-1}
        if token_position:
            yield data[min(token_position):]

        # Finally, iterate over the data one character at a time.
        # Note: that path should be rarely taken, it likely means
        #       the data is a simple JSON string prefixed with a
        #       custom XSSI prefix, or not JSON at all.
        data = data[1:]
        while data:
            yield data
            data = data[1:]

    def _load_json(self, data: str) -> json.JSONDecoder:
        """Simple JSON loading from bodies found in `data`.
        See :meth:`._find_potential_json_bodies()` for how bodies are detected.

        """
        for body in self._find_potential_json_bodies(data):
            try:
                return json.loads(body)
            except ValueError:
                pass
        raise ValueError('Invalid JSON')

    def format_body(self, body: str, mime: str) -> str:
        maybe_json = [
            'json',
            'javascript',
            'text',
        ]
        if (self.kwargs['explicit_json']
                or any(token in mime for token in maybe_json)):
            try:
                obj = self._load_json(body)
            except ValueError:
                pass  # Invalid JSON, ignore.

Review comment (security):
This may be insane to do that. It is present only to handle simple JSON strings prefixed with custom XSSI string.

Code after:
import json

from ...plugins import FormatterPlugin


class JSONFormatter(FormatterPlugin):

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.enabled = self.format_options['json']['format']

    def format_body(self, body: str, mime: str) -> str:
        maybe_json = [
            'json',
            'javascript',
            'text',
        ]
        if (self.kwargs['explicit_json']
                or any(token in mime for token in maybe_json)):
            from ..utils import load_prefixed_json
            try:
                data_prefix, json_obj = load_prefixed_json(body)
            except ValueError:
                pass  # Invalid JSON, ignore.
            else:
                # Indent, sort keys by name, and avoid
                # unicode escapes to improve readability.
                body = data_prefix + json.dumps(
                    obj=json_obj,
                    sort_keys=self.format_options['json']['sort_keys'],
                    ensure_ascii=False,
                    indent=self.format_options['json']['indent']
                )
        return body
Repository: httpie/httpie
File: httpie/output/formatters/json.py (Python)

Code before:
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.enabled = self.format_options['json']['format']

    def _find_potential_json_bodies(self, data: str) -> Generator[str, None, None]:
        """Yield possible JSON bodies derived from `data`.

        Sometimes the JSON body is prefixed with a XSSI token, specific to the server.
        In order to be able to display well-formatted JSON, that method will try
        several tricks to find the actual body of the JSON.

        Here are yielded bodies (if one is successfully parsed by :meth:`._load_json()`,
        then the process is stopped early):

            1. First, the full `data`.
            2. Then, strip off eventual :data::`XSSI_PREFIXES`.
            3. Finally, try to find the start of the actual body by applying some heuristics.

        """
        # First, the full data.
        yield data

        # Then, strip off eventual XSSI prefix.
        for prefix in XXSI_PREFIXES:
            if data.startswith(prefix):
                yield data[len(prefix)]

        # Finally, try to find the start of the actual body.
        token_position = {data.find('{'), data.find('[')} - {-1}
        if token_position:
            yield data[min(token_position):]

    def _load_json(self, data: str) -> json.JSONDecoder:
        """Simple JSON loading from bodies found in `data`.
        See :meth:`._find_potential_json_bodies()` for how bodies are detected.

        """
        for body in self._find_potential_json_bodies(data):
            try:
                return json.loads(body)
            except ValueError:
                pass
        raise ValueError('Invalid JSON')

    def format_body(self, body: str, mime: str) -> str:
        maybe_json = [
            'json',
            'javascript',
            'text',
        ]
        if (self.kwargs['explicit_json']

Review comment (suggestion):
```suggestion
                yield data[len(prefix):]
```

Code after:
import json

from ...plugins import FormatterPlugin


class JSONFormatter(FormatterPlugin):

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.enabled = self.format_options['json']['format']

    def format_body(self, body: str, mime: str) -> str:
        maybe_json = [
            'json',
            'javascript',
            'text',
        ]
        if (self.kwargs['explicit_json']
                or any(token in mime for token in maybe_json)):
            from ..utils import load_prefixed_json
            try:
                data_prefix, json_obj = load_prefixed_json(body)
            except ValueError:
                pass  # Invalid JSON, ignore.
            else:
                # Indent, sort keys by name, and avoid
                # unicode escapes to improve readability.
                body = data_prefix + json.dumps(
                    obj=json_obj,
                    sort_keys=self.format_options['json']['sort_keys'],
                    ensure_ascii=False,
                    indent=self.format_options['json']['indent']
                )
        return body