Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
tree: 1efea59a8d
Fetching contributors…

Cannot retrieve contributors at this time

executable file 949 lines (781 sloc) 26.59 kb
#!/usr/bin/env python
# coding=utf8
"""
Many of the test cases here use httpbin.org.
To make it run faster and offline you can::
# Install `httpbin` locally
pip install git+https://github.com/kennethreitz/httpbin.git
# Run it
httpbin
# Run the tests against it
HTTPBIN_URL=http://localhost:5000 python setup.py test
# Test all Python environments
HTTPBIN_URL=http://localhost:5000 tox
"""
import os
import sys
import json
import tempfile
import unittest
import argparse
try:
from urllib.request import urlopen
except ImportError:
from urllib2 import urlopen
import requests
from requests.compat import is_py26, is_py3, bytes, str
#################################################################
# Utils/setup
#################################################################
# HACK: Prepend ../ to PYTHONPATH so that we can import httpie form there.
TESTS_ROOT = 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, get_output
from httpie.output import BINARY_SUPPRESSED_NOTICE
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
class BytesResponse(bytes):
exit_status = None
stderr = None
json = None
def __eq__(self, other):
return super(BytesResponse, self).__eq__(other)
class StrResponse(str):
exit_status = None
stderr = None
json = None
def __eq__(self, other):
return super(StrResponse, self).__eq__(other)
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,
)
stdout = kwargs['env'].stdout = tempfile.TemporaryFile()
stderr = kwargs['env'].stderr = tempfile.TemporaryFile()
exit_status = main(args=['--debug'] + list(args), **kwargs)
stdout.seek(0)
stderr.seek(0)
output = stdout.read()
try:
r = StrResponse(output.decode('utf8'))
except UnicodeDecodeError:
r = BytesResponse(output)
else:
if TERMINAL_COLOR_PRESENCE_CHECK not in r:
# De-serialize JSON body if possible.
if r.strip().startswith('{'):
#noinspection PyTypeChecker
r.json = json.loads(r)
elif r.count('Content-Type:') == 1 and 'application/json' in r:
try:
j = r.strip()[r.strip().rindex('\n\n'):]
except ValueError:
pass
else:
try:
r.json = json.loads(j)
except ValueError:
pass
r.stderr = stderr.read().decode('utf8')
r.exit_status = exit_status
stdout.close()
stderr.close()
return r
class BaseTestCase(unittest.TestCase):
if is_py26:
def assertIn(self, member, container, msg=None):
self.assert_(member in container, msg)
def assertNotIn(self, member, container, msg=None):
self.assert_(member not in container, msg)
def assertDictEqual(self, d1, d2, msg=None):
self.assertEqual(set(d1.keys()), set(d2.keys()), msg)
self.assertEqual(sorted(d1.values()), sorted(d2.values()), msg)
#################################################################
# High-level tests using httpbin.
#################################################################
class HTTPieTest(BaseTestCase):
def test_GET(self):
r = http(
'GET',
httpbin('/get')
)
self.assertIn('HTTP/1.1 200', r)
def test_DELETE(self):
r = http(
'DELETE',
httpbin('/delete')
)
self.assertIn('HTTP/1.1 200', r)
def test_PUT(self):
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',
httpbin('/post'),
'foo=bar'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"foo": "bar"', r)
def test_POST_form(self):
r = http(
'--form',
'POST',
httpbin('/post'),
'foo=bar'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"foo": "bar"', r)
def test_POST_form_multiple_values(self):
r = http(
'--form',
'POST',
httpbin('/post'),
'foo=bar',
'foo=baz',
)
self.assertIn('HTTP/1.1 200', r)
self.assertDictEqual(r.json['form'], {
'foo': ['bar', 'baz']
})
def test_POST_stdin(self):
env = Environment(
stdin=open(TEST_FILE_PATH),
stdin_isatty=False,
stdout_isatty=True,
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',
httpbin('/headers'),
'Foo:bar'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"User-Agent": "HTTPie', r)
self.assertIn('"Foo": "bar"', r)
class QuerystringTest(BaseTestCase):
def test_query_string_params_in_url(self):
r = http(
'--print=Hhb',
'GET',
httpbin('/get?a=1&b=2')
)
path = '/get?a=1&b=2'
url = httpbin(path)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('GET %s HTTP/1.1' % path, r)
self.assertIn('"url": "%s"' % url, r)
def test_query_string_params_items(self):
r = http(
'--print=Hhb',
'GET',
httpbin('/get'),
'a==1',
'b==2'
)
path = '/get?a=1&b=2'
url = httpbin(path)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('GET %s HTTP/1.1' % path, r)
self.assertIn('"url": "%s"' % url, r)
def test_query_string_params_in_url_and_items_with_duplicates(self):
r = http(
'--print=Hhb',
'GET',
httpbin('/get?a=1&a=1'),
'a==1',
'a==1',
'b==2',
)
path = '/get?a=1&a=1&a=1&a=1&b=2'
url = httpbin(path)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('GET %s HTTP/1.1' % path, r)
self.assertIn('"url": "%s"' % url, r)
class AutoContentTypeAndAcceptHeadersTest(BaseTestCase):
"""
Test that Accept and Content-Type correctly defaults to JSON,
but can still be overridden. The same with Content-Type when --form
-f is used.
"""
def test_GET_no_data_no_auto_headers(self):
# https://github.com/jkbr/httpie/issues/62
r = http(
'GET',
httpbin('/headers')
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"Accept": "*/*"', 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',
httpbin('/post')
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"Accept": "*/*"', r)
self.assertNotIn('"Content-Type": "application/json', r)
def test_POST_with_data_auto_JSON_headers(self):
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',
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(
'--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(
'--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(
'--form',
'POST',
httpbin('/post')
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn(
'"Content-Type":'
' "application/x-www-form-urlencoded; charset=utf-8"',
r
)
def test_POST_form_Content_Type_override(self):
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('/get'),
env=Environment(
stdin_isatty=True,
stdout_isatty=False
)
)
self.assertNotIn('HTTP/', r)
def test_print_overridable_when_stdout_redirected(self):
r = http(
'--print=h',
'GET',
httpbin('/get'),
env=Environment(
stdin_isatty=True,
stdout_isatty=False
)
)
self.assertIn('HTTP/1.1 200', r)
class ImplicitHTTPMethodTest(BaseTestCase):
def test_implicit_GET(self):
r = http(httpbin('/get'))
self.assertIn('HTTP/1.1 200', r)
def test_implicit_GET_with_headers(self):
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(
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',
httpbin('/post'),
'foo=bar'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"foo": "bar"', r)
def test_implicit_POST_stdin(self):
env = Environment(
stdin_isatty=False,
stdin=open(TEST_FILE_PATH),
stdout_isatty=True,
colors=0,
)
r = http(
'--form',
httpbin('/post'),
env=env
)
self.assertIn('HTTP/1.1 200', r)
class PrettyFlagTest(BaseTestCase):
"""Test the --pretty / --ugly flag handling."""
def test_pretty_enabled_by_default(self):
r = http(
'GET',
httpbin('/get'),
env=Environment(
stdin_isatty=True,
stdout_isatty=True,
),
)
self.assertIn(TERMINAL_COLOR_PRESENCE_CHECK, r)
def test_pretty_enabled_by_default_unless_stdout_redirected(self):
r = http(
'GET',
httpbin('/get')
)
self.assertNotIn(TERMINAL_COLOR_PRESENCE_CHECK, r)
def test_force_pretty(self):
r = http(
'--pretty',
'GET',
httpbin('/get'),
env=Environment(
stdin_isatty=True,
stdout_isatty=False
),
)
self.assertIn(TERMINAL_COLOR_PRESENCE_CHECK, r)
def test_force_ugly(self):
r = http(
'--ugly',
'GET',
httpbin('/get'),
)
self.assertNotIn(TERMINAL_COLOR_PRESENCE_CHECK, r)
class VerboseFlagTest(BaseTestCase):
def test_verbose(self):
r = http(
'--verbose',
'GET',
httpbin('/get'),
'test-header:__test__'
)
self.assertIn('HTTP/1.1 200', r)
self.assertEqual(r.count('__test__'), 2)
def test_verbose_form(self):
# https://github.com/jkbr/httpie/issues/53
r = http(
'--verbose',
'--form',
'POST',
httpbin('/post'),
'foo=bar',
'baz=bar'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('foo=bar&baz=bar', r)
class MultipartFormDataFileUploadTest(BaseTestCase):
def test_non_existent_file_raises_parse_error(self):
self.assertRaises(SystemExit, http,
'--form',
'--traceback',
'POST',
httpbin('/post'),
'foo@/__does_not_exist__'
)
def test_upload_ok(self):
r = http(
'--form',
'--verbose',
'POST',
httpbin('/post'),
'test-file@%s' % TEST_FILE_PATH,
'foo=bar'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('Content-Disposition: form-data; name="foo"', r)
self.assertIn('Content-Disposition: form-data; name="test-file";'
' filename="test-file"', r)
self.assertEqual(r.count(TEST_FILE_CONTENT), 2)
self.assertIn('"foo": "bar"', r)
class TestBinaryResponses(BaseTestCase):
url = 'http://www.google.com/favicon.ico'
@property
def bindata(self):
if not hasattr(self, '_bindata'):
self._bindata = urlopen(self.url).read()
return self._bindata
def test_binary_suppresses_when_terminal(self):
r = http(
'GET',
self.url
)
self.assertIn(BINARY_SUPPRESSED_NOTICE, r)
def test_binary_suppresses_when_not_terminal_but_pretty(self):
r = http(
'--pretty',
'GET',
self.url,
env=Environment(stdin_isatty=True,
stdout_isatty=False)
)
self.assertIn(BINARY_SUPPRESSED_NOTICE, r)
def test_binary_included_and_correct_when_suitable(self):
r = http(
'GET',
self.url,
env=Environment(stdin_isatty=True,
stdout_isatty=False)
)
self.assertEqual(r, self.bindata)
class RequestBodyFromFilePathTest(BaseTestCase):
"""
`http URL @file'
"""
def test_request_body_from_file_by_path(self):
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)
def test_request_body_from_file_by_path_with_explicit_content_type(self):
r = http(
'POST',
httpbin('/post'),
'@' + TEST_FILE_PATH,
'Content-Type:x-foo/bar'
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn(TEST_FILE_CONTENT, r)
self.assertIn('"Content-Type": "x-foo/bar"', r)
def test_request_body_from_file_by_path_only_one_file_allowed(self):
self.assertRaises(SystemExit, lambda: http(
'POST',
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',
httpbin('/post'),
'@' + TEST_FILE_PATH,
'foo=bar')
)
class AuthTest(BaseTestCase):
def test_basic_auth(self):
r = http(
'--auth=user:password',
'GET',
httpbin('/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('/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):
input.AuthCredentials._getpass = lambda self, prompt: 'password'
r = http(
'--auth',
'user',
'GET',
httpbin('/basic-auth/user/password')
)
self.assertIn('HTTP/1.1 200', r)
self.assertIn('"authenticated": true', r)
self.assertIn('"user": "user"', r)
class ExitStatusTest(BaseTestCase):
def test_ok_response_exits_0(self):
r = http(
'GET',
httpbin('/status/200')
)
self.assertIn('HTTP/1.1 200', r)
self.assertEqual(r.exit_status, 0)
def test_error_response_exits_0_without_check_status(self):
r = http(
'GET',
httpbin('/status/500')
)
self.assertIn('HTTP/1.1 500', r)
self.assertEqual(r.exit_status, 0)
def test_3xx_check_status_exits_3_and_stderr_when_stdout_redirected(self):
r = http(
'--check-status',
'--headers', # non-terminal, force headers
'GET',
httpbin('/status/301'),
env=Environment(
stdout_isatty=False,
stdin_isatty=True,
)
)
self.assertIn('HTTP/1.1 301', r)
self.assertEqual(r.exit_status, 3)
self.assertIn('301 moved permanently', r.stderr.lower())
def test_3xx_check_status_redirects_allowed_exits_0(self):
r = http(
'--check-status',
'--allow-redirects',
'GET',
httpbin('/status/301')
)
# The redirect will be followed so 200 is expected.
self.assertIn('HTTP/1.1 200 OK', r)
self.assertEqual(r.exit_status, 0)
def test_4xx_check_status_exits_4(self):
r = http(
'--check-status',
'GET',
httpbin('/status/401')
)
self.assertIn('HTTP/1.1 401', r)
self.assertEqual(r.exit_status, 4)
# Also stderr should be empty since stdout isn't redirected.
self.assert_(not r.stderr)
def test_5xx_check_status_exits_5(self):
r = http(
'--check-status',
'GET',
httpbin('/status/500')
)
self.assertIn('HTTP/1.1 500', r)
self.assertEqual(r.exit_status, 5)
#################################################################
# CLI argument parsing related tests.
#################################################################
class ItemParsingTest(BaseTestCase):
def setUp(self):
self.key_value_type = input.KeyValueArgType(
input.SEP_HEADERS,
input.SEP_QUERY,
input.SEP_DATA,
input.SEP_DATA_RAW_JSON,
input.SEP_FILES,
)
def test_invalid_items(self):
items = ['no-separator']
for item in items:
self.assertRaises(argparse.ArgumentTypeError,
lambda: self.key_value_type(item))
def test_escape(self):
headers, data, files, params = input.parse_items([
# headers
self.key_value_type('foo\\:bar:baz'),
self.key_value_type('jack\\@jill:hill'),
# data
self.key_value_type('baz\\=bar=foo'),
# files
self.key_value_type('bar\\@baz@%s' % TEST_FILE_PATH)
])
self.assertDictEqual(headers, {
'foo:bar': 'baz',
'jack@jill': 'hill',
})
self.assertDictEqual(data, {
'baz=bar': 'foo',
})
self.assertIn('bar@baz', files)
def test_escape_longsep(self):
headers, data, files, params = input.parse_items([
self.key_value_type('bob\\:==foo'),
])
self.assertDictEqual(params, {
'bob:': 'foo',
})
def test_valid_items(self):
headers, data, files, params = input.parse_items([
self.key_value_type('string=value'),
self.key_value_type('header:value'),
self.key_value_type('list:=["a", 1, {}, false]'),
self.key_value_type('obj:={"a": "b"}'),
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('query==value'),
])
self.assertDictEqual(headers, {
'header': 'value',
'eh': ''
})
self.assertDictEqual(data, {
"ed": "",
"string": "value",
"bool": True,
"list": ["a", 1, {}, False],
"obj": {"a": "b"},
})
self.assertDictEqual(params, {
'query': 'value',
})
self.assertIn('test-file', files)
class ArgumentParserTestCase(unittest.TestCase):
def setUp(self):
self.parser = input.Parser()
def test_guess_when_method_set_and_valid(self):
args = argparse.Namespace()
args.method = 'GET'
args.url = 'http://example.com/'
args.items = []
self.parser._guess_method(args, Environment())
self.assertEqual(args.method, 'GET')
self.assertEqual(args.url, 'http://example.com/')
self.assertEqual(args.items, [])
def test_guess_when_method_not_set(self):
args = argparse.Namespace()
args.method = None
args.url = 'http://example.com/'
args.items = []
self.parser._guess_method(args, Environment(
stdin_isatty=True,
stdout_isatty=True,
))
self.assertEqual(args.method, 'GET')
self.assertEqual(args.url, 'http://example.com/')
self.assertEqual(args.items, [])
def test_guess_when_method_set_but_invalid_and_data_field(self):
args = argparse.Namespace()
args.method = 'http://example.com/'
args.url = 'data=field'
args.items = []
self.parser._guess_method(args, Environment())
self.assertEqual(args.method, 'POST')
self.assertEqual(args.url, 'http://example.com/')
self.assertEqual(
args.items,
[input.KeyValue(
key='data', value='field', sep='=', orig='data=field')])
def test_guess_when_method_set_but_invalid_and_header_field(self):
args = argparse.Namespace()
args.method = 'http://example.com/'
args.url = 'test:header'
args.items = []
self.parser._guess_method(args, Environment(
stdin_isatty=True,
stdout_isatty=True,
))
self.assertEqual(args.method, 'GET')
self.assertEqual(args.url, 'http://example.com/')
self.assertEqual(
args.items,
[input.KeyValue(
key='test', value='header', sep=':', orig='test:header')])
def test_guess_when_method_set_but_invalid_and_item_exists(self):
args = argparse.Namespace()
args.method = 'http://example.com/'
args.url = 'new_item=a'
args.items = [
input.KeyValue(
key='old_item', value='b', sep='=', orig='old_item=b')
]
self.parser._guess_method(args, Environment())
self.assertEqual(args.items, [
input.KeyValue(
key='new_item', value='a', sep='=', orig='new_item=a'),
input.KeyValue(key
='old_item', value='b', sep='=', orig='old_item=b'),
])
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 = get_output(args, Environment(), response.request, response).decode('utf8')
for key, value in response_dict.items():
self.assertIn(key, output)
self.assertIn(value, output)
if __name__ == '__main__':
#noinspection PyCallingNonCallable
unittest.main()
Jump to Line
Something went wrong with that request. Please try again.