Add converter plugin streaming tests (#1117)

Also fixed minor glitches here and there and re-enabled a unicode test.
This commit is contained in:
Mickaël Schoentgen 2021-08-05 14:37:08 +02:00 committed by GitHub
parent a94d6d807c
commit a66af2497a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 68 additions and 4 deletions

View File

@ -12,6 +12,7 @@ tests_require = [
'docutils',
'pytest',
'pytest-httpbin>=0.0.6',
'responses',
]
dev_require = [
*tests_require,

View File

@ -207,7 +207,7 @@ class TestPrettyOptions:
def test_force_pretty(self, httpbin):
env = MockEnvironment(stdout_isatty=False, colors=256)
r = http('--pretty=all', 'GET', httpbin.url + '/get', env=env, )
r = http('--pretty=all', 'GET', httpbin.url + '/get', env=env)
assert COLOR in r
def test_force_ugly(self, httpbin):

View File

@ -1,10 +1,30 @@
import json
import pytest
import responses
from httpie.compat import is_windows
from httpie.cli.constants import PRETTY_MAP
from httpie.output.streams import BINARY_SUPPRESSED_NOTICE
from httpie.plugins import ConverterPlugin
from httpie.plugins.registry import plugin_manager
from .utils import StdinBytesIO, http, MockEnvironment
from .fixtures import BIN_FILE_CONTENT, BIN_FILE_PATH
PRETTY_OPTIONS = list(PRETTY_MAP.keys())
class SortJSONConverterPlugin(ConverterPlugin):
@classmethod
def supports(cls, mime):
return mime == 'json/bytes'
def convert(self, body):
body = body.lstrip(b'\x00')
data = json.loads(body)
return 'application/json', json.dumps(data, sort_keys=True)
# GET because httpbin 500s with binary POST body.
@ -24,6 +44,47 @@ def test_pretty_redirected_stream(httpbin):
assert BINARY_SUPPRESSED_NOTICE.decode() in r
def test_pretty_stream_ensure_full_stream_is_retrieved(httpbin):
env = MockEnvironment(
stdin=StdinBytesIO(),
stdin_isatty=False,
stdout_isatty=False,
)
r = http('--pretty=format', '--stream', 'GET',
httpbin.url + '/stream/3', env=env)
assert r.count('/stream/3') == 3
@pytest.mark.parametrize('pretty', PRETTY_OPTIONS)
@pytest.mark.parametrize('stream', [True, False])
@responses.activate
def test_pretty_options_with_and_without_stream_with_converter(pretty, stream):
plugin_manager.register(SortJSONConverterPlugin)
try:
# Cover PluginManager.__repr__()
assert 'SortJSONConverterPlugin' in str(plugin_manager)
url = 'http://example.org' # Note: URL never fetched
body = b'\x00{"foo":42,\n"bar":"baz"}'
responses.add(responses.GET, url, body=body,
stream=True, content_type='json/bytes')
args = ['--pretty=' + pretty, 'GET', url]
if stream:
args.insert(0, '--stream')
r = http(*args)
assert 'json/bytes' in r
if pretty == 'none':
assert BINARY_SUPPRESSED_NOTICE.decode() in r
else:
# Ensure the plugin was effectively used and the resulting JSON is sorted
assert '"bar": "baz",' in r
assert '"foo": 42' in r
finally:
plugin_manager.unregister(SortJSONConverterPlugin)
def test_encoded_stream(httpbin):
"""Test that --stream works with non-prettified
redirected terminal output."""

View File

@ -89,9 +89,11 @@ def test_unicode_url(httpbin):
assert HTTP_OK in r
assert r.json['args'] == {'test': UNICODE}
# def test_unicode_url_verbose(self):
# r = http(httpbin.url + '--verbose', f'/get?test={UNICODE}')
# assert HTTP_OK in r
def test_unicode_url_verbose(httpbin):
r = http('--verbose', f'{httpbin.url}/get?test={UNICODE}')
assert HTTP_OK in r
assert r.json['args'] == {'test': UNICODE}
def test_unicode_basic_auth(httpbin):