Added models.Environment().

Refactoring and general cleanup.
This commit is contained in:
Jakub Roztocil 2012-07-21 02:59:43 +02:00
parent c271715a98
commit 87042f65c9
8 changed files with 566 additions and 350 deletions

View File

@ -1,135 +1,9 @@
#!/usr/bin/env python
import sys
import json
"""
The main entry point. Invoke as `http' or `python -m httpie'.
import requests
from requests.compat import str
from . import httpmessage
from . import cliparse
from . import cli
from . import pretty
TYPE_FORM = 'application/x-www-form-urlencoded; charset=utf-8'
TYPE_JSON = 'application/json; charset=utf-8'
def _get_response(args):
auto_json = args.data and not args.form
if args.json or auto_json:
# JSON
if 'Content-Type' not in args.headers:
args.headers['Content-Type'] = TYPE_JSON
if 'Accept' not in args.headers:
# Default Accept to JSON as well.
args.headers['Accept'] = 'application/json'
if isinstance(args.data, dict):
# If not empty, serialize the data `dict` parsed from arguments.
# Otherwise set it to `None` avoid sending "{}".
args.data = json.dumps(args.data) if args.data else None
elif args.form:
# Form
if not args.files and 'Content-Type' not in args.headers:
# If sending files, `requests` will set
# the `Content-Type` for us.
args.headers['Content-Type'] = TYPE_FORM
# Fire the request.
try:
credentials = None
if args.auth:
auth_type = (requests.auth.HTTPDigestAuth
if args.auth_type == 'digest'
else requests.auth.HTTPBasicAuth)
credentials = auth_type(args.auth.key, args.auth.value)
return requests.request(
method=args.method.lower(),
url=args.url if '://' in args.url else 'http://%s' % args.url,
headers=args.headers,
data=args.data,
verify={'yes': True, 'no': False}.get(args.verify, args.verify),
timeout=args.timeout,
auth=credentials,
proxies=dict((p.key, p.value) for p in args.proxy),
files=args.files,
allow_redirects=args.allow_redirects,
params=args.queries,
)
except (KeyboardInterrupt, SystemExit):
sys.stderr.write('\n')
sys.exit(1)
except Exception as e:
if args.traceback:
raise
sys.stderr.write(str(e.message) + '\n')
sys.exit(1)
def _get_output(args, stdout_isatty, response):
do_prettify = (args.prettify is True or
(args.prettify == cliparse.PRETTIFY_STDOUT_TTY_ONLY
and stdout_isatty))
do_output_request = (cliparse.OUT_REQ_HEADERS in args.output_options
or cliparse.OUT_REQ_BODY in args.output_options)
do_output_response = (cliparse.OUT_RESP_HEADERS in args.output_options
or cliparse.OUT_RESP_BODY in args.output_options)
prettifier = pretty.PrettyHttp(args.style) if do_prettify else None
output = []
if do_output_request:
output.append(httpmessage.format(
message=httpmessage.from_request(response.request),
prettifier=prettifier,
with_headers=cliparse.OUT_REQ_HEADERS in args.output_options,
with_body=cliparse.OUT_REQ_BODY in args.output_options
))
output.append('\n')
if do_output_response:
output.append('\n')
if do_output_response:
output.append(httpmessage.format(
message=httpmessage.from_response(response),
prettifier=prettifier,
with_headers=cliparse.OUT_RESP_HEADERS in args.output_options,
with_body=cliparse.OUT_RESP_BODY in args.output_options
))
output.append('\n')
return ''.join(output)
def main(args=None,
stdin=sys.stdin, stdin_isatty=sys.stdin.isatty(),
stdout=sys.stdout, stdout_isatty=sys.stdout.isatty()):
parser = cli.parser
args = parser.parse_args(
args=args if args is not None else sys.argv[1:],
stdin=stdin,
stdin_isatty=stdin_isatty,
stdout_isatty=stdout_isatty,
)
response = _get_response(args)
output = _get_output(args, stdout_isatty, response)
output_bytes = output.encode('utf8')
f = (stdout.buffer if hasattr(stdout, 'buffer') else stdout)
f.write(output_bytes)
"""
from .core import main
if __name__ == '__main__':

View File

@ -2,10 +2,10 @@
CLI definition.
"""
from . import pretty
from . import __doc__
from . import __version__
from . import cliparse
from .output import AVAILABLE_STYLES
def _(text):
@ -83,9 +83,9 @@ output_options.add_argument('--print', '-p', dest='output_options',
If the output is piped to another program or to a file,
then only the body is printed by default.
'''.format(
request_headers=cliparse.OUT_REQ_HEADERS,
request_headers=cliparse.OUT_REQ_HEAD,
request_body=cliparse.OUT_REQ_BODY,
response_headers=cliparse.OUT_RESP_HEADERS,
response_headers=cliparse.OUT_RESP_HEAD,
response_body=cliparse.OUT_RESP_BODY,
))
)
@ -99,11 +99,11 @@ output_options.add_argument(
)
output_options.add_argument(
'--headers', '-t', dest='output_options',
action='store_const', const=cliparse.OUT_RESP_HEADERS,
action='store_const', const=cliparse.OUT_RESP_HEAD,
help=_('''
Print only the response headers.
Shortcut for --print={0}.
'''.format(cliparse.OUT_RESP_HEADERS))
'''.format(cliparse.OUT_RESP_HEAD))
)
output_options.add_argument(
'--body', '-b', dest='output_options',
@ -116,13 +116,13 @@ output_options.add_argument(
parser.add_argument(
'--style', '-s', dest='style', default='solarized', metavar='STYLE',
choices=pretty.AVAILABLE_STYLES,
choices=AVAILABLE_STYLES,
help=_('''
Output coloring style, one of %s. Defaults to 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).
''') % ', '.join(sorted(pretty.AVAILABLE_STYLES))
''') % ', '.join(sorted(AVAILABLE_STYLES))
)
# ``requests.request`` keyword arguments.

View File

@ -33,13 +33,13 @@ DATA_ITEM_SEPARATORS = [
]
OUT_REQ_HEADERS = 'H'
OUT_REQ_HEAD = 'H'
OUT_REQ_BODY = 'B'
OUT_RESP_HEADERS = 'h'
OUT_RESP_HEAD = 'h'
OUT_RESP_BODY = 'b'
OUTPUT_OPTIONS = [OUT_REQ_HEADERS,
OUTPUT_OPTIONS = [OUT_REQ_HEAD,
OUT_REQ_BODY,
OUT_RESP_HEADERS,
OUT_RESP_HEAD,
OUT_RESP_BODY]
@ -50,20 +50,17 @@ DEFAULT_UA = 'HTTPie/%s' % __version__
class Parser(argparse.ArgumentParser):
def parse_args(self, args=None, namespace=None,
stdin=sys.stdin,
stdin_isatty=sys.stdin.isatty(),
stdout_isatty=sys.stdout.isatty()):
def parse_args(self, env, args=None, namespace=None):
args = super(Parser, self).parse_args(args, namespace)
self._process_output_options(args, stdout_isatty)
self._process_output_options(args, env)
self._validate_auth_options(args)
self._guess_method(args, stdin_isatty)
self._guess_method(args, env)
self._parse_items(args)
if not stdin_isatty:
self._body_from_file(args, stdin)
if not env.stdin_isatty:
self._body_from_file(args, env.stdin)
if args.auth and not args.auth.has_password():
# stdin has already been read (if not a tty) so
@ -78,7 +75,7 @@ class Parser(argparse.ArgumentParser):
'data (key=value) cannot be mixed.')
args.data = f.read()
def _guess_method(self, args, stdin_isatty=sys.stdin.isatty()):
def _guess_method(self, args, env):
"""
Set `args.method`, if not specified, to either POST or GET
based on whether the request has data or not.
@ -87,7 +84,7 @@ class Parser(argparse.ArgumentParser):
if args.method is None:
# Invoked as `http URL'.
assert not args.items
if not stdin_isatty:
if not env.stdin_isatty:
args.method = 'POST'
else:
args.method = 'GET'
@ -112,7 +109,7 @@ class Parser(argparse.ArgumentParser):
args.url = args.method
args.items.insert(0, item)
has_data = not stdin_isatty or any(
has_data = not env.stdin_isatty or any(
item.sep in DATA_ITEM_SEPARATORS for item in args.items)
if has_data:
args.method = 'POST'
@ -162,10 +159,10 @@ class Parser(argparse.ArgumentParser):
content_type = '%s; charset=%s' % (mime, encoding)
args.headers['Content-Type'] = content_type
def _process_output_options(self, args, stdout_isatty):
def _process_output_options(self, args, env):
if not args.output_options:
if stdout_isatty:
args.output_options = OUT_RESP_HEADERS + OUT_RESP_BODY
if env.stdout_isatty:
args.output_options = OUT_RESP_HEAD + OUT_RESP_BODY
else:
args.output_options = OUT_RESP_BODY

124
httpie/core.py Normal file
View File

@ -0,0 +1,124 @@
import sys
import json
import requests
from requests.compat import str
from .models import HTTPMessage, Environment
from .output import OutputProcessor
from . import cliparse
from . import cli
TYPE_FORM = 'application/x-www-form-urlencoded; charset=utf-8'
TYPE_JSON = 'application/json; charset=utf-8'
def get_response(args):
auto_json = args.data and not args.form
if args.json or auto_json:
if 'Content-Type' not in args.headers:
args.headers['Content-Type'] = TYPE_JSON
if 'Accept' not in args.headers:
# Default Accept to JSON as well.
args.headers['Accept'] = 'application/json'
if isinstance(args.data, dict):
# If not empty, serialize the data `dict` parsed from arguments.
# Otherwise set it to `None` avoid sending "{}".
args.data = json.dumps(args.data) if args.data else None
elif args.form:
if not args.files and 'Content-Type' not in args.headers:
# If sending files, `requests` will set
# the `Content-Type` for us.
args.headers['Content-Type'] = TYPE_FORM
try:
credentials = None
if args.auth:
auth_type = (requests.auth.HTTPDigestAuth
if args.auth_type == 'digest'
else requests.auth.HTTPBasicAuth)
credentials = auth_type(args.auth.key, args.auth.value)
return requests.request(
method=args.method.lower(),
url=args.url if '://' in args.url else 'http://%s' % args.url,
headers=args.headers,
data=args.data,
verify={'yes': True, 'no': False}.get(args.verify, args.verify),
timeout=args.timeout,
auth=credentials,
proxies=dict((p.key, p.value) for p in args.proxy),
files=args.files,
allow_redirects=args.allow_redirects,
params=args.queries,
)
except (KeyboardInterrupt, SystemExit):
sys.stderr.write('\n')
sys.exit(1)
except Exception as e:
if args.traceback:
raise
sys.stderr.write(str(e.message) + '\n')
sys.exit(1)
def get_output(args, env, response):
do_prettify = (
args.prettify is True or
(args.prettify == cliparse.PRETTIFY_STDOUT_TTY_ONLY
and env.stdout_isatty)
)
do_output_request = (
cliparse.OUT_REQ_HEAD in args.output_options
or cliparse.OUT_REQ_BODY in args.output_options
)
do_output_response = (
cliparse.OUT_RESP_HEAD in args.output_options
or cliparse.OUT_RESP_BODY in args.output_options
)
prettifier = None
if do_prettify:
prettifier = OutputProcessor(
env, pygments_style=args.style)
output = []
if do_output_request:
req = HTTPMessage.from_request(response.request).format(
prettifier=prettifier,
with_headers=cliparse.OUT_REQ_HEAD in args.output_options,
with_body=cliparse.OUT_REQ_BODY in args.output_options
)
output.append(req)
output.append('\n')
if do_output_response:
output.append('\n')
if do_output_response:
resp = HTTPMessage.from_response(response).format(
prettifier=prettifier,
with_headers=cliparse.OUT_RESP_HEAD in args.output_options,
with_body=cliparse.OUT_RESP_BODY in args.output_options
)
output.append(resp)
output.append('\n')
return ''.join(output)
def main(args=sys.argv[1:], env=Environment()):
parser = cli.parser
args = parser.parse_args(args=args, env=env)
response = get_response(args)
output = get_output(args, env, response)
output_bytes = output.encode('utf8')
f = getattr(env.stdout, 'buffer', env.stdout)
f.write(output_bytes)

View File

@ -1,79 +0,0 @@
from requests.compat import urlparse
class HTTPMessage(object):
"""Model representing an HTTP message."""
def __init__(self, line, headers, body, content_type=None):
# {Request,Status}-Line
self.line = line
self.headers = headers
self.body = body
self.content_type = content_type
def from_request(request):
"""Make an `HTTPMessage` from `requests.models.Request`."""
url = urlparse(request.url)
request_headers = dict(request.headers)
if 'Host' not in request_headers:
request_headers['Host'] = url.netloc
try:
body = request.data
except AttributeError:
# requests < 0.12.1
body = request._enc_data
if isinstance(body, dict):
# --form
body = request.__class__._encode_params(body)
return HTTPMessage(
line='{method} {path}{query} HTTP/1.1'.format(
method=request.method,
path=url.path or '/',
query='' if url.query is '' else '?' + url.query),
headers='\n'.join(str('%s: %s') % (name, value)
for name, value
in request_headers.items()),
body=body,
content_type=request_headers.get('Content-Type')
)
def from_response(response):
"""Make an `HTTPMessage` from `requests.models.Response`."""
encoding = response.encoding or 'ISO-8859-1'
original = response.raw._original_response
response_headers = response.headers
return HTTPMessage(
line='HTTP/{version} {status} {reason}'.format(
version='.'.join(str(original.version)),
status=original.status, reason=original.reason),
headers=str(original.msg),
body=response.content.decode(encoding) if response.content else '',
content_type=response_headers.get('Content-Type'))
def format(message, prettifier=None,
with_headers=True, with_body=True):
"""Return a `unicode` representation of `message`. """
pretty = prettifier is not None
bits = []
if with_headers:
bits.append(message.line)
bits.append(message.headers)
if pretty:
bits = [prettifier.headers('\n'.join(bits))]
if with_body and message.body:
bits.append('\n')
if with_body and message.body:
if pretty and message.content_type:
bits.append(prettifier.body(message.body, message.content_type))
else:
bits.append(message.body)
return '\n'.join(bit.strip() for bit in bits)

103
httpie/models.py Normal file
View File

@ -0,0 +1,103 @@
import os
import sys
from requests.compat import urlparse
class Environment(object):
stdin_isatty = sys.stdin.isatty()
stdin = sys.stdin
stdout_isatty = sys.stdout.isatty()
stdout = sys.stdout
# Can be set to 0 to disable colors completely.
colors = 256 if '256color' in os.environ.get('TERM', '') else 88
def __init__(self, **kwargs):
self.__dict__.update(**kwargs)
class HTTPMessage(object):
"""Model representing an HTTP message."""
def __init__(self, line, headers, body, content_type=None):
# {Request,Status}-Line
self.line = line
self.headers = headers
self.body = body
self.content_type = content_type
def format(self, prettifier=None, with_headers=True, with_body=True):
"""Return a `unicode` representation of `self`. """
pretty = prettifier is not None
bits = []
if with_headers:
bits.append(self.line)
bits.append(self.headers)
if pretty:
bits = [
prettifier.process_headers('\n'.join(bits))
]
if with_body and self.body:
bits.append('\n')
if with_body and self.body:
if pretty and self.content_type:
bits.append(prettifier.process_body(
self.body, self.content_type))
else:
bits.append(self.body)
return '\n'.join(bit.strip() for bit in bits)
@staticmethod
def from_request(request):
"""Make an `HTTPMessage` from `requests.models.Request`."""
url = urlparse(request.url)
request_headers = dict(request.headers)
if 'Host' not in request_headers:
request_headers['Host'] = url.netloc
try:
body = request.data
except AttributeError:
# requests < 0.12.1
body = request._enc_data
if isinstance(body, dict):
# --form
body = request.__class__._encode_params(body)
request_line = '{method} {path}{query} HTTP/1.1'.format(
method=request.method,
path=url.path or '/',
query='' if url.query is '' else '?' + url.query
)
headers = '\n'.join(
str('%s: %s') % (name, value)
for name, value
in request_headers.items()
)
return HTTPMessage(
line=request_line,
headers=headers,
body=body,
content_type=request_headers.get('Content-Type')
)
@classmethod
def from_response(cls, response):
"""Make an `HTTPMessage` from `requests.models.Response`."""
encoding = response.encoding or 'ISO-8859-1'
original = response.raw._original_response
response_headers = response.headers
status_line = 'HTTP/{version} {status} {reason}'.format(
version='.'.join(str(original.version)),
status=original.status,
reason=original.reason
)
body = response.content.decode(encoding) if response.content else ''
return cls(
line=status_line,
headers=str(original.msg),
body=body,
content_type=response_headers.get('Content-Type'))

View File

@ -22,13 +22,6 @@ if is_windows:
import colorama
colorama.init()
# 256 looks better on Windows
formatter_class = Terminal256Formatter
else:
formatter_class = (
Terminal256Formatter
if '256color' in os.environ.get('TERM', '')
else TerminalFormatter
)
class HTTPLexer(lexer.RegexLexer):
@ -82,23 +75,96 @@ class HTTPLexer(lexer.RegexLexer):
]}
class PrettyHttp(object):
"""HTTP headers & body prettyfier."""
class BaseProcessor(object):
enabled = True
def __init__(self, env, **kwargs):
self.env = env
self.kwargs = kwargs
def process_headers(self, headers):
return headers
def process_body(self, content, content_type):
return content
class JSONProcessor(BaseProcessor):
def process_body(self, content, content_type):
if content_type == 'application/json':
try:
# Indent and sort the JSON data.
content = json.dumps(
json.loads(content),
sort_keys=True,
ensure_ascii=False,
indent=4,
)
except ValueError:
# Invalid JSON - we don't care.
pass
return content
class PygmentsProcessor(BaseProcessor):
def __init__(self, *args, **kwargs):
super(PygmentsProcessor, self).__init__(*args, **kwargs)
if not self.env.colors:
self.enabled = False
return
def __init__(self, style_name):
try:
style = get_style_by_name(style_name)
style = get_style_by_name(
self.kwargs.get('pygments_style', DEFAULT_STYLE))
except ClassNotFound:
style = solarized.SolarizedStyle
self.formatter = formatter_class(style=style)
def headers(self, content):
"""Pygmentize HTTP headers."""
return pygments.highlight(content, HTTPLexer(), self.formatter)
if is_windows or self.env.colors == 256:
fmt_class = Terminal256Formatter
else:
fmt_class = TerminalFormatter
self.formatter = fmt_class(style=style)
def body(self, content, content_type):
"""Pygmentize `content` based on `content_type`."""
def process_headers(self, headers):
return pygments.highlight(
headers, HTTPLexer(), self.formatter)
def process_body(self, content, content_type):
try:
lexer = get_lexer_for_mimetype(content_type)
except ClassNotFound:
pass
else:
content = pygments.highlight(content, lexer, self.formatter)
return content
class OutputProcessor(object):
"""."""
installed_processors = [
JSONProcessor,
PygmentsProcessor
]
def __init__(self, env, **kwargs):
self.env = env
processors = [
cls(env, **kwargs)
for cls in self.installed_processors
]
self.processors = [p for p in processors if p.enabled]
def process_headers(self, headers):
for processor in self.processors:
headers = processor.process_headers(headers)
return headers
def process_body(self, content, content_type):
content_type = content_type.split(';')[0]
application_match = re.match(
@ -110,19 +176,7 @@ class PrettyHttp(object):
vendor, extension = application_match.groups()
content_type = content_type.replace(vendor, '')
try:
lexer = get_lexer_for_mimetype(content_type)
except ClassNotFound:
return content
for processor in self.processors:
content = processor.process_body(content, content_type)
if content_type == 'application/json':
try:
# Indent and sort the JSON data.
content = json.dumps(json.loads(content),
sort_keys=True, indent=4,
ensure_ascii=False)
except ValueError:
# Invalid JSON - we don't care.
pass
return pygments.highlight(content, lexer, self.formatter)
return content

View File

@ -1,4 +1,18 @@
# -*- coding: utf-8 -*-
""" -*- coding: utf-8 -*-
Many of the test cases here use httpbin.org.
To make it run faster and offline you can::
# Install `httpbin` locally
pip install httpbin
# Run it
httpbin
# Run the tests against it
HTTPBIN_URL=http://localhost:5000 python setup.py test
"""
import unittest
import argparse
import os
@ -17,38 +31,43 @@ from requests import Response
TESTS_ROOT = os.path.dirname(__file__)
sys.path.insert(0, os.path.realpath(os.path.join(TESTS_ROOT, '..')))
from httpie import __main__, cliparse
from httpie import cliparse
from httpie.models import Environment
from httpie.core import main, get_output
HTTPBIN_URL = os.environ.get('HTTPBIN_URL',
'http://httpbin.org')
TEST_FILE_PATH = os.path.join(TESTS_ROOT, 'file.txt')
TEST_FILE2_PATH = os.path.join(TESTS_ROOT, 'file2.txt')
TEST_FILE_CONTENT = open(TEST_FILE_PATH).read().strip()
TERMINAL_COLOR_PRESENCE_CHECK = '\x1b['
def httpbin(path):
return HTTPBIN_URL + path
def http(*args, **kwargs):
"""
Invoke `httpie.__main__.main` with `args` and `kwargs`,
and return a unicode response.
"""
http_kwargs = {
'stdin_isatty': True,
'stdout_isatty': True
}
http_kwargs.update(kwargs)
command_line = ' '.join(args)
if ('stdout_isatty' not in kwargs
and '--pretty' not in command_line
and '--ugly' not in command_line):
# Make ugly default for testing purposes unless we're
# being explicit about it. It's so that we can test for
# strings in the response without having to always pass --ugly.
args = ['--ugly'] + list(args)
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,
)
stdout = http_kwargs.setdefault('stdout', tempfile.TemporaryFile())
__main__.main(args=args, **http_kwargs)
stdout = kwargs['env'].stdout = tempfile.TemporaryFile()
main(args=args, **kwargs)
stdout.seek(0)
response = stdout.read().decode('utf8')
stdout.close()
@ -70,42 +89,77 @@ class BaseTestCase(unittest.TestCase):
#################################################################
# High-level tests using httpbin.org.
# High-level tests using httpbin.
#################################################################
class HTTPieTest(BaseTestCase):
def test_GET(self):
r = http('GET', 'http://httpbin.org/get')
r = http(
'GET',
httpbin('/get')
)
self.assertIn('HTTP/1.1 200', r)
def test_DELETE(self):
r = http('DELETE', 'http://httpbin.org/delete')
r = http(
'DELETE',
httpbin('/delete')
)
self.assertIn('HTTP/1.1 200', r)
def test_PUT(self):
r = http('PUT', 'http://httpbin.org/put', 'foo=bar')
r = http(
'PUT',
httpbin('/put'),
'foo=bar'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"foo": "bar"', r)
def test_POST_JSON_data(self):
r = http('POST', 'http://httpbin.org/post', 'foo=bar')
r = http(
'POST',
httpbin('/post'),
'foo=bar'
)
print r
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"foo": "bar"', r)
def test_POST_form(self):
r = http('--form', 'POST', 'http://httpbin.org/post', 'foo=bar')
r = http(
'--form',
'POST',
httpbin('/post'),
'foo=bar'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"foo": "bar"', r)
def test_POST_stdin(self):
r = http('--form', 'POST', 'http://httpbin.org/post',
stdin=open(TEST_FILE_PATH), stdin_isatty=False)
env = Environment(
stdin=open(TEST_FILE_PATH),
stdin_isatty=False,
colors=0,
)
r = http(
'--form',
'POST',
httpbin('/post'),
env=env
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn(TEST_FILE_CONTENT, r)
def test_headers(self):
r = http('GET', 'http://httpbin.org/headers', 'Foo:bar')
r = http(
'GET',
httpbin('/headers'),
'Foo:bar'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"User-Agent": "HTTPie', r)
self.assertIn('"Foo": "bar"', r)
@ -120,51 +174,73 @@ class AutoContentTypeAndAcceptHeadersTest(BaseTestCase):
"""
def test_GET_no_data_no_auto_headers(self):
# https://github.com/jkbr/httpie/issues/62
r = http('GET', 'http://httpbin.org/headers')
r = http(
'GET',
httpbin('/headers')
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"Accept": "*/*"', r)
# Although an empty header is present in the response from httpbin,
# it's not included in the request.
self.assertIn('"Content-Type": ""', r)
self.assertNotIn('"Content-Type": "application/json', r)
def test_POST_no_data_no_auto_headers(self):
# JSON headers shouldn't be automatically set for POST with no data.
r = http('POST', 'http://httpbin.org/post')
r = http(
'POST',
httpbin('/post')
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"Accept": "*/*"', r)
# Although an empty header is present in the response from httpbin,
# it's not included in the request.
self.assertIn(' "Content-Type": ""', r)
self.assertNotIn('"Content-Type": "application/json', r)
def test_POST_with_data_auto_JSON_headers(self):
r = http('POST', 'http://httpbin.org/post', 'a=b')
r = http(
'POST',
httpbin('/post'),
'a=b'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"Accept": "application/json"', r)
self.assertIn('"Content-Type": "application/json; charset=utf-8', r)
def test_GET_with_data_auto_JSON_headers(self):
# JSON headers should automatically be set also for GET with data.
r = http('POST', 'http://httpbin.org/post', 'a=b')
r = http(
'POST',
httpbin('/post'),
'a=b'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"Accept": "application/json"', r)
self.assertIn('"Content-Type": "application/json; charset=utf-8', r)
def test_POST_explicit_JSON_auto_JSON_headers(self):
r = http('-j', 'POST', 'http://httpbin.org/post')
r = http(
'--json',
'POST',
httpbin('/post')
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"Accept": "application/json"', r)
self.assertIn('"Content-Type": "application/json; charset=utf-8', r)
def test_GET_explicit_JSON_explicit_headers(self):
r = http('-j', 'GET', 'http://httpbin.org/headers',
'Accept:application/xml',
'Content-Type:application/xml')
r = http(
'--json',
'GET',
httpbin('/headers'),
'Accept:application/xml',
'Content-Type:application/xml'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"Accept": "application/xml"', r)
self.assertIn('"Content-Type": "application/xml"', r)
def test_POST_form_auto_Content_Type(self):
r = http('-f', 'POST', 'http://httpbin.org/post')
r = http(
'--form',
'POST',
httpbin('/post')
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn(
'"Content-Type":'
@ -173,44 +249,77 @@ class AutoContentTypeAndAcceptHeadersTest(BaseTestCase):
)
def test_POST_form_Content_Type_override(self):
r = http('-f', 'POST', 'http://httpbin.org/post',
'Content-Type:application/xml')
r = http(
'--form',
'POST',
httpbin('/post'),
'Content-Type:application/xml'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"Content-Type": "application/xml"', r)
def test_print_only_body_when_stdout_redirected_by_default(self):
r = http('GET', 'httpbin.org/get', stdout_isatty=False)
r = http(
'GET',
'httpbin.org/get',
env=Environment(stdout_isatty=False)
)
self.assertNotIn('HTTP/', r)
def test_print_overridable_when_stdout_redirected(self):
r = http('--print=h', 'GET', 'httpbin.org/get', stdout_isatty=False)
r = http(
'--print=h',
'GET',
'httpbin.org/get',
env=Environment(stdout_isatty=False)
)
self.assertIn('HTTP/1.1 200', r)
class ImplicitHTTPMethodTest(BaseTestCase):
def test_implicit_GET(self):
r = http('http://httpbin.org/get')
r = http(httpbin('/get'))
self.assertIn('HTTP/1.1 200', r)
def test_implicit_GET_with_headers(self):
r = http('http://httpbin.org/headers', 'Foo:bar')
r = http(
httpbin('/headers'),
'Foo:bar'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"Foo": "bar"', r)
def test_implicit_POST_json(self):
r = http('http://httpbin.org/post', 'hello=world')
r = http(
httpbin('/post'),
'hello=world'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"hello": "world"', r)
def test_implicit_POST_form(self):
r = http('--form', 'http://httpbin.org/post', 'foo=bar')
r = http(
'--form',
httpbin('/post'),
'foo=bar'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"foo": "bar"', r)
def test_implicit_POST_stdin(self):
r = http('--form', 'http://httpbin.org/post',
stdin=open(TEST_FILE_PATH), stdin_isatty=False)
env = Environment(
stdin_isatty=False,
stdin=open(TEST_FILE_PATH),
colors=0,
)
r = http(
'--form',
httpbin('/post'),
env=env
)
self.assertIn('HTTP/1.1 200', r)
@ -218,19 +327,26 @@ class PrettyFlagTest(BaseTestCase):
"""Test the --pretty / --ugly flag handling."""
def test_pretty_enabled_by_default(self):
r = http('GET', 'http://httpbin.org/get', stdout_isatty=True)
r = http(
'GET',
httpbin('/get'),
env=Environment(stdout_isatty=True),
)
self.assertIn(TERMINAL_COLOR_PRESENCE_CHECK, r)
def test_pretty_enabled_by_default_unless_stdout_redirected(self):
r = http('GET', 'http://httpbin.org/get', stdout_isatty=False)
r = http(
'GET',
httpbin('/get')
)
self.assertNotIn(TERMINAL_COLOR_PRESENCE_CHECK, r)
def test_force_pretty(self):
r = http(
'--pretty',
'GET',
'http://httpbin.org/get',
stdout_isatty=False
httpbin('/get'),
env=Environment(stdout_isatty=False),
)
self.assertIn(TERMINAL_COLOR_PRESENCE_CHECK, r)
@ -238,8 +354,7 @@ class PrettyFlagTest(BaseTestCase):
r = http(
'--ugly',
'GET',
'http://httpbin.org/get',
stdout_isatty=True
httpbin('/get'),
)
self.assertNotIn(TERMINAL_COLOR_PRESENCE_CHECK, r)
@ -250,7 +365,7 @@ class VerboseFlagTest(BaseTestCase):
r = http(
'--verbose',
'GET',
'http://httpbin.org/get',
httpbin('/get'),
'test-header:__test__'
)
self.assertIn('HTTP/1.1 200', r)
@ -262,7 +377,7 @@ class VerboseFlagTest(BaseTestCase):
'--verbose',
'--form',
'POST',
'http://httpbin.org/post',
httpbin('/post'),
'foo=bar',
'baz=bar'
)
@ -274,13 +389,21 @@ class MultipartFormDataFileUploadTest(BaseTestCase):
def test_non_existent_file_raises_parse_error(self):
self.assertRaises(cliparse.ParseError, http,
'--form', '--traceback',
'POST', 'http://httpbin.org/post',
'foo@/__does_not_exist__')
'--form',
'--traceback',
'POST',
httpbin('/post'),
'foo@/__does_not_exist__'
)
def test_upload_ok(self):
r = http('--form', 'POST', 'http://httpbin.org/post',
'test-file@%s' % TEST_FILE_PATH, 'foo=bar')
r = http(
'--form',
'POST',
httpbin('/post'),
'test-file@%s' % TEST_FILE_PATH,
'foo=bar'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"test-file": "%s' % TEST_FILE_CONTENT, r)
self.assertIn('"foo": "bar"', r)
@ -292,7 +415,11 @@ class RequestBodyFromFilePathTest(BaseTestCase):
"""
def test_request_body_from_file_by_path(self):
r = http('POST', 'http://httpbin.org/post', '@' + TEST_FILE_PATH)
r = http(
'POST',
httpbin('/post'),
'@' + TEST_FILE_PATH
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn(TEST_FILE_CONTENT, r)
self.assertIn('"Content-Type": "text/plain"', r)
@ -300,7 +427,7 @@ class RequestBodyFromFilePathTest(BaseTestCase):
def test_request_body_from_file_by_path_with_explicit_content_type(self):
r = http(
'POST',
'http://httpbin.org/post',
httpbin('/post'),
'@' + TEST_FILE_PATH,
'Content-Type:x-foo/bar'
)
@ -311,39 +438,55 @@ class RequestBodyFromFilePathTest(BaseTestCase):
def test_request_body_from_file_by_path_only_one_file_allowed(self):
self.assertRaises(SystemExit, lambda: http(
'POST',
'http://httpbin.org/post',
'@' + TEST_FILE_PATH,
'@' + TEST_FILE2_PATH))
httpbin('/post'),
'@' + TEST_FILE_PATH,
'@' + TEST_FILE2_PATH)
)
def test_request_body_from_file_by_path_no_data_items_allowed(self):
self.assertRaises(SystemExit, lambda: http(
'POST',
'http://httpbin.org/post',
'@' + TEST_FILE_PATH,
'foo=bar'))
httpbin('/post'),
'@' + TEST_FILE_PATH,
'foo=bar')
)
class AuthTest(BaseTestCase):
def test_basic_auth(self):
r = http('--auth', 'user:password',
'GET', 'httpbin.org/basic-auth/user/password')
r = http(
'--auth',
'user:password',
'GET',
'httpbin.org/basic-auth/user/password'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"authenticated": true', r)
self.assertIn('"user": "user"', r)
def test_digest_auth(self):
r = http('--auth-type=digest', '--auth', 'user:password',
'GET', 'httpbin.org/digest-auth/auth/user/password')
r = http(
'--auth-type=digest',
'--auth',
'user:password',
'GET',
'httpbin.org/digest-auth/auth/user/password'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"authenticated": true', r)
self.assertIn('"user": "user"', r)
def test_password_prompt(self):
cliparse.AuthCredentials._getpass = lambda self, prompt: 'password'
r = http('--auth', 'user',
'GET', 'httpbin.org/basic-auth/user/password')
r = http(
'--auth',
'user',
'GET',
'httpbin.org/basic-auth/user/password'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"authenticated": true', r)
@ -446,7 +589,7 @@ class ArgumentParserTestCase(unittest.TestCase):
args.url = 'http://example.com/'
args.items = []
self.parser._guess_method(args)
self.parser._guess_method(args, Environment())
self.assertEquals(args.method, 'GET')
self.assertEquals(args.url, 'http://example.com/')
@ -458,7 +601,7 @@ class ArgumentParserTestCase(unittest.TestCase):
args.url = 'http://example.com/'
args.items = []
self.parser._guess_method(args)
self.parser._guess_method(args, Environment())
self.assertEquals(args.method, 'GET')
self.assertEquals(args.url, 'http://example.com/')
@ -470,7 +613,7 @@ class ArgumentParserTestCase(unittest.TestCase):
args.url = 'data=field'
args.items = []
self.parser._guess_method(args)
self.parser._guess_method(args, Environment())
self.assertEquals(args.method, 'POST')
self.assertEquals(args.url, 'http://example.com/')
@ -485,7 +628,7 @@ class ArgumentParserTestCase(unittest.TestCase):
args.url = 'test:header'
args.items = []
self.parser._guess_method(args)
self.parser._guess_method(args, Environment())
self.assertEquals(args.method, 'GET')
self.assertEquals(args.url, 'http://example.com/')
@ -503,7 +646,7 @@ class ArgumentParserTestCase(unittest.TestCase):
key='old_item', value='b', sep='=', orig='old_item=b')
]
self.parser._guess_method(args)
self.parser._guess_method(args, Environment())
self.assertEquals(args.items, [
cliparse.KeyValue(
@ -557,7 +700,7 @@ class UnicodeOutputTestCase(BaseTestCase):
args.style = 'default'
# colorized output contains escape sequences
output = __main__._get_output(args, True, response)
output = get_output(args, Environment(), response)
for key, value in response_dict.items():
self.assertIn(key, output)