Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

Streamed terminal output

`--stream` can be used to enable streaming also with `--pretty` and to ensure
a more frequent output flushing.
  • Loading branch information...
commit c7657e3c4b51d86be9d62c2b9d05f5371a2b02fc 1 parent 4615011
@jkbrzt authored
View
3  .gitignore
@@ -4,3 +4,6 @@ build
*.pyc
.tox
README.html
+.coverage
+htmlcov
+
View
43 README.rst
@@ -79,7 +79,7 @@ There are five different types of key/value pair ``items`` available:
| | nested ``Object``, or an ``Array``. It's because |
| | simple data items are always serialized as a |
| | ``String``. E.g., ``pies:=[1,2,3]``, or |
-| | ``'meals:=["ham","spam"]'`` (note the quotes). |
+| | ``meals:='["ham","spam"]'`` (note the quotes). |
| | It may be more convenient to pass the whole JSON |
| | body via ``stdin`` when it's more complex |
| | (see examples bellow). |
@@ -221,18 +221,31 @@ respectively:
esac
fi
+**The output is always streamed** unless ``--pretty`` is set or implied. You
+can use ``--stream`` / ``-S`` to enable streaming even with ``--pretty``, in
+which case every line of the output will processed and flushed as soon as it's
+avaialbe (as opossed to buffering the whole response which wouldn't work for
+long-lived requests). You can test it with the Twitter streaming API:
+
+.. code-block:: shell
+
+ http -Sfa <your-twitter-username> https://stream.twitter.com/1/statuses/filter.json track='Justin Bieber'
+ # \/
+ # The short options for --stream, --form and --auth.
+
+``--stream`` can also be used regardless of ``--pretty`` to ensure a more
+frequent output flushing (sort of like ``tail -f``).
Flags
-----
``$ http --help``::
- usage: http [--help] [--version] [--json | --form] [--traceback]
- [--pretty | --ugly]
+ usage: http [--help] [--version] [--json | --form] [--pretty | --ugly]
[--print OUTPUT_OPTIONS | --verbose | --headers | --body]
- [--style STYLE] [--check-status] [--auth AUTH]
+ [--style STYLE] [--stream] [--check-status] [--auth AUTH]
[--auth-type {basic,digest}] [--verify VERIFY] [--proxy PROXY]
- [--allow-redirects] [--timeout TIMEOUT]
+ [--allow-redirects] [--timeout TIMEOUT] [--debug]
[METHOD] URL [ITEM [ITEM ...]]
HTTPie - cURL for humans. <http://httpie.org>
@@ -266,7 +279,6 @@ Flags
-www-form-urlencoded (if not specified). The presence
of any file fields results into a multipart/form-data
request.
- --traceback Print exception traceback should one occur.
--pretty If stdout is a terminal, the response is prettified by
default (colorized and indented if it is JSON). This
flag ensures prettifying even when stdout is
@@ -282,7 +294,7 @@ Flags
piped to another program or to a file, then only the
body is printed by default.
--verbose, -v Print the whole request as well as the response.
- Shortcut for --print=HBhb.
+ Shortcut for --print=HBbh.
--headers, -h Print only the response headers. Shortcut for
--print=h.
--body, -b Print only the response body. Shortcut for --print=b.
@@ -291,10 +303,19 @@ Flags
colorful, default, emacs, friendly, fruity, manni,
monokai, murphy, native, pastie, perldoc, rrt,
solarized, tango, trac, vim, vs. Defaults to
- solarized. For this option to work properly, please
+ "solarized". For this option to work properly, please
make sure that the $TERM environment variable is set
to "xterm-256color" or similar (e.g., via `export TERM
=xterm-256color' in your ~/.bashrc).
+ --stream, -S Always stream the output by line, i.e., behave like
+ `tail -f'. Without --stream and with --pretty (either
+ set or implied), HTTPie fetches the whole response
+ before it outputs the processed data. Set this option
+ when you want to continuously display a prettified
+ long-lived response, such as one from the Twitter
+ streaming API. It is useful also without --pretty: It
+ ensures that the output is flushed more often and in
+ smaller chunks.
--check-status By default, HTTPie exits with 0 when no network or
other fatal errors occur. This flag instructs HTTPie
to also check the HTTP status code and exit with an
@@ -321,6 +342,9 @@ Flags
POST-ing of data at new ``Location``)
--timeout TIMEOUT Float describes the timeout of the request (Use
socket.setdefaulttimeout() as fallback).
+ --debug Prints exception traceback should one occur and other
+ information useful for debugging HTTPie itself.
+
Contribute
@@ -373,6 +397,9 @@ Changelog
=========
* `0.2.7dev`_
+ * Streamed terminal output. ``--stream`` / ``-S`` can be used to enable
+ streaming also with ``--pretty`` and to ensure a more frequent output
+ flushing.
* Support for efficient large file downloads.
* Response body is fetched only when needed (e.g., not with ``--headers``).
* Improved content type matching.
View
3  httpie/__init__.py
@@ -5,6 +5,3 @@
__author__ = 'Jakub Roztocil'
__version__ = '0.2.7dev'
__licence__ = 'BSD'
-
-
-CONTENT_TYPE = 'Content-Type'
View
25 httpie/cli.py
@@ -9,7 +9,7 @@
from . import __doc__
from . import __version__
-from .output import AVAILABLE_STYLES
+from .output import AVAILABLE_STYLES, DEFAULT_STYLE
from .input import (Parser, AuthCredentialsArgType, KeyValueArgType,
PRETTIFY_STDOUT_TTY_ONLY,
SEP_PROXY, SEP_CREDENTIALS, SEP_GROUP_ITEMS,
@@ -56,7 +56,7 @@ def _(text):
parser.add_argument(
- '--output', '-o', type=argparse.FileType('wb'),
+ '--output', '-o', type=argparse.FileType('w+b'),
metavar='FILE',
help= argparse.SUPPRESS if not is_windows else _(
'''
@@ -131,16 +131,31 @@ def _(text):
)
parser.add_argument(
- '--style', '-s', dest='style', default='solarized', metavar='STYLE',
+ '--style', '-s', dest='style', default=DEFAULT_STYLE, metavar='STYLE',
choices=AVAILABLE_STYLES,
help=_('''
- Output coloring style, one of %s. Defaults to solarized.
+ Output coloring style, one of %s. Defaults to "%s".
For this option to work properly, please make sure that the
$TERM environment variable is set to "xterm-256color" or similar
(e.g., via `export TERM=xterm-256color' in your ~/.bashrc).
- ''') % ', '.join(sorted(AVAILABLE_STYLES))
+ ''') % (', '.join(sorted(AVAILABLE_STYLES)), DEFAULT_STYLE)
)
+parser.add_argument('--stream', '-S', action='store_true', default=False, help=_(
+ '''
+ Always stream the output by line, i.e., behave like `tail -f'.
+
+ Without --stream and with --pretty (either set or implied),
+ HTTPie fetches the whole response before it outputs the processed data.
+
+ Set this option when you want to continuously display a prettified
+ long-lived response, such as one from the Twitter streaming API.
+
+ It is useful also without --pretty: It ensures that the output is flushed
+ more often and in smaller chunks.
+
+ '''
+))
parser.add_argument(
'--check-status', default=False, action='store_true',
help=_('''
View
100 httpie/core.py
@@ -3,20 +3,27 @@
Invocation flow:
1. Read, validate and process the input (args, `stdin`).
- 2. Create a request and send it, get the response.
- 3. Process and format the requested parts of the request-response exchange.
- 4. Write to `stdout` and exit.
+ 2. Create and send a request.
+ 3. Stream, and possibly process and format, the requested parts
+ of the request-response exchange.
+ 4. Simultaneously write to `stdout`
+ 5. Exit.
"""
import sys
import json
+import errno
+from itertools import chain
+from functools import partial
import requests
import requests.auth
from requests.compat import str
from .models import HTTPRequest, HTTPResponse, Environment
-from .output import OutputProcessor, formatted_stream
+from .output import (OutputProcessor, RawStream, PrettyStream,
+ BufferedPrettyStream, EncodedStream)
+
from .input import (OUT_REQ_BODY, OUT_REQ_HEAD,
OUT_RESP_HEAD, OUT_RESP_BODY)
from .cli import parser
@@ -85,41 +92,50 @@ def output_stream(args, env, request, response):
"""
- prettifier = (OutputProcessor(env, pygments_style=args.style)
- if args.prettify else None)
+ # Pick the right stream type for this exchange based on `env` and `args`.
+ if not env.stdout_isatty and not args.prettify:
+ Stream = partial(
+ RawStream,
+ chunk_size=RawStream.CHUNK_SIZE_BY_LINE
+ if args.stream
+ else RawStream.CHUNK_SIZE)
+ elif args.prettify:
+ Stream = partial(
+ PrettyStream if args.stream else BufferedPrettyStream,
+ processor=OutputProcessor(env, pygments_style=args.style),
+ env=env)
+ else:
+ Stream = partial(EncodedStream, env=env)
- with_request = (OUT_REQ_HEAD in args.output_options
- or OUT_REQ_BODY in args.output_options)
- with_response = (OUT_RESP_HEAD in args.output_options
- or OUT_RESP_BODY in args.output_options)
+ req_h = OUT_REQ_HEAD in args.output_options
+ req_b = OUT_REQ_BODY in args.output_options
+ resp_h = OUT_RESP_HEAD in args.output_options
+ resp_b = OUT_RESP_BODY in args.output_options
- if with_request:
- request_iter = formatted_stream(
- msg=HTTPRequest(request),
- env=env,
- prettifier=prettifier,
- with_headers=OUT_REQ_HEAD in args.output_options,
- with_body=OUT_REQ_BODY in args.output_options)
+ req = req_h or req_b
+ resp = resp_h or resp_b
- for chunk in request_iter:
- yield chunk
+ output = []
- if with_request and with_response:
- yield b'\n\n\n'
+ if req:
+ output.append(Stream(
+ msg=HTTPRequest(request),
+ with_headers=req_h,
+ with_body=req_b))
- if with_response:
- response_iter = formatted_stream(
- msg=HTTPResponse(response),
- env=env,
- prettifier=prettifier,
- with_headers=OUT_RESP_HEAD in args.output_options,
- with_body=OUT_RESP_BODY in args.output_options)
+ if req and resp:
+ output.append([b'\n\n\n'])
- for chunk in response_iter:
- yield chunk
+ if resp:
+ output.append(Stream(
+ msg=HTTPResponse(response),
+ with_headers=resp_h,
+ with_body=resp_b))
if env.stdout_isatty:
- yield b'\n\n'
+ output.append([b'\n\n'])
+
+ return chain(*output)
def get_exist_status(code, allow_redirects=False):
@@ -170,18 +186,30 @@ def main(args=sys.argv[1:], env=Environment()):
except AttributeError:
buffer = env.stdout
- for chunk in output_stream(args, env, response.request, response):
- buffer.write(chunk)
- if env.stdout_isatty:
- env.stdout.flush()
+ try:
+ for chunk in output_stream(args, env, response.request, response):
+ buffer.write(chunk)
+ if env.stdout_isatty or args.stream:
+ env.stdout.flush()
+
+ except IOError as e:
+ if debug:
+ raise
+ if e.errno == errno.EPIPE:
+ env.stderr.write('\n')
+ else:
+ env.stderr.write(str(e) + '\n')
+ return 1
except (KeyboardInterrupt, SystemExit):
+ if debug:
+ raise
env.stderr.write('\n')
return 1
except Exception as e:
if debug:
raise
- env.stderr.write(str(e.message) + '\n')
+ env.stderr.write(str(e) + '\n')
return 1
return status
View
96 httpie/models.py
@@ -18,6 +18,10 @@ class Environment(object):
if progname not in ['http', 'https']:
progname = 'http'
+ if is_windows:
+ import colorama.initialise
+ colorama.initialise.init()
+
stdin_isatty = sys.stdin.isatty()
stdin = sys.stdin
stdout_isatty = sys.stdout.isatty()
@@ -30,50 +34,65 @@ class Environment(object):
def __init__(self, **kwargs):
self.__dict__.update(**kwargs)
- def init_colors(self):
- # We check for real Window here, not self.is_windows as
- # it could be mocked.
- if (is_windows and not self.__colors_initialized
- and self.stdout == sys.stdout):
- import colorama.initialise
- self.stdout = colorama.initialise.wrap_stream(
- self.stdout, autoreset=False,
- convert=None, strip=None, wrap=True)
- self.__colors_initialized = True
- __colors_initialized = False
-
class HTTPMessage(object):
- """Model representing an HTTP message."""
+ """Abstract class for HTTP messages."""
def __init__(self, orig):
self._orig = orig
+ def iter_body(self, chunk_size):
+ """Return an iterator over the body."""
+ raise NotImplementedError()
+
+ def iter_lines(self, chunk_size):
+ """Return an iterator over the body yielding (`line`, `line_feed`)."""
+ raise NotImplementedError()
+
+ @property
+ def headers(self):
+ """Return a `str` with the message's headers."""
+ raise NotImplementedError()
+
+ @property
+ def encoding(self):
+ """Return a `str` with the message's encoding, if known."""
+ raise NotImplementedError()
+
+ @property
+ def body(self):
+ """Return a `bytes` with the message's body."""
+ raise NotImplementedError()
+
@property
def content_type(self):
- return str(self._orig.headers.get('Content-Type', ''))
+ """Return the message content type."""
+ ct = self._orig.headers.get('Content-Type', '')
+ if isinstance(ct, bytes):
+ ct = ct.decode()
+ return ct
class HTTPResponse(HTTPMessage):
"""A `requests.models.Response` wrapper."""
- def __iter__(self):
- mb = 1024 * 1024
- return self._orig.iter_content(chunk_size=2 * mb)
+ def iter_body(self, chunk_size=1):
+ return self._orig.iter_content(chunk_size=chunk_size)
+
+ def iter_lines(self, chunk_size):
+ for line in self._orig.iter_lines(chunk_size):
+ yield line, b'\n'
@property
- def line(self):
- """Return Status-Line"""
+ def headers(self):
original = self._orig.raw._original_response
- return str('HTTP/{version} {status} {reason}'.format(
+ status_line = 'HTTP/{version} {status} {reason}'.format(
version='.'.join(str(original.version)),
status=original.status,
reason=original.reason
- ))
-
- @property
- def headers(self):
- return str(self._orig.raw._original_response.msg)
+ )
+ headers = str(original.msg)
+ return '\n'.join([status_line, headers]).strip()
@property
def encoding(self):
@@ -89,11 +108,14 @@ def body(self):
class HTTPRequest(HTTPMessage):
"""A `requests.models.Request` wrapper."""
- def __iter__(self):
+ def iter_body(self, chunk_size):
yield self.body
+ def iter_lines(self, chunk_size):
+ yield self.body, b''
+
@property
- def line(self):
+ def headers(self):
"""Return Request-Line"""
url = urlparse(self._orig.url)
@@ -111,27 +133,23 @@ def line(self):
qs += type(self._orig)._encode_params(self._orig.params)
# Request-Line
- return str('{method} {path}{query} HTTP/1.1'.format(
+ request_line = '{method} {path}{query} HTTP/1.1'.format(
method=self._orig.method,
path=url.path or '/',
query=qs
- ))
+ )
- @property
- def headers(self):
headers = dict(self._orig.headers)
- content_type = headers.get('Content-Type')
-
- if isinstance(content_type, bytes):
- # Happens when uploading files.
- # TODO: submit a bug report for Requests
- headers['Content-Type'] = str(content_type)
if 'Host' not in headers:
headers['Host'] = urlparse(self._orig.url).netloc
- return '\n'.join('%s: %s' % (name, value)
- for name, value in headers.items())
+ headers = ['%s: %s' % (name, value)
+ for name, value in headers.items()]
+
+ headers.insert(0, request_line)
+
+ return '\n'.join(headers).strip()
@property
def encoding(self):
View
303 httpie/output.py
@@ -1,7 +1,6 @@
-"""Output processing and formatting.
+"""Output streaming, processing and formatting.
"""
-import re
import json
import pygments
@@ -17,92 +16,193 @@
from .models import Environment
-DEFAULT_STYLE = 'solarized'
-AVAILABLE_STYLES = [DEFAULT_STYLE] + list(STYLE_MAP.keys())
+# Colors on Windows via colorama aren't that great and fruity
+# seems to give the best result there.
+DEFAULT_STYLE = 'solarized' if not is_windows else 'fruity'
+
+#noinspection PySetFunctionToLiteral
+AVAILABLE_STYLES = set([DEFAULT_STYLE]) | set(STYLE_MAP.keys())
+
+
BINARY_SUPPRESSED_NOTICE = (
- '+-----------------------------------------+\n'
- '| NOTE: binary data not shown in terminal |\n'
- '+-----------------------------------------+'
+ b'\n'
+ b'+-----------------------------------------+\n'
+ b'| NOTE: binary data not shown in terminal |\n'
+ b'+-----------------------------------------+'
)
-def formatted_stream(msg, prettifier=None, with_headers=True, with_body=True,
- env=Environment()):
- """Return an iterator yielding `bytes` representing `msg`
- (a `models.HTTPMessage` subclass).
+class BinarySuppressedError(Exception):
+ """An error indicating that the body is binary and won't be written,
+ e.g., for terminal output)."""
- The body can be binary so we always yield `bytes`.
+ message = BINARY_SUPPRESSED_NOTICE
- If `prettifier` is set or the output is a terminal then a binary
- body is not included in the output and is replaced with notice.
- Generally, when the `stdout` is redirected, the output matches the actual
- message as much as possible (formatting and character encoding-wise).
- When `--pretty` is set (or implied), or when the output is a terminal,
- then we prefer readability over precision.
+###############################################################################
+# Output Streams
+###############################################################################
+
+class BaseStream(object):
+ """Base HTTP message stream class."""
+
+ def __init__(self, msg, with_headers=True, with_body=True):
+ """
+ :param msg: a :class:`models.HTTPMessage` subclass
+ :param with_headers: if `True`, headers will be included
+ :param with_body: if `True`, body will be included
+
+ """
+ self.msg = msg
+ self.with_headers = with_headers
+ self.with_body = with_body
+
+ def _headers(self):
+ """Return the headers' bytes."""
+ return self.msg.headers.encode('ascii')
+
+ def _body(self):
+ """Return an iterator over the message body."""
+ raise NotImplementedError()
+
+ def __iter__(self):
+ """Return an iterator over `self.msg`."""
+ if self.with_headers:
+ yield self._headers()
+
+ if self.with_body:
+ it = self._body()
- """
- # Output encoding.
- if env.stdout_isatty:
- # Use encoding suitable for the terminal. Unsupported characters
- # will be replaced in the output.
- errors = 'replace'
- output_encoding = getattr(env.stdout, 'encoding', None)
- else:
- # Preserve the message encoding.
- errors = 'strict'
- output_encoding = msg.encoding
- if not output_encoding:
- # Default to utf8
- output_encoding = 'utf8'
-
- if prettifier:
- env.init_colors()
-
- if with_headers:
- headers = '\n'.join([msg.line, msg.headers])
-
- if prettifier:
- headers = prettifier.process_headers(headers)
-
- yield headers.encode(output_encoding, errors).strip()
-
- if with_body:
-
- prefix = b'\n\n' if with_headers else None
-
- if not (env.stdout_isatty or prettifier):
- # Verbatim body even if it's binary.
- for body_chunk in msg:
- if prefix:
- yield prefix
- prefix = None
- yield body_chunk
- elif msg.body:
try:
- body = msg.body.decode(msg.encoding)
- except UnicodeDecodeError:
- # Suppress binary data.
- body = BINARY_SUPPRESSED_NOTICE.encode(output_encoding)
- if not with_headers:
- yield b'\n'
- else:
- if prettifier and msg.content_type:
- body = prettifier.process_body(
- body, msg.content_type).strip()
-
- body = body.encode(output_encoding, errors)
- if prefix:
- yield prefix
- yield body
+ if self.with_headers:
+ # Yield the headers/body separator only if needed.
+ chunk = next(it)
+ if chunk:
+ yield b'\n\n'
+ yield chunk
+
+ for chunk in it:
+ yield chunk
+
+ except BinarySuppressedError as e:
+ yield e.message
+
+
+class RawStream(BaseStream):
+ """The message is streamed in chunks with no processing."""
+
+ CHUNK_SIZE = 1024 * 100
+ CHUNK_SIZE_BY_LINE = 1024 * 5
+
+ def __init__(self, chunk_size=CHUNK_SIZE, **kwargs):
+ super(RawStream, self).__init__(**kwargs)
+ self.chunk_size = chunk_size
+
+ def _body(self):
+ return self.msg.iter_body(self.chunk_size)
+
+
+class EncodedStream(BaseStream):
+ """Encoded HTTP message stream.
+
+ The message bytes are converted to an encoding suitable for
+ `self.env.stdout`. Unicode errors are replaced and binary data
+ is suppressed. The body is always streamed by line.
+
+ """
+ CHUNK_SIZE = 1024 * 5
+ def __init__(self, env=Environment(), **kwargs):
+
+ super(EncodedStream, self).__init__(**kwargs)
+
+ if env.stdout_isatty:
+ # Use the encoding supported by the terminal.
+ output_encoding = getattr(env.stdout, 'encoding', None)
+ else:
+ # Preserve the message encoding.
+ output_encoding = self.msg.encoding
+
+ # Default to utf8 when unsure.
+ self.output_encoding = output_encoding or 'utf8'
+
+ def _body(self):
+
+ for line, lf in self.msg.iter_lines(self.CHUNK_SIZE):
+
+ if b'\0' in line:
+ raise BinarySuppressedError()
+
+ yield line.decode(self.msg.encoding)\
+ .encode(self.output_encoding, 'replace') + lf
+
+
+class PrettyStream(EncodedStream):
+ """In addition to :class:`EncodedStream` behaviour, this stream applies
+ content processing.
+ Useful for long-lived HTTP responses that stream by lines
+ such as the Twitter streaming API.
+
+ """
+
+ CHUNK_SIZE = 1024 * 5
+
+ def __init__(self, processor, **kwargs):
+ super(PrettyStream, self).__init__(**kwargs)
+ self.processor = processor
+
+ def _headers(self):
+ return self.processor.process_headers(
+ self.msg.headers).encode(self.output_encoding)
+
+ def _body(self):
+ for line, lf in self.msg.iter_lines(self.CHUNK_SIZE):
+ if b'\0' in line:
+ raise BinarySuppressedError()
+ yield self._process_body(line) + lf
+
+ def _process_body(self, chunk):
+ return (self.processor
+ .process_body(
+ chunk.decode(self.msg.encoding, 'replace'),
+ self.msg.content_type)
+ .encode(self.output_encoding, 'replace'))
+
+
+class BufferedPrettyStream(PrettyStream):
+ """The same as :class:`PrettyStream` except that the body is fully
+ fetched before it's processed.
+
+ Suitable regular HTTP responses.
+
+ """
+
+ CHUNK_SIZE = 1024 * 10
+
+ def _body(self):
+
+ #noinspection PyArgumentList
+ # Read the whole body before prettifying it,
+ # but bail out immediately if the body is binary.
+ body = bytearray()
+ for chunk in self.msg.iter_body(self.CHUNK_SIZE):
+ if b'\0' in chunk:
+ raise BinarySuppressedError()
+ body.extend(chunk)
+
+ yield self._process_body(body)
+
+
+###############################################################################
+# Processing
+###############################################################################
class HTTPLexer(lexer.RegexLexer):
"""Simplified HTTP lexer for Pygments.
It only operates on headers and provides a stronger contrast between
their names and values than the original one bundled with Pygments
- (`pygments.lexers.text import HttpLexer`), especially when
+ (:class:`pygments.lexers.text import HttpLexer`), especially when
Solarized color scheme is used.
"""
@@ -111,7 +211,6 @@ class HTTPLexer(lexer.RegexLexer):
filenames = ['*.http']
tokens = {
'root': [
-
# Request-Line
(r'([A-Z]+)( +)([^ ]+)( +)(HTTP)(/)(\d+\.\d+)',
lexer.bygroups(
@@ -123,7 +222,6 @@ class HTTPLexer(lexer.RegexLexer):
token.Operator,
token.Number
)),
-
# Response Status-Line
(r'(HTTP)(/)(\d+\.\d+)( +)(\d{3})( +)(.+)',
lexer.bygroups(
@@ -135,7 +233,6 @@ class HTTPLexer(lexer.RegexLexer):
token.Text,
token.Name.Exception, # Reason
)),
-
# Header
(r'(.*?)( *)(:)( *)(.+)', lexer.bygroups(
token.Name.Attribute, # Name
@@ -148,28 +245,48 @@ class HTTPLexer(lexer.RegexLexer):
class BaseProcessor(object):
+ """Base, noop output processor class."""
enabled = True
def __init__(self, env, **kwargs):
+ """
+ :param env:
+ an class:`Environment` instance
+ :param kwargs:
+ additional keyword argument that some processor might require.
+
+ """
self.env = env
self.kwargs = kwargs
def process_headers(self, headers):
+ """Return processed `headers`
+
+ :param headers:
+ The headers as text.
+
+ """
return headers
def process_body(self, content, content_type, subtype):
"""Return processed `content`.
- :param content: `str`
- :param content_type: full content type, e.g., 'application/atom+xml'
- :param subtype: e.g., 'xml'
+ :param content:
+ The body content as text
+
+ :param content_type:
+ Full content type, e.g., 'application/atom+xml'.
+
+ :param subtype:
+ E.g. 'xml'.
"""
return content
class JSONProcessor(BaseProcessor):
+ """JSON body processor."""
def process_body(self, content, content_type, subtype):
if subtype == 'json':
@@ -187,21 +304,26 @@ def process_body(self, content, content_type, subtype):
class PygmentsProcessor(BaseProcessor):
+ """A processor that applies syntax-highlighting using Pygments
+ to the headers, and to the body as well if its content type is recognized.
+ """
def __init__(self, *args, **kwargs):
super(PygmentsProcessor, self).__init__(*args, **kwargs)
+ # Cache that speeds up when we process streamed body by line.
+ self.lexers_by_type = {}
+
if not self.env.colors:
self.enabled = False
return
try:
- style = get_style_by_name(
- self.kwargs.get('pygments_style', DEFAULT_STYLE))
+ style = get_style_by_name(self.kwargs['pygments_style'])
except ClassNotFound:
style = Solarized256Style
- if is_windows or self.env.colors == 256:
+ if self.env.is_windows or self.env.colors == 256:
fmt_class = Terminal256Formatter
else:
fmt_class = TerminalFormatter
@@ -209,24 +331,26 @@ def __init__(self, *args, **kwargs):
def process_headers(self, headers):
return pygments.highlight(
- headers, HTTPLexer(), self.formatter)
+ headers, HTTPLexer(), self.formatter).strip()
def process_body(self, content, content_type, subtype):
try:
- try:
- lexer = get_lexer_for_mimetype(content_type)
- except ClassNotFound:
- lexer = get_lexer_by_name(subtype)
+ lexer = self.lexers_by_type.get(content_type)
+ if not lexer:
+ try:
+ lexer = get_lexer_for_mimetype(content_type)
+ except ClassNotFound:
+ lexer = get_lexer_by_name(subtype)
+ self.lexers_by_type[content_type] = lexer
except ClassNotFound:
pass
else:
content = pygments.highlight(content, lexer, self.formatter)
- return content
+ return content.strip()
class HeadersProcessor(BaseProcessor):
- """
- Sorts headers by name retaining relative order of multiple headers
+ """Sorts headers by name retaining relative order of multiple headers
with the same name.
"""
@@ -237,6 +361,7 @@ def process_headers(self, headers):
class OutputProcessor(object):
+ """A delegate class that invokes the actual processors."""
installed_processors = [
JSONProcessor,
View
433 tests/tests.py
@@ -22,64 +22,94 @@
import os
import sys
import json
-import unittest
import argparse
import tempfile
+import unittest
+try:
+ from unittest import skipIf
+except ImportError:
+ def skipIf(cond, test_method):
+ if cond:
+ return test_method
+ return lambda self: None
try:
from urllib.request import urlopen
except ImportError:
from urllib2 import urlopen
-import requests
-from requests.compat import is_py26, is_py3, bytes, str
+from requests.compat import is_windows, is_py26, bytes, str
+
#################################################################
# Utils/setup
#################################################################
# HACK: Prepend ../ to PYTHONPATH so that we can import httpie form there.
-TESTS_ROOT = os.path.dirname(__file__)
+TESTS_ROOT = os.path.abspath(os.path.dirname(__file__))
sys.path.insert(0, os.path.realpath(os.path.join(TESTS_ROOT, '..')))
from httpie import input
from httpie.models import Environment
-from httpie.core import main, output_stream
+from httpie.core import main
from httpie.output import BINARY_SUPPRESSED_NOTICE
from httpie.input import ParseError
HTTPBIN_URL = os.environ.get('HTTPBIN_URL',
- 'http://httpbin.org')
+ 'http://httpbin.org').rstrip('/')
+
+
+OK = 'HTTP/1.1 200'
+OK_COLOR = (
+ 'HTTP\x1b[39m\x1b[38;5;245m/\x1b[39m\x1b'
+ '[38;5;37m1.1\x1b[39m\x1b[38;5;245m \x1b[39m\x1b[38;5;37m200'
+ '\x1b[39m\x1b[38;5;245m \x1b[39m\x1b[38;5;136mOK'
+)
+COLOR = '\x1b['
+
-TEST_FILE_PATH = os.path.join(TESTS_ROOT, 'fixtures', 'file.txt')
-TEST_FILE2_PATH = os.path.join(TESTS_ROOT, 'fixtures', 'file2.txt')
+def patharg(path):
+ """Back slashes need to be escaped in ITEM args, even in Windows paths."""
+ return path.replace('\\', '\\\\\\')
-with open(TEST_FILE_PATH) as f:
- TEST_FILE_CONTENT = f.read().strip()
+# Test files
+FILE_PATH = os.path.join(TESTS_ROOT, 'fixtures', 'file.txt')
+FILE2_PATH = os.path.join(TESTS_ROOT, 'fixtures', 'file2.txt')
+BIN_FILE_PATH = os.path.join(TESTS_ROOT, 'fixtures', 'file.bin')
-TEST_BIN_FILE_PATH = os.path.join(TESTS_ROOT, 'fixtures', 'file.bin')
-with open(TEST_BIN_FILE_PATH, 'rb') as f:
- TEST_BIN_FILE_CONTENT = f.read()
+FILE_PATH_ARG = patharg(FILE_PATH)
+FILE2_PATH_ARG = patharg(FILE2_PATH)
+BIN_FILE_PATH_ARG = patharg(BIN_FILE_PATH)
-TERMINAL_COLOR_PRESENCE_CHECK = '\x1b['
+with open(FILE_PATH) as f:
+ FILE_CONTENT = f.read().strip()
+with open(BIN_FILE_PATH, 'rb') as f:
+ BIN_FILE_CONTENT = f.read()
def httpbin(path):
return HTTPBIN_URL + path
-class ResponseMixin(object):
- exit_status = None
- stderr = None
- json = None
+class TestEnvironment(Environment):
+ colors = 0
+ stdin_isatty = True,
+ stdout_isatty = True
+ is_windows = False
+ def __init__(self, **kwargs):
-class BytesResponse(bytes, ResponseMixin):
- pass
+ if 'stdout' not in kwargs:
+ kwargs['stdout'] = tempfile.TemporaryFile('w+b')
+ if 'stderr' not in kwargs:
+ kwargs['stderr'] = tempfile.TemporaryFile('w+t')
-class StrResponse(str, ResponseMixin):
- pass
+ super(TestEnvironment, self).__init__(**kwargs)
+
+
+class BytesResponse(bytes): pass
+class StrResponse(str): pass
def http(*args, **kwargs):
@@ -87,37 +117,41 @@ def http(*args, **kwargs):
Invoke `httpie.core.main()` with `args` and `kwargs`,
and return a unicode response.
- """
- if 'env' not in kwargs:
- # Ensure that we have terminal by default (needed for Travis).
- kwargs['env'] = Environment(
- colors=0,
- stdin_isatty=True,
- stdout_isatty=True,
- )
+ Return a `StrResponse`, or `BytesResponse` if unable to decode the output.
+ The response has the following attributes:
+
+ `stderr`: text written to stderr
+ `exit_status`: the exit status
+ `json`: decoded JSON (if possible) or `None`
- stdout = kwargs['env'].stdout = tempfile.TemporaryFile('w+b')
- stderr = kwargs['env'].stderr = tempfile.TemporaryFile('w+t')
+ Exceptions are propagated except for SystemExit.
+
+ """
+ env = kwargs.get('env')
+ if not env:
+ env = kwargs['env'] = TestEnvironment()
try:
- exit_status = main(args=['--debug'] + list(args), **kwargs)
- except (Exception, SystemExit) as e:
- sys.stderr.write(stderr.read())
- raise
- else:
- stdout.seek(0)
- stderr.seek(0)
- output = stdout.read()
+ try:
+ exit_status = main(args=['--debug'] + list(args), **kwargs)
+ except Exception:
+ sys.stderr.write(env.stderr.read())
+ raise
+ except SystemExit:
+ exit_status = 1
+
+ env.stdout.seek(0)
+ env.stderr.seek(0)
+
+ output = env.stdout.read()
try:
- #noinspection PyArgumentList
r = StrResponse(output.decode('utf8'))
except UnicodeDecodeError:
- #noinspection PyArgumentList
r = BytesResponse(output)
else:
- if TERMINAL_COLOR_PRESENCE_CHECK not in r:
+ if COLOR not in r:
# De-serialize JSON body if possible.
if r.strip().startswith('{'):
#noinspection PyTypeChecker
@@ -133,14 +167,14 @@ def http(*args, **kwargs):
except ValueError:
pass
- r.stderr = stderr.read()
+ r.stderr = env.stderr.read()
r.exit_status = exit_status
- return r
- finally:
- stdout.close()
- stderr.close()
+ return r
+ finally:
+ env.stdout.close()
+ env.stderr.close()
class BaseTestCase(unittest.TestCase):
@@ -168,14 +202,14 @@ def test_GET(self):
'GET',
httpbin('/get')
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
def test_DELETE(self):
r = http(
'DELETE',
httpbin('/delete')
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
def test_PUT(self):
r = http(
@@ -183,7 +217,7 @@ def test_PUT(self):
httpbin('/put'),
'foo=bar'
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('"foo": "bar"', r)
def test_POST_JSON_data(self):
@@ -192,7 +226,7 @@ def test_POST_JSON_data(self):
httpbin('/post'),
'foo=bar'
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('"foo": "bar"', r)
def test_POST_form(self):
@@ -202,7 +236,7 @@ def test_POST_form(self):
httpbin('/post'),
'foo=bar'
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('"foo": "bar"', r)
def test_POST_form_multiple_values(self):
@@ -213,19 +247,17 @@ def test_POST_form_multiple_values(self):
'foo=bar',
'foo=baz',
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertDictEqual(r.json['form'], {
'foo': ['bar', 'baz']
})
def test_POST_stdin(self):
- with open(TEST_FILE_PATH) as f:
- env = Environment(
+ with open(FILE_PATH) as f:
+ env = TestEnvironment(
stdin=f,
stdin_isatty=False,
- stdout_isatty=True,
- colors=0,
)
r = http(
@@ -234,8 +266,8 @@ def test_POST_stdin(self):
httpbin('/post'),
env=env
)
- self.assertIn('HTTP/1.1 200', r)
- self.assertIn(TEST_FILE_CONTENT, r)
+ self.assertIn(OK, r)
+ self.assertIn(FILE_CONTENT, r)
def test_headers(self):
r = http(
@@ -243,7 +275,7 @@ def test_headers(self):
httpbin('/headers'),
'Foo:bar'
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('"User-Agent": "HTTPie', r)
self.assertIn('"Foo": "bar"', r)
@@ -260,7 +292,7 @@ def test_query_string_params_in_url(self):
path = '/get?a=1&b=2'
url = httpbin(path)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('GET %s HTTP/1.1' % path, r)
self.assertIn('"url": "%s"' % url, r)
@@ -276,7 +308,7 @@ def test_query_string_params_items(self):
path = '/get?a=1&b=2'
url = httpbin(path)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('GET %s HTTP/1.1' % path, r)
self.assertIn('"url": "%s"' % url, r)
@@ -293,7 +325,7 @@ def test_query_string_params_in_url_and_items_with_duplicates(self):
path = '/get?a=1&a=1&a=1&a=1&b=2'
url = httpbin(path)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('GET %s HTTP/1.1' % path, r)
self.assertIn('"url": "%s"' % url, r)
@@ -311,7 +343,7 @@ def test_GET_no_data_no_auto_headers(self):
'GET',
httpbin('/headers')
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('"Accept": "*/*"', r)
self.assertNotIn('"Content-Type": "application/json', r)
@@ -321,7 +353,7 @@ def test_POST_no_data_no_auto_headers(self):
'POST',
httpbin('/post')
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('"Accept": "*/*"', r)
self.assertNotIn('"Content-Type": "application/json', r)
@@ -331,7 +363,7 @@ def test_POST_with_data_auto_JSON_headers(self):
httpbin('/post'),
'a=b'
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('"Accept": "application/json"', r)
self.assertIn('"Content-Type": "application/json; charset=utf-8', r)
@@ -342,7 +374,7 @@ def test_GET_with_data_auto_JSON_headers(self):
httpbin('/post'),
'a=b'
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('"Accept": "application/json"', r)
self.assertIn('"Content-Type": "application/json; charset=utf-8', r)
@@ -352,7 +384,7 @@ def test_POST_explicit_JSON_auto_JSON_headers(self):
'POST',
httpbin('/post')
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('"Accept": "application/json"', r)
self.assertIn('"Content-Type": "application/json; charset=utf-8', r)
@@ -364,7 +396,7 @@ def test_GET_explicit_JSON_explicit_headers(self):
'Accept:application/xml',
'Content-Type:application/xml'
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('"Accept": "application/xml"', r)
self.assertIn('"Content-Type": "application/xml"', r)
@@ -374,7 +406,7 @@ def test_POST_form_auto_Content_Type(self):
'POST',
httpbin('/post')
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn(
'"Content-Type":'
' "application/x-www-form-urlencoded; charset=utf-8"',
@@ -388,7 +420,7 @@ def test_POST_form_Content_Type_override(self):
httpbin('/post'),
'Content-Type:application/xml'
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('"Content-Type": "application/xml"', r)
def test_print_only_body_when_stdout_redirected_by_default(self):
@@ -396,7 +428,7 @@ def test_print_only_body_when_stdout_redirected_by_default(self):
r = http(
'GET',
httpbin('/get'),
- env=Environment(
+ env=TestEnvironment(
stdin_isatty=True,
stdout_isatty=False
)
@@ -409,26 +441,26 @@ def test_print_overridable_when_stdout_redirected(self):
'--print=h',
'GET',
httpbin('/get'),
- env=Environment(
+ env=TestEnvironment(
stdin_isatty=True,
stdout_isatty=False
)
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
class ImplicitHTTPMethodTest(BaseTestCase):
def test_implicit_GET(self):
r = http(httpbin('/get'))
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
def test_implicit_GET_with_headers(self):
r = http(
httpbin('/headers'),
'Foo:bar'
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('"Foo": "bar"', r)
def test_implicit_POST_json(self):
@@ -436,7 +468,7 @@ def test_implicit_POST_json(self):
httpbin('/post'),
'hello=world'
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('"hello": "world"', r)
def test_implicit_POST_form(self):
@@ -445,23 +477,21 @@ def test_implicit_POST_form(self):
httpbin('/post'),
'foo=bar'
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('"foo": "bar"', r)
def test_implicit_POST_stdin(self):
- with open(TEST_FILE_PATH) as f:
- env = Environment(
+ with open(FILE_PATH) as f:
+ env = TestEnvironment(
stdin_isatty=False,
stdin=f,
- stdout_isatty=True,
- colors=0,
)
r = http(
'--form',
httpbin('/post'),
env=env
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
class PrettyFlagTest(BaseTestCase):
@@ -471,31 +501,25 @@ def test_pretty_enabled_by_default(self):
r = http(
'GET',
httpbin('/get'),
- env=Environment(
- stdin_isatty=True,
- stdout_isatty=True,
- ),
+ env=TestEnvironment(colors=256),
)
- self.assertIn(TERMINAL_COLOR_PRESENCE_CHECK, r)
+ self.assertIn(COLOR, r)
def test_pretty_enabled_by_default_unless_stdout_redirected(self):
r = http(
'GET',
httpbin('/get')
)
- self.assertNotIn(TERMINAL_COLOR_PRESENCE_CHECK, r)
+ self.assertNotIn(COLOR, r)
def test_force_pretty(self):
r = http(
'--pretty',
'GET',
httpbin('/get'),
- env=Environment(
- stdin_isatty=True,
- stdout_isatty=False
- ),
+ env=TestEnvironment(stdout_isatty=False, colors=256),
)
- self.assertIn(TERMINAL_COLOR_PRESENCE_CHECK, r)
+ self.assertIn(COLOR, r)
def test_force_ugly(self):
r = http(
@@ -503,7 +527,7 @@ def test_force_ugly(self):
'GET',
httpbin('/get'),
)
- self.assertNotIn(TERMINAL_COLOR_PRESENCE_CHECK, r)
+ self.assertNotIn(COLOR, r)
def test_subtype_based_pygments_lexer_match(self):
"""Test that media subtype is used if type/subtype doesn't
@@ -516,9 +540,9 @@ def test_subtype_based_pygments_lexer_match(self):
httpbin('/post'),
'Content-Type:text/foo+json',
'a=b',
- env=Environment()
+ env=TestEnvironment(colors=256)
)
- self.assertIn(TERMINAL_COLOR_PRESENCE_CHECK, r)
+ self.assertIn(COLOR, r)
class VerboseFlagTest(BaseTestCase):
@@ -530,7 +554,7 @@ def test_verbose(self):
httpbin('/get'),
'test-header:__test__'
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
#noinspection PyUnresolvedReferences
self.assertEqual(r.count('__test__'), 2)
@@ -544,7 +568,7 @@ def test_verbose_form(self):
'foo=bar',
'baz=bar'
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('foo=bar&baz=bar', r)
def test_verbose_json(self):
@@ -555,7 +579,7 @@ def test_verbose_json(self):
'foo=bar',
'baz=bar'
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
#noinspection PyUnresolvedReferences
self.assertEqual(r.count('"baz": "bar"'), 2)
@@ -576,24 +600,24 @@ def test_upload_ok(self):
'--verbose',
'POST',
httpbin('/post'),
- 'test-file@%s' % TEST_FILE_PATH,
+ 'test-file@%s' % FILE_PATH_ARG,
'foo=bar'
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('Content-Disposition: form-data; name="foo"', r)
self.assertIn('Content-Disposition: form-data; name="test-file";'
- ' filename="%s"' % os.path.basename(TEST_FILE_PATH), r)
+ ' filename="%s"' % os.path.basename(FILE_PATH), r)
#noinspection PyUnresolvedReferences
- self.assertEqual(r.count(TEST_FILE_CONTENT), 2)
+ self.assertEqual(r.count(FILE_CONTENT), 2)
self.assertIn('"foo": "bar"', r)
class BinaryRequestDataTest(BaseTestCase):
def test_binary_stdin(self):
- with open(TEST_BIN_FILE_PATH, 'rb') as stdin:
- env = Environment(
+ with open(BIN_FILE_PATH, 'rb') as stdin:
+ env = TestEnvironment(
stdin=stdin,
stdin_isatty=False,
stdout_isatty=False
@@ -604,10 +628,10 @@ def test_binary_stdin(self):
httpbin('/post'),
env=env,
)
- self.assertEqual(r, TEST_BIN_FILE_CONTENT)
+ self.assertEqual(r, BIN_FILE_CONTENT)
def test_binary_file_path(self):
- env = Environment(
+ env = TestEnvironment(
stdin_isatty=True,
stdout_isatty=False
)
@@ -615,14 +639,14 @@ def test_binary_file_path(self):
'--print=B',
'POST',
httpbin('/post'),
- '@' + TEST_BIN_FILE_PATH,
+ '@' + BIN_FILE_PATH_ARG,
env=env,
)
- self.assertEqual(r, TEST_BIN_FILE_CONTENT)
+ self.assertEqual(r, BIN_FILE_CONTENT)
def test_binary_file_form(self):
- env = Environment(
+ env = TestEnvironment(
stdin_isatty=True,
stdout_isatty=False
)
@@ -631,10 +655,10 @@ def test_binary_file_form(self):
'--form',
'POST',
httpbin('/post'),
- 'test@' + TEST_BIN_FILE_PATH,
+ 'test@' + BIN_FILE_PATH_ARG,
env=env,
)
- self.assertIn(bytes(TEST_BIN_FILE_CONTENT), bytes(r))
+ self.assertIn(bytes(BIN_FILE_CONTENT), bytes(r))
class BinaryResponseDataTest(BaseTestCase):
@@ -652,23 +676,23 @@ def test_binary_suppresses_when_terminal(self):
'GET',
self.url
)
- self.assertIn(BINARY_SUPPRESSED_NOTICE, r)
+ self.assertIn(BINARY_SUPPRESSED_NOTICE.decode(), r)
def test_binary_suppresses_when_not_terminal_but_pretty(self):
r = http(
'--pretty',
'GET',
self.url,
- env=Environment(stdin_isatty=True,
+ env=TestEnvironment(stdin_isatty=True,
stdout_isatty=False)
)
- self.assertIn(BINARY_SUPPRESSED_NOTICE, r)
+ self.assertIn(BINARY_SUPPRESSED_NOTICE.decode(), r)
def test_binary_included_and_correct_when_suitable(self):
r = http(
'GET',
self.url,
- env=Environment(stdin_isatty=True,
+ env=TestEnvironment(stdin_isatty=True,
stdout_isatty=False)
)
self.assertEqual(r, self.bindata)
@@ -683,41 +707,40 @@ def test_request_body_from_file_by_path(self):
r = http(
'POST',
httpbin('/post'),
- '@' + TEST_FILE_PATH
+ '@' + FILE_PATH_ARG
)
- self.assertIn('HTTP/1.1 200', r)
- self.assertIn(TEST_FILE_CONTENT, r)
+ self.assertIn(OK, r)
+ self.assertIn(FILE_CONTENT, r)
self.assertIn('"Content-Type": "text/plain"', r)
def test_request_body_from_file_by_path_with_explicit_content_type(self):
r = http(
'POST',
httpbin('/post'),
- '@' + TEST_FILE_PATH,
+ '@' + FILE_PATH_ARG,
'Content-Type:x-foo/bar'
)
- self.assertIn('HTTP/1.1 200', r)
- self.assertIn(TEST_FILE_CONTENT, r)
+ self.assertIn(OK, r)
+ self.assertIn(FILE_CONTENT, r)
self.assertIn('"Content-Type": "x-foo/bar"', r)
def test_request_body_from_file_by_path_no_field_name_allowed(self):
- env = Environment(stdin_isatty=True)
+ env = TestEnvironment(stdin_isatty=True)
r = http(
'POST',
httpbin('/post'),
- 'field-name@' + TEST_FILE_PATH,
+ 'field-name@' + FILE_PATH_ARG,
env=env
)
self.assertIn('perhaps you meant --form?', r.stderr)
def test_request_body_from_file_by_path_no_data_items_allowed(self):
- env = Environment(stdin_isatty=True)
r = http(
'POST',
httpbin('/post'),
- '@' + TEST_FILE_PATH,
+ '@' + FILE_PATH_ARG,
'foo=bar',
- env=env
+ env=TestEnvironment(stdin_isatty=False)
)
self.assertIn('cannot be mixed', r.stderr)
@@ -730,7 +753,7 @@ def test_basic_auth(self):
'GET',
httpbin('/basic-auth/user/password')
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('"authenticated": true', r)
self.assertIn('"user": "user"', r)
@@ -741,7 +764,7 @@ def test_digest_auth(self):
'GET',
httpbin('/digest-auth/auth/user/password')
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('"authenticated": true', r)
self.assertIn('"user": "user"', r)
@@ -756,7 +779,7 @@ def test_password_prompt(self):
httpbin('/basic-auth/user/password')
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertIn('"authenticated": true', r)
self.assertIn('"user": "user"', r)
@@ -768,7 +791,7 @@ def test_ok_response_exits_0(self):
'GET',
httpbin('/status/200')
)
- self.assertIn('HTTP/1.1 200', r)
+ self.assertIn(OK, r)
self.assertEqual(r.exit_status, 0)
def test_error_response_exits_0_without_check_status(self):
@@ -785,10 +808,7 @@ def test_3xx_check_status_exits_3_and_stderr_when_stdout_redirected(self):
'--headers', # non-terminal, force headers
'GET',
httpbin('/status/301'),
- env=Environment(
- stdout_isatty=False,
- stdin_isatty=True,
- )
+ env=TestEnvironment(stdout_isatty=False,)
)
self.assertIn('HTTP/1.1 301', r)
self.assertEqual(r.exit_status, 3)
@@ -829,7 +849,7 @@ def test_5xx_check_status_exits_5(self):
class FakeWindowsTest(BaseTestCase):
def test_stdout_redirect_not_supported_on_windows(self):
- env = Environment(is_windows=True, stdout_isatty=False)
+ env = TestEnvironment(is_windows=True, stdout_isatty=False)
r = http(
'GET',
httpbin('/get'),
@@ -840,11 +860,6 @@ def test_stdout_redirect_not_supported_on_windows(self):
self.assertIn('--output', r.stderr)
def test_output_file_pretty_not_allowed_on_windows(self):
- env = Environment(
- is_windows=True,
- stdout_isatty=True,
- stdin_isatty=True
- )
r = http(
'--output',
@@ -852,12 +867,71 @@ def test_output_file_pretty_not_allowed_on_windows(self):
'--pretty',
'GET',
httpbin('/get'),
- env=env
+ env=TestEnvironment(is_windows=True)
)
self.assertIn(
'Only terminal output can be prettified on Windows', r.stderr)
+class StreamTest(BaseTestCase):
+ # GET because httpbin 500s with binary POST body.
+
+ @skipIf(is_windows, 'Pretty redirect not supported under Windows')
+ def test_pretty_redirected_stream(self):
+ """Test that --stream works with prettified redirected output."""
+ with open(BIN_FILE_PATH, 'rb') as f:
+ r = http(
+ '--verbose',
+ '--pretty',
+ '--stream',
+ 'GET',
+ httpbin('/get'),
+ env=TestEnvironment(
+ colors=256,
+ stdin=f,
+ stdin_isatty=False,
+ stdout_isatty=False,
+ )
+ )
+ self.assertIn(BINARY_SUPPRESSED_NOTICE.decode(), r)
+ self.assertIn(OK_COLOR, r)
+
+ def test_encoded_stream(self):
+ """Test that --stream works with non-prettified redirected terminal output."""
+ with open(BIN_FILE_PATH, 'rb') as f:
+ r = http(
+ '--ugly',
+ '--stream',
+ '--verbose',
+ 'GET',
+ httpbin('/get'),
+ env=TestEnvironment(
+ stdin=f,
+ stdin_isatty=False
+ ),
+ )
+ self.assertIn(BINARY_SUPPRESSED_NOTICE.decode(), r)
+ self.assertIn(OK, r)
+
+ def test_redirected_stream(self):
+ """Test that --stream works with non-prettified redirected terminal output."""
+ with open(BIN_FILE_PATH, 'rb') as f:
+ r = http(
+ '--ugly',
+ '--stream',
+ '--verbose',
+ 'GET',
+ httpbin('/get'),
+ env=TestEnvironment(
+ stdout_isatty=False,
+ stdin=f,
+ stdin_isatty=False
+ )
+ )
+ self.assertIn(OK.encode(), r)
+ self.assertIn(BIN_FILE_CONTENT, r)
+
+
#################################################################
# CLI argument parsing related tests.
#################################################################
@@ -887,7 +961,7 @@ def test_escape(self):
# data
self.key_value_type('baz\\=bar=foo'),
# files
- self.key_value_type('bar\\@baz@%s' % TEST_FILE_PATH)
+ self.key_value_type('bar\\@baz@%s' % FILE_PATH_ARG)
])
self.assertDictEqual(headers, {
'foo:bar': 'baz',
@@ -915,7 +989,7 @@ def test_valid_items(self):
self.key_value_type('eh:'),
self.key_value_type('ed='),
self.key_value_type('bool:=true'),
- self.key_value_type('test-file@%s' % TEST_FILE_PATH),
+ self.key_value_type('test-file@%s' % FILE_PATH_ARG),
self.key_value_type('query==value'),
])
self.assertDictEqual(headers, {
@@ -946,7 +1020,7 @@ def test_guess_when_method_set_and_valid(self):
args.url = 'http://example.com/'
args.items = []
- self.parser._guess_method(args, Environment())
+ self.parser._guess_method(args, TestEnvironment())
self.assertEqual(args.method, 'GET')
self.assertEqual(args.url, 'http://example.com/')
@@ -958,10 +1032,7 @@ def test_guess_when_method_not_set(self):
args.url = 'http://example.com/'
args.items = []
- self.parser._guess_method(args, Environment(
- stdin_isatty=True,
- stdout_isatty=True,
- ))
+ self.parser._guess_method(args, TestEnvironment())
self.assertEqual(args.method, 'GET')
self.assertEqual(args.url, 'http://example.com/')
@@ -973,7 +1044,7 @@ def test_guess_when_method_set_but_invalid_and_data_field(self):
args.url = 'data=field'
args.items = []
- self.parser._guess_method(args, Environment())
+ self.parser._guess_method(args, TestEnvironment())
self.assertEqual(args.method, 'POST')
self.assertEqual(args.url, 'http://example.com/')
@@ -988,10 +1059,7 @@ def test_guess_when_method_set_but_invalid_and_header_field(self):
args.url = 'test:header'
args.items = []
- self.parser._guess_method(args, Environment(
- stdin_isatty=True,
- stdout_isatty=True,
- ))
+ self.parser._guess_method(args, TestEnvironment())
self.assertEqual(args.method, 'GET')
self.assertEqual(args.url, 'http://example.com/')
@@ -1009,7 +1077,7 @@ def test_guess_when_method_set_but_invalid_and_item_exists(self):
key='old_item', value='b', sep='=', orig='old_item=b')
]
- self.parser._guess_method(args, Environment())
+ self.parser._guess_method(args, TestEnvironment())
self.assertEqual(args.items, [
input.KeyValue(
@@ -1019,57 +1087,6 @@ def test_guess_when_method_set_but_invalid_and_item_exists(self):
])
-class FakeResponse(requests.Response):
-
- class Mock(object):
-
- def __getattr__(self, item):
- return self
-
- def __repr__(self):
- return 'Mock string'
-
- def __unicode__(self):
- return self.__repr__()
-
- def __init__(self, content=None, encoding='utf-8'):
- super(FakeResponse, self).__init__()
- self.headers['Content-Type'] = 'application/json'
- self.encoding = encoding
- self._content = content.encode(encoding)
- self.raw = self.Mock()
-
-
-class UnicodeOutputTestCase(BaseTestCase):
-
- def test_unicode_output(self):
- # some cyrillic and simplified chinese symbols
- response_dict = {'Привет': 'Мир!',
- 'Hello': '世界'}
- if not is_py3:
- response_dict = dict(
- (k.decode('utf8'), v.decode('utf8'))
- for k, v in response_dict.items()
- )
- response_body = json.dumps(response_dict)
- # emulate response
- response = FakeResponse(response_body)
-
- # emulate cli arguments
- args = argparse.Namespace()
- args.prettify = True
- args.output_options = 'b'
- args.forced_content_type = None
- args.style = 'default'
-
- # colorized output contains escape sequences
- output = output_stream(args, Environment(), response.request, response)
- output = b''.join(output).decode('utf8')
- for key, value in response_dict.items():
- self.assertIn(key, output)
- self.assertIn(value, output)
-
-
if __name__ == '__main__':
#noinspection PyCallingNonCallable
unittest.main()
Please sign in to comment.
Something went wrong with that request. Please try again.