holm/httpie-cli
1
1
Fork 0
forked from extern/httpie-cli
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Streamed terminal output

Browse Source 
`--stream` can be used to enable streaming also with `--pretty` and to ensure
a more frequent output flushing.
This commit is contained in:
Jakub Roztocil 2012-08-03 01:01:15 +02:00
parent 4615011f2e
commit c7657e3c4b
8 changed files with 615 additions and 385 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@ -4,3 +4,6 @@ build
*.pyc
.tox
README.html
.coverage
htmlcov

43
README.rst
View File

@ -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.

3
httpie/__init__.py
View File

@ -5,6 +5,3 @@ HTTPie - cURL for humans.
__author__ = 'Jakub Roztocil'
__version__ = '0.2.7dev'
__licence__ = 'BSD'
CONTENT_TYPE = 'Content-Type'

25
httpie/cli.py
View File

@ -9,7 +9,7 @@ from requests.compat import is_windows
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 @@ group_type.add_argument(
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 @@ output_options.add_argument(
)
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=_('''

100
httpie/core.py
View File

@ -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(
req = req_h or req_b
resp = resp_h or resp_b
output = []
if req:
output.append(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)
with_headers=req_h,
with_body=req_b))
for chunk in request_iter:
yield chunk
if req and resp:
output.append([b'\n\n\n'])
if with_request and with_response:
yield b'\n\n\n'
if with_response:
response_iter = formatted_stream(
if resp:
output.append(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)
for chunk in response_iter:
yield chunk
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

100
httpie/models.py
View File

@ -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)
@property
def line(self):
"""Return Status-Line"""
original = self._orig.raw._original_response
return str('HTTP/{version} {status} {reason}'.format(
version='.'.join(str(original.version)),
status=original.status,
reason=original.reason
))
def iter_lines(self, chunk_size):
for line in self._orig.iter_lines(chunk_size):
yield line, b'\n'
@property
def headers(self):
return str(self._orig.raw._original_response.msg)
original = self._orig.raw._original_response
status_line = 'HTTP/{version} {status} {reason}'.format(
version='.'.join(str(original.version)),
status=original.status,
reason=original.reason
)
headers = str(original.msg)
return '\n'.join([status_line, headers]).strip()
@property
def encoding(self):
@ -89,11 +108,14 @@ class HTTPResponse(HTTPMessage):
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 @@ class HTTPRequest(HTTPMessage):
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):

287
httpie/output.py
View File

@ -1,7 +1,6 @@
"""Output processing and formatting.
"""Output streaming, processing and formatting.
"""
import re
import json
import pygments
@ -17,92 +16,193 @@ from .solarized import Solarized256Style
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()
try:
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.
"""
# 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'
CHUNK_SIZE = 1024 * 5
def __init__(self, env=Environment(), **kwargs):
if prettifier:
env.init_colors()
super(EncodedStream, self).__init__(**kwargs)
if with_headers:
headers = '\n'.join([msg.line, msg.headers])
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
if prettifier:
headers = prettifier.process_headers(headers)
# Default to utf8 when unsure.
self.output_encoding = output_encoding or 'utf8'
yield headers.encode(output_encoding, errors).strip()
def _body(self):
if with_body:
for line, lf in self.msg.iter_lines(self.CHUNK_SIZE):
prefix = b'\n\n' if with_headers else None
if b'\0' in line:
raise BinarySuppressedError()
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()
yield line.decode(self.msg.encoding)\
.encode(self.output_encoding, 'replace') + lf
body = body.encode(output_encoding, errors)
if prefix:
yield prefix
yield body
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 @@ class JSONProcessor(BaseProcessor):
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 @@ class PygmentsProcessor(BaseProcessor):
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 @@ class HeadersProcessor(BaseProcessor):
class OutputProcessor(object):
"""A delegate class that invokes the actual processors."""
installed_processors = [
JSONProcessor,

439
tests/tests.py
View File

@ -22,64 +22,94 @@ To make it run faster and offline you can::
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('/')
TEST_FILE_PATH = os.path.join(TESTS_ROOT, 'fixtures', 'file.txt')
TEST_FILE2_PATH = os.path.join(TESTS_ROOT, 'fixtures', 'file2.txt')
with open(TEST_FILE_PATH) as f:
TEST_FILE_CONTENT = f.read().strip()
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_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()
TERMINAL_COLOR_PRESENCE_CHECK = '\x1b['
def patharg(path):
"""Back slashes need to be escaped in ITEM args, even in Windows paths."""
return path.replace('\\', '\\\\\\')
# 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')
FILE_PATH_ARG = patharg(FILE_PATH)
FILE2_PATH_ARG = patharg(FILE2_PATH)
BIN_FILE_PATH_ARG = patharg(BIN_FILE_PATH)
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):
if 'stdout' not in kwargs:
kwargs['stdout'] = tempfile.TemporaryFile('w+b')
if 'stderr' not in kwargs:
kwargs['stderr'] = tempfile.TemporaryFile('w+t')
super(TestEnvironment, self).__init__(**kwargs)
class BytesResponse(bytes, ResponseMixin):
pass
class StrResponse(str, ResponseMixin):
pass
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:
stdout = kwargs['env'].stdout = tempfile.TemporaryFile('w+b')
stderr = kwargs['env'].stderr = tempfile.TemporaryFile('w+t')
`stderr`: text written to stderr
`exit_status`: the exit status
`json`: decoded JSON (if possible) or `None`
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:
#noinspection PyArgumentList
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:
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()
env.stdout.close()
env.stderr.close()
class BaseTestCase(unittest.TestCase):
@ -168,14 +202,14 @@ class HTTPieTest(BaseTestCase):
'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 @@ class HTTPieTest(BaseTestCase):
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 @@ class HTTPieTest(BaseTestCase):
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 @@ class HTTPieTest(BaseTestCase):
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 @@ class HTTPieTest(BaseTestCase):
'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 @@ class HTTPieTest(BaseTestCase):
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 @@ class HTTPieTest(BaseTestCase):
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 @@ class QuerystringTest(BaseTestCase):
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 @@ class QuerystringTest(BaseTestCase):
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 @@ class QuerystringTest(BaseTestCase):
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 @@ class AutoContentTypeAndAcceptHeadersTest(BaseTestCase):
'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 @@ class AutoContentTypeAndAcceptHeadersTest(BaseTestCase):
'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 @@ class AutoContentTypeAndAcceptHeadersTest(BaseTestCase):
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 @@ class AutoContentTypeAndAcceptHeadersTest(BaseTestCase):
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 @@ class AutoContentTypeAndAcceptHeadersTest(BaseTestCase):
'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 @@ class AutoContentTypeAndAcceptHeadersTest(BaseTestCase):
'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 @@ class AutoContentTypeAndAcceptHeadersTest(BaseTestCase):
'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 @@ class AutoContentTypeAndAcceptHeadersTest(BaseTestCase):
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 @@ class AutoContentTypeAndAcceptHeadersTest(BaseTestCase):
r = http(
'GET',
httpbin('/get'),
env=Environment(
env=TestEnvironment(
stdin_isatty=True,
stdout_isatty=False
)
@ -409,26 +441,26 @@ class AutoContentTypeAndAcceptHeadersTest(BaseTestCase):
'--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 @@ class ImplicitHTTPMethodTest(BaseTestCase):
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 @@ class ImplicitHTTPMethodTest(BaseTestCase):
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 @@ class PrettyFlagTest(BaseTestCase):
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 @@ class PrettyFlagTest(BaseTestCase):
'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 @@ class PrettyFlagTest(BaseTestCase):
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 @@ class VerboseFlagTest(BaseTestCase):
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 @@ class VerboseFlagTest(BaseTestCase):
'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 @@ class VerboseFlagTest(BaseTestCase):
'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 @@ class MultipartFormDataFileUploadTest(BaseTestCase):
'--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 @@ class BinaryRequestDataTest(BaseTestCase):
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 @@ class BinaryRequestDataTest(BaseTestCase):
'--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 @@ class BinaryRequestDataTest(BaseTestCase):
'--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 @@ class BinaryResponseDataTest(BaseTestCase):
'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 @@ class RequestBodyFromFilePathTest(BaseTestCase):
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 @@ class AuthTest(BaseTestCase):
'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 @@ class AuthTest(BaseTestCase):
'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 @@ class AuthTest(BaseTestCase):
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 @@ class ExitStatusTest(BaseTestCase):
'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 @@ class ExitStatusTest(BaseTestCase):
'--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 @@ class ExitStatusTest(BaseTestCase):
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 @@ class FakeWindowsTest(BaseTestCase):
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 @@ class FakeWindowsTest(BaseTestCase):
'--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 @@ class ItemParsingTest(BaseTestCase):
# 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 @@ class ItemParsingTest(BaseTestCase):
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 @@ class ArgumentParserTestCase(unittest.TestCase):
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 @@ class ArgumentParserTestCase(unittest.TestCase):
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 @@ class ArgumentParserTestCase(unittest.TestCase):
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 @@ class ArgumentParserTestCase(unittest.TestCase):
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 @@ class ArgumentParserTestCase(unittest.TestCase):
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 @@ class ArgumentParserTestCase(unittest.TestCase):
])
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()