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

Code before:
    assert r.json['headers']['Vary'] == 'XXX,YYY'
    assert r.json['headers']['Foo'] == 'bar'
    assert r.json['headers']['Bar'] == 'baz'
    assert r.json['headers']['Baz'] == 'quux'

    r = http('GET', httpbin_both + '/headers', 'Foo:bar', 'Vary:XXX',
             'Foo:baz', 'Vary:YYY', 'Foo:quux')
    assert r.json['headers']['Vary'] == 'XXX,YYY'
    assert r.json['headers']['Foo'] == 'bar,baz,quux'

    r = http('GET', httpbin_both + '/headers', 'Foo:bar', 'Foo:baz',
             'Foo:')
    assert 'Foo' not in r.json['headers']

    r = http('GET', httpbin_both + '/headers', 'Foo:bar', 'Foo:baz',
             'Foo:', 'Foo:quux')
    assert r.json['headers']['Foo'] == 'quux'


def test_headers_preserve_prepared_headers(httpbin_both):
    r = http('POST', httpbin_both + '/post', '--raw', 'foo')
    assert r.json['headers']['Content-Length'] == '3'

    r = http('POST', httpbin_both + '/post', 'Content-Length:0',
             '--raw', 'foo')
    assert r.json['headers']['Content-Length'] == '3'


def test_json_input_preserve_order(httpbin_both):
    r = http('PATCH', httpbin_both + '/patch',
             'order:={"map":{"1":"first","2":"second"}}')
    assert HTTP_OK in r
    assert r.json['data'] == \
        '{"order": {"map": {"1": "first", "2": "second"}}}'


Review comment (suggestion):
Since `httpbin` converts the repeated headers lines in a single comma-separated header, we should put a comment about this peculiarity here in the tests. 


Code after:
             'Foo:')
    assert 'Foo' not in r.json['headers']


def test_headers_multiple_values(httpbin_both):
    r = http('GET', httpbin_both + '/headers', 'Foo:bar', 'Foo:baz')
    assert r.json['headers']['Foo'] == 'bar,baz'


def test_headers_multiple_values_repeated(httpbin_both):
    r = http('GET', httpbin_both + '/headers', 'Foo:bar', 'Foo:baz',
             'Foo:bar')
    assert r.json['headers']['Foo'] == 'bar,baz,bar'


@pytest.mark.parametrize("headers, expected", [
    (
        ["Foo;", "Foo:bar"],
        ",bar"
    ),
    (
        ["Foo:bar", "Foo;"],
        "bar,"
    ),
    (
        ["Foo:bar", "Foo;", "Foo:baz"],
        "bar,,baz"
    ),
])
def test_headers_multiple_values_with_empty(httpbin_both, headers, expected):
    r = http('GET', httpbin_both + '/headers', *headers)
    assert r.json['headers']['Foo'] == expected


def test_headers_multiple_values_mixed(httpbin_both):
    r = http('GET', httpbin_both + '/headers', 'Foo:bar', 'Vary:XXX',
             'Foo:baz', 'Vary:YYY', 'Foo:quux')
    assert r.json['headers']['Vary'] == 'XXX,YYY'
    assert r.json['headers']['Foo'] == 'bar,baz,quux'


def test_headers_preserve_prepared_headers(httpbin_both):
    r = http('POST', httpbin_both + '/post', 'Content-Length:0',
             '--raw', 'foo')
    assert r.json['headers']['Content-Length'] == '3'


@pytest.mark.parametrize('pretty', ['format', 'none'])
def test_headers_multiple_headers_representation(httpbin_both, pretty):
    r = http('--offline', '--pretty', pretty, 'example.org',
             'A:A', 'A:B', 'A:C', 'B:A', 'B:B', 'C:C', 'c:c')
Repository: httpie/httpie
File: tests/utils/http_server.py (Python)

Code before:

class TestHandler(BaseHTTPRequestHandler):
    handlers = defaultdict(dict)

    @classmethod
    def handler(cls, method, path):
        def inner(func):
            cls.handlers[method][path] = func
            return func
        return inner

    def do_GET(self):
        parse_result = urlparse(self.path)
        func = self.handlers['GET'].get(parse_result.path)
        if func is None:
            return self.send_error(HTTPStatus.NOT_FOUND)

        return func(self)


@TestHandler.handler('GET', '/headers')
def get_headers(handler):
    handler.send_response(200)
    for key, value in handler.headers.items():
        handler.send_header(key, value)
    handler.send_header("Content-Length", 0)
    handler.end_headers()


@pytest.fixture(scope="function")
def http_server():
    """A custom HTTP server implementation for our tests, that is
    built on top of the http.server module. Handy when we need to
    deal with details which httpbin can not capture."""

    server = HTTPServer(('localhost', 0), TestHandler)
    thread = threading.Thread(target=server.serve_forever)
    thread.start()
    yield "{}:{}".format(*server.socket.getsockname())
    server.shutdown()
    thread.join(timeout=0.5)


Review comment (suggestion):
```suggestion
    handler.send_header('Content-Length', 0)
```

Code after:

class TestHandler(BaseHTTPRequestHandler):
    handlers = defaultdict(dict)

    @classmethod
    def handler(cls, method, path):
        def inner(func):
            cls.handlers[method][path] = func
            return func
        return inner

    def do_GET(self):
        parse_result = urlparse(self.path)
        func = self.handlers['GET'].get(parse_result.path)
        if func is None:
            return self.send_error(HTTPStatus.NOT_FOUND)

        return func(self)


@TestHandler.handler('GET', '/headers')
def get_headers(handler):
    handler.send_response(200)
    for key, value in handler.headers.items():
        handler.send_header(key, value)
    handler.send_header('Content-Length', 0)
    handler.end_headers()


@pytest.fixture(scope="function")
def http_server():
    """A custom HTTP server implementation for our tests, that is
    built on top of the http.server module. Handy when we need to
    deal with details which httpbin can not capture."""

    server = HTTPServer(('localhost', 0), TestHandler)
    thread = threading.Thread(target=server.serve_forever)
    thread.start()
    yield '{}:{}'.format(*server.socket.getsockname())
    server.shutdown()
    thread.join(timeout=0.5)
Repository: httpie/httpie
File: tests/utils/http_server.py (Python)

Code before:
        func = self.handlers['GET'].get(parse_result.path)
        if func is None:
            return self.send_error(HTTPStatus.NOT_FOUND)

        return func(self)


@TestHandler.handler('GET', '/headers')
def get_headers(handler):
    handler.send_response(200)
    for key, value in handler.headers.items():
        handler.send_header(key, value)
    handler.send_header("Content-Length", 0)
    handler.end_headers()


@pytest.fixture(scope="function")
def http_server():
    """A custom HTTP server implementation for our tests, that is
    built on top of the http.server module. Handy when we need to
    deal with details which httpbin can not capture."""

    server = HTTPServer(('localhost', 0), TestHandler)
    thread = threading.Thread(target=server.serve_forever)
    thread.start()
    yield "{}:{}".format(*server.socket.getsockname())
    server.shutdown()
    thread.join(timeout=0.5)


Review comment (suggestion):
```suggestion
    yield '{}:{}'.format(*server.socket.getsockname())
```

Code after:
        func = self.handlers['GET'].get(parse_result.path)
        if func is None:
            return self.send_error(HTTPStatus.NOT_FOUND)

        return func(self)


@TestHandler.handler('GET', '/headers')
def get_headers(handler):
    handler.send_response(200)
    for key, value in handler.headers.items():
        handler.send_header(key, value)
    handler.send_header('Content-Length', 0)
    handler.end_headers()


@pytest.fixture(scope="function")
def http_server():
    """A custom HTTP server implementation for our tests, that is
    built on top of the http.server module. Handy when we need to
    deal with details which httpbin can not capture."""

    server = HTTPServer(('localhost', 0), TestHandler)
    thread = threading.Thread(target=server.serve_forever)
    thread.start()
    yield '{}:{}'.format(*server.socket.getsockname())
    server.shutdown()
    thread.join(timeout=0.5)
Repository: httpie/httpie
File: httpie/adapters.py (Python)

Code before:
from httpie.cli.dicts import HTTPHeadersDict
from requests.adapters import HTTPAdapter


class HTTPieHTTPAdapter(HTTPAdapter):

    def build_response(self, req, resp):
        """Wrap the original headers with the `HTTPHeadersDict`
        to preserve multiple headers that have the same name"""

        response = super().build_response(req, resp)
        response.headers = HTTPHeadersDict(getattr(resp, "headers", {}))
        return response


Review comment (suggestion):
```suggestion
        response.headers = HTTPHeadersDict(getattr(resp, 'headers', {}))
```

Code after:
from httpie.cli.dicts import HTTPHeadersDict
from requests.adapters import HTTPAdapter


class HTTPieHTTPAdapter(HTTPAdapter):

    def build_response(self, req, resp):
        """Wrap the original headers with the `HTTPHeadersDict`
        to preserve multiple headers that have the same name"""

        response = super().build_response(req, resp)
        response.headers = HTTPHeadersDict(getattr(resp, 'headers', {}))
        return response
Repository: httpie/httpie
File: httpie/output/formatters/xml.py (Python)

Code before:
import sys
from typing import TYPE_CHECKING, Optional

from ...encoding import UTF8
from ...plugins import FormatterPlugin

if TYPE_CHECKING:
    from xml.dom.minidom import Document


XML_OPEN_TAG = '<?xml'
XML_CLOSE_TAG = '?>'


def parse_xml(data: str) -> 'Document':
    """Parse given XML `data` string into an appropriate :class:`~xml.dom.minidom.Document` object."""
    from defusedxml.minidom import parseString
    return parseString(data)


def parse_declaration(raw_body: str) -> Optional[str]:
    body = raw_body.strip()
    # XMLDecl ::= '<?xml' DECL_CONTENT '?>'
    if body.startswith(XML_OPEN_TAG):
        end = body.find(XML_CLOSE_TAG)
        if end != -1:
            return body[:end + len(XML_CLOSE_TAG)]


def pretty_xml(document: 'Document',
               declaration: Optional[str] = None,
               encoding: Optional[str] = UTF8,
               indent: int = 2,
               standalone: Optional[bool] = None) -> str:
    """Render the given :class:`~xml.dom.minidom.Document` `document` into a prettified string."""
    kwargs = {
        'encoding': encoding or UTF8,

Review comment (suggestion):
Pedantic: Would be nice to have `DECL` in the name, and not have `TAG` in there as it can be confused with elements tags. 


Code after:
from typing import TYPE_CHECKING, Optional

from ...encoding import UTF8
from ...plugins import FormatterPlugin

if TYPE_CHECKING:
    from xml.dom.minidom import Document


XML_DECLARATION_OPEN = '<?xml'
XML_DECLARATION_CLOSE = '?>'


def parse_xml(data: str) -> 'Document':
    """Parse given XML `data` string into an appropriate :class:`~xml.dom.minidom.Document` object."""
    from defusedxml.minidom import parseString
    return parseString(data)


def parse_declaration(raw_body: str) -> Optional[str]:
    body = raw_body.strip()
    # XMLDecl ::= '<?xml' DECL_CONTENT '?>'
    if body.startswith(XML_DECLARATION_OPEN):
        end = body.find(XML_DECLARATION_CLOSE)
        if end != -1:
            return body[:end + len(XML_DECLARATION_CLOSE)]


def pretty_xml(document: 'Document',
               declaration: Optional[str] = None,
               encoding: Optional[str] = UTF8,
               indent: int = 2) -> str:
    """Render the given :class:`~xml.dom.minidom.Document` `document` into a prettified string."""
    kwargs = {
        'encoding': encoding or UTF8,
        'indent': ' ' * indent,
    }
Repository: httpie/httpie
File: httpie/output/formatters/colors.py (Python)

Code before:
import pygments.token
from pygments.formatters.terminal import TerminalFormatter
from pygments.formatters.terminal256 import Terminal256Formatter
from pygments.lexer import Lexer
from pygments.lexers.data import JsonLexer
from pygments.lexers.special import TextLexer
from pygments.lexers.text import HttpLexer as PygmentsHttpLexer
from pygments.util import ClassNotFound

from ..lexers.json import EnhancedJsonLexer
from ..ui.palette import COLOR_PALETTE
from ...compat import is_windows
from ...context import Environment
from ...plugins import FormatterPlugin


AUTO_STYLE = 'auto'  # Follows terminal ANSI color styles
DEFAULT_STYLE = AUTO_STYLE

# Bundled here
SOLARIZED_STYLE = 'solarized'
PIE_STYLES = {
    'pie',
    'pie-light',
    'pie-dark'
}

if is_windows:
    # Colors on Windows via colorama don't look that
    # great and fruity seems to give the best result there.
    DEFAULT_STYLE = 'fruity'

BUNDLED_STYLES = {
    SOLARIZED_STYLE,
    AUTO_STYLE,
    *PIE_STYLES,
}


def get_available_styles():
    return BUNDLED_STYLES | set(pygments.styles.get_all_styles())


class ColorFormatter(FormatterPlugin):
    """
    Colorize using Pygments

    This processor that applies syntax highlighting to the headers,
    and also to the body if its content type is recognized.

    """

Review comment (suggestion):
Looks like an enum/constants candidate (we refer to it from multiple places). Maybe it could also go to `palette`.

Code after:
import pygments.token
from pygments.formatters.terminal import TerminalFormatter
from pygments.formatters.terminal256 import Terminal256Formatter
from pygments.lexer import Lexer
from pygments.lexers.data import JsonLexer
from pygments.lexers.special import TextLexer
from pygments.lexers.text import HttpLexer as PygmentsHttpLexer
from pygments.util import ClassNotFound

from ..lexers.json import EnhancedJsonLexer
from ..ui.palette import SHADE_NAMES, get_color
from ...compat import is_windows
from ...context import Environment
from ...plugins import FormatterPlugin


AUTO_STYLE = 'auto'  # Follows terminal ANSI color styles
DEFAULT_STYLE = AUTO_STYLE
SOLARIZED_STYLE = 'solarized'  # Bundled here

if is_windows:
    # Colors on Windows via colorama don't look that
    # great and fruity seems to give the best result there.
    DEFAULT_STYLE = 'fruity'

BUNDLED_STYLES = {
    SOLARIZED_STYLE,
    AUTO_STYLE
}


def get_available_styles():
    return BUNDLED_STYLES | set(pygments.styles.get_all_styles())


class ColorFormatter(FormatterPlugin):
    """
    Colorize using Pygments

    This processor that applies syntax highlighting to the headers,
    and also to the body if its content type is recognized.

    """
    group_name = 'colors'

    def __init__(
        self,
        env: Environment,
        explicit_json=False,
        color_scheme=DEFAULT_STYLE,
        **kwargs
Repository: httpie/httpie
File: httpie/cli/argtypes.py (Python)

Code before:

        There are only two token types - strings and escaped characters:

        >>> KeyValueArgType('=').tokenize(r'foo\=bar\\baz')
        ['foo', Escaped('='), 'bar', Escaped('\\'), 'baz']

        """
        tokens = ['']
        characters = iter(s)
        for char in characters:
            if char == '\\':
                char = next(characters, '')
                if char not in self.special_characters:
                    tokens[-1] += '\\' + char
                else:
                    tokens.extend([Escaped(char), ''])
            else:
                tokens[-1] += char
        return tokens


class AuthCredentials(KeyValueArg):
    """Represents parsed credentials."""

    def has_password(self) -> bool:
        return self.value

    def prompt_password(self, host: str):
        prompt_text = f'http: password for {self.key}@{host}: '
        try:
            self.value = self._getpass(prompt_text)
        except (EOFError, KeyboardInterrupt):
            sys.stderr.write('\n')
            sys.exit(0)

    @staticmethod
    def _getpass(prompt):
        # To allow easy mocking.
        return getpass.getpass(str(prompt))


class AuthCredentialsArgType(KeyValueArgType):
    """A key-value arg type that parses credentials."""

    key_value_class = AuthCredentials

    def __call__(self, s):
        """Parse credentials from `s`.

        ("username" or "username:password").


Review comment (suggestion):
I don't understand this change, why are we returning the password in `has_password`. `value` can be either a string (which is not a bool, check the function's signature) or `None`.

Code after:

        There are only two token types - strings and escaped characters:

        >>> KeyValueArgType('=').tokenize(r'foo\=bar\\baz')
        ['foo', Escaped('='), 'bar', Escaped('\\'), 'baz']

        """
        tokens = ['']
        characters = iter(s)
        for char in characters:
            if char == '\\':
                char = next(characters, '')
                if char not in self.special_characters:
                    tokens[-1] += '\\' + char
                else:
                    tokens.extend([Escaped(char), ''])
            else:
                tokens[-1] += char
        return tokens


class AuthCredentials(KeyValueArg):
    """Represents parsed credentials."""

    def has_password(self) -> bool:
        return self.value is not None

    def prompt_password(self, host: str):
        prompt_text = f'http: password for {self.key}@{host}: '
        try:
            self.value = self._getpass(prompt_text)
        except (EOFError, KeyboardInterrupt):
            sys.stderr.write('\n')
            sys.exit(0)

    @staticmethod
    def _getpass(prompt):
        # To allow easy mocking.
        return getpass.getpass(str(prompt))


class AuthCredentialsArgType(KeyValueArgType):
    """A key-value arg type that parses credentials."""

    key_value_class = AuthCredentials

    def __call__(self, s):
        """Parse credentials from `s`.

        ("username" or "username:password").
Repository: httpie/httpie
File: httpie/plugins/builtin.py (Python)

Code before:

    @staticmethod
    def make_header(username: str, password: str) -> str:
        credentials = f'{username}:{password}'
        token = b64encode(credentials.encode()).strip().decode('latin1')
        return f'Basic {token}'


class HTTPBearerAuth(requests.auth.AuthBase):

    def __init__(self, token: str) -> None:
        self.token = token

    def __call__(self, request: requests.PreparedRequest) -> requests.PreparedRequest:
        request.headers['Authorization'] = f'Bearer {self.token}'
        return request


class BasicAuthPlugin(BuiltinAuthPlugin):
    name = 'Basic HTTP auth'
    auth_type = 'basic'
    netrc_parse = True

    # noinspection PyMethodOverriding
    def get_auth(self, username: str, password: str) -> HTTPBasicAuth:
        self.raw_auth = username + ":" + password
        return HTTPBasicAuth(username, password)


class DigestAuthPlugin(BuiltinAuthPlugin):
    name = 'Digest HTTP auth'
    auth_type = 'digest'
    netrc_parse = True

    # noinspection PyMethodOverriding
    def get_auth(
        self,
        username: str,
        password: str
    ) -> requests.auth.HTTPDigestAuth:
        return requests.auth.HTTPDigestAuth(username, password)


class BearerAuthPlugin(BuiltinAuthPlugin):
    name = 'Bearer HTTP Auth'
    auth_type = 'bearer'
    netrc_parse = False
    auth_parse = False

    # noinspection PyMethodOverriding
    def get_auth(self, **kwargs) -> requests.auth.HTTPDigestAuth:

Review comment (security):
This seems very localized to `BasicAuth`, what about others? I don't think the right solution would be fixing this here, but rather on a higher level where we ask the password.

Code after:

    @staticmethod
    def make_header(username: str, password: str) -> str:
        credentials = f'{username}:{password}'
        token = b64encode(credentials.encode()).strip().decode('latin1')
        return f'Basic {token}'


class HTTPBearerAuth(requests.auth.AuthBase):

    def __init__(self, token: str) -> None:
        self.token = token

    def __call__(self, request: requests.PreparedRequest) -> requests.PreparedRequest:
        request.headers['Authorization'] = f'Bearer {self.token}'
        return request


class BasicAuthPlugin(BuiltinAuthPlugin):
    name = 'Basic HTTP auth'
    auth_type = 'basic'
    netrc_parse = True

    # noinspection PyMethodOverriding
    def get_auth(self, username: str, password: str) -> HTTPBasicAuth:
        return HTTPBasicAuth(username, password)


class DigestAuthPlugin(BuiltinAuthPlugin):
    name = 'Digest HTTP auth'
    auth_type = 'digest'
    netrc_parse = True

    # noinspection PyMethodOverriding
    def get_auth(
        self,
        username: str,
        password: str
    ) -> requests.auth.HTTPDigestAuth:
        return requests.auth.HTTPDigestAuth(username, password)


class BearerAuthPlugin(BuiltinAuthPlugin):
    name = 'Bearer HTTP Auth'
    auth_type = 'bearer'
    netrc_parse = False
    auth_parse = False

    # noinspection PyMethodOverriding
    def get_auth(self, **kwargs) -> requests.auth.HTTPDigestAuth:
        return HTTPBearerAuth(self.raw_auth)
Repository: httpie/httpie
File: tests/utils/http_server.py (Python)

Code before:
@TestHandler.handler('GET', '/drip')
def chunked_drip(handler):
    handler.send_response(200)
    accept = handler.headers.get('Accept')
    if accept is not None:
        handler.send_header('Content-Type', accept)
    handler.send_header('Transfer-Encoding', 'chunked')
    handler.end_headers()

    for _ in range(3):
        body = 'test\n'
        handler.wfile.write(f'{len(body):X}\r\n{body}\r\n'.encode('utf-8'))

    handler.wfile.write('0\r\n\r\n'.encode('utf-8'))


@TestHandler.handler('GET', '/html')
def basic_html_page(handler):
    content = b"\x00<some></some>"

    handler.send_response(200)
    handler.send_header('Content-Type', 'text/html; charset=utf-8')
    handler.send_header('Content-Length', len(content))
    handler.end_headers()

    handler.wfile.write(content)


@pytest.fixture(scope="function")
def http_server():
    """A custom HTTP server implementation for our tests, that is
    built on top of the http.server module. Handy when we need to
    deal with details which httpbin can not capture."""

    server = HTTPServer(('localhost', 0), TestHandler)
    thread = threading.Thread(target=server.serve_forever)
    thread.start()
    yield '{}:{}'.format(*server.socket.getsockname())
    server.shutdown()
    thread.join(timeout=0.5)


Review comment (performance):
You could just use the `httpbin` fixture (`httpbin + '/html'`).

```
$ http pie.dev/html
HTTP/1.1 200 OK
CF-Cache-Status: DYNAMIC
CF-RAY: 6be0bf830dce2780-PRG
Connection: keep-alive
Content-Encoding: gzip
Content-Type: text/html; charset=utf-8
```

Code after:
@TestHandler.handler('GET', '/drip')
def chunked_drip(handler):
    handler.send_response(200)
    accept = handler.headers.get('Accept')
    if accept is not None:
        handler.send_header('Content-Type', accept)
    handler.send_header('Transfer-Encoding', 'chunked')
    handler.end_headers()

    for _ in range(3):
        body = 'test\n'
        handler.wfile.write(f'{len(body):X}\r\n{body}\r\n'.encode('utf-8'))

    handler.wfile.write('0\r\n\r\n'.encode('utf-8'))


@pytest.fixture(scope="function")
def http_server():
    """A custom HTTP server implementation for our tests, that is
    built on top of the http.server module. Handy when we need to
    deal with details which httpbin can not capture."""

    server = HTTPServer(('localhost', 0), TestHandler)
    thread = threading.Thread(target=server.serve_forever)
    thread.start()
    yield '{}:{}'.format(*server.socket.getsockname())
    server.shutdown()
    thread.join(timeout=0.5)