forked from extern/httpie-cli
3865fabf09
https://docs.python.org/3.10/whatsnew/3.10.html#enum Python 3.10 changed the repr of enum members, and the doctest of tests.utils.http failed. Exact reprs are unfortunately not considered stable API between Python releases: =================================== FAILURES =================================== __________________________ [doctest] tests.utils.http __________________________ 209 210 Example: 211 212 $ http --auth=user:password GET pie.dev/basic-auth/user/password 213 214 >>> httpbin = getfixture('httpbin') 215 >>> r = http('-a', 'user:pw', httpbin.url + '/basic-auth/user/pw') 216 >>> type(r) == StrCLIResponse 217 True 218 >>> r.exit_status Expected: <ExitStatus.SUCCESS: 0> Got: ExitStatus.SUCCESS A simple replacement of the expected output however breaks the doctest on Python 3.9. This is the best solution I could think of that keeps the docstring readable and doctest working in Pythons both old and new.
312 lines
8.9 KiB
Python
312 lines
8.9 KiB
Python
"""Utilities for HTTPie test suite."""
|
|
import re
|
|
import shlex
|
|
import sys
|
|
import time
|
|
import json
|
|
import tempfile
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
from typing import Optional, Union, List
|
|
|
|
from httpie.status import ExitStatus
|
|
from httpie.config import Config
|
|
from httpie.context import Environment
|
|
from httpie.core import main
|
|
|
|
|
|
# pytest-httpbin currently does not support chunked requests:
|
|
# <https://github.com/kevin1024/pytest-httpbin/issues/33>
|
|
# <https://github.com/kevin1024/pytest-httpbin/issues/28>
|
|
HTTPBIN_WITH_CHUNKED_SUPPORT_DOMAIN = 'pie.dev'
|
|
HTTPBIN_WITH_CHUNKED_SUPPORT = 'http://' + HTTPBIN_WITH_CHUNKED_SUPPORT_DOMAIN
|
|
|
|
|
|
TESTS_ROOT = Path(__file__).parent.parent
|
|
CRLF = '\r\n'
|
|
COLOR = '\x1b['
|
|
HTTP_OK = '200 OK'
|
|
# noinspection GrazieInspection
|
|
HTTP_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'
|
|
)
|
|
|
|
|
|
def mk_config_dir() -> Path:
|
|
dirname = tempfile.mkdtemp(prefix='httpie_config_')
|
|
return Path(dirname)
|
|
|
|
|
|
def add_auth(url, auth):
|
|
proto, rest = url.split('://', 1)
|
|
return f'{proto}://{auth}@{rest}'
|
|
|
|
|
|
class StdinBytesIO(BytesIO):
|
|
"""To be used for `MockEnvironment.stdin`"""
|
|
len = 0 # See `prepare_request_body()`
|
|
|
|
|
|
class MockEnvironment(Environment):
|
|
"""Environment subclass with reasonable defaults for testing."""
|
|
colors = 0 # For easier debugging
|
|
stdin_isatty = True,
|
|
stdout_isatty = True
|
|
is_windows = False
|
|
|
|
def __init__(self, create_temp_config_dir=True, **kwargs):
|
|
if 'stdout' not in kwargs:
|
|
kwargs['stdout'] = tempfile.TemporaryFile(
|
|
mode='w+b',
|
|
prefix='httpie_stdout'
|
|
)
|
|
if 'stderr' not in kwargs:
|
|
kwargs['stderr'] = tempfile.TemporaryFile(
|
|
mode='w+t',
|
|
prefix='httpie_stderr'
|
|
)
|
|
super().__init__(**kwargs)
|
|
self._create_temp_config_dir = create_temp_config_dir
|
|
self._delete_config_dir = False
|
|
self._temp_dir = Path(tempfile.gettempdir())
|
|
|
|
@property
|
|
def config(self) -> Config:
|
|
if (self._create_temp_config_dir
|
|
and self._temp_dir not in self.config_dir.parents):
|
|
self.create_temp_config_dir()
|
|
return super().config
|
|
|
|
def create_temp_config_dir(self):
|
|
self.config_dir = mk_config_dir()
|
|
self._delete_config_dir = True
|
|
|
|
def cleanup(self):
|
|
self.stdout.close()
|
|
self.stderr.close()
|
|
if self._delete_config_dir:
|
|
assert self._temp_dir in self.config_dir.parents
|
|
from shutil import rmtree
|
|
rmtree(self.config_dir, ignore_errors=True)
|
|
|
|
def __del__(self):
|
|
# noinspection PyBroadException
|
|
try:
|
|
self.cleanup()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
class BaseCLIResponse:
|
|
"""
|
|
Represents the result of simulated `$ http' invocation via `http()`.
|
|
|
|
Holds and provides access to:
|
|
|
|
- stdout output: print(self)
|
|
- stderr output: print(self.stderr)
|
|
- devnull output: print(self.devnull)
|
|
- exit_status output: print(self.exit_status)
|
|
|
|
"""
|
|
stderr: str = None
|
|
devnull: str = None
|
|
json: dict = None
|
|
exit_status: ExitStatus = None
|
|
command: str = None
|
|
args: List[str] = []
|
|
complete_args: List[str] = []
|
|
|
|
@property
|
|
def command(self):
|
|
cmd = ' '.join(shlex.quote(arg) for arg in ['http', *self.args])
|
|
# pytest-httpbin to real httpbin.
|
|
return re.sub(r'127\.0\.0\.1:\d+', 'httpbin.org', cmd)
|
|
|
|
|
|
class BytesCLIResponse(bytes, BaseCLIResponse):
|
|
"""
|
|
Used as a fallback when a StrCLIResponse cannot be used.
|
|
|
|
E.g. when the output contains binary data or when it is colorized.
|
|
|
|
`.json` will always be None.
|
|
|
|
"""
|
|
|
|
|
|
class StrCLIResponse(str, BaseCLIResponse):
|
|
|
|
@property
|
|
def json(self) -> Optional[dict]:
|
|
"""
|
|
Return deserialized the request or response JSON body,
|
|
if one (and only one) included in the output and is parsable.
|
|
|
|
"""
|
|
if not hasattr(self, '_json'):
|
|
self._json = None
|
|
# De-serialize JSON body if possible.
|
|
if COLOR in self:
|
|
# Colorized output cannot be parsed.
|
|
pass
|
|
elif self.strip().startswith('{'):
|
|
# Looks like JSON body.
|
|
self._json = json.loads(self)
|
|
elif self.count('Content-Type:') == 1:
|
|
# Looks like a HTTP message,
|
|
# try to extract JSON from its body.
|
|
try:
|
|
j = self.strip()[self.strip().rindex('\r\n\r\n'):]
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
try:
|
|
# noinspection PyAttributeOutsideInit
|
|
self._json = json.loads(j)
|
|
except ValueError:
|
|
pass
|
|
return self._json
|
|
|
|
|
|
class ExitStatusError(Exception):
|
|
pass
|
|
|
|
|
|
def http(
|
|
*args,
|
|
program_name='http',
|
|
tolerate_error_exit_status=False,
|
|
**kwargs,
|
|
) -> Union[StrCLIResponse, BytesCLIResponse]:
|
|
# noinspection PyUnresolvedReferences
|
|
"""
|
|
Run HTTPie and capture stderr/out and exit status.
|
|
Content writtent to devnull will be captured only if
|
|
env.devnull is set manually.
|
|
|
|
Invoke `httpie.core.main()` with `args` and `kwargs`,
|
|
and return a `CLIResponse` subclass instance.
|
|
|
|
The return value is either a `StrCLIResponse`, or `BytesCLIResponse`
|
|
if unable to decode the output. Devnull is string when possible,
|
|
bytes otherwise.
|
|
|
|
The response has the following attributes:
|
|
|
|
`stdout` is represented by the instance itself (print r)
|
|
`stderr`: text written to stderr
|
|
`devnull` text written to devnull.
|
|
`exit_status`: the exit status
|
|
`json`: decoded JSON (if possible) or `None`
|
|
|
|
Exceptions are propagated.
|
|
|
|
If you pass ``tolerate_error_exit_status=True``, then error exit statuses
|
|
won't result into an exception.
|
|
|
|
Example:
|
|
|
|
$ http --auth=user:password GET pie.dev/basic-auth/user/password
|
|
|
|
>>> httpbin = getfixture('httpbin')
|
|
>>> r = http('-a', 'user:pw', httpbin.url + '/basic-auth/user/pw')
|
|
>>> type(r) == StrCLIResponse
|
|
True
|
|
>>> r.exit_status is ExitStatus.SUCCESS
|
|
True
|
|
>>> r.stderr
|
|
''
|
|
>>> 'HTTP/1.1 200 OK' in r
|
|
True
|
|
>>> r.json == {'authenticated': True, 'user': 'user'}
|
|
True
|
|
|
|
"""
|
|
env = kwargs.get('env')
|
|
if not env:
|
|
env = kwargs['env'] = MockEnvironment()
|
|
|
|
stdout = env.stdout
|
|
stderr = env.stderr
|
|
devnull = env.devnull
|
|
|
|
args = list(args)
|
|
args_with_config_defaults = args + env.config.default_options
|
|
add_to_args = []
|
|
if '--debug' not in args_with_config_defaults:
|
|
if (not tolerate_error_exit_status
|
|
and '--traceback' not in args_with_config_defaults):
|
|
add_to_args.append('--traceback')
|
|
if not any('--timeout' in arg for arg in args_with_config_defaults):
|
|
add_to_args.append('--timeout=3')
|
|
|
|
complete_args = [program_name, *add_to_args, *args]
|
|
# print(' '.join(complete_args))
|
|
|
|
def dump_stderr():
|
|
stderr.seek(0)
|
|
sys.stderr.write(stderr.read())
|
|
|
|
try:
|
|
try:
|
|
exit_status = main(args=complete_args, **kwargs)
|
|
if '--download' in args:
|
|
# Let the progress reporter thread finish.
|
|
time.sleep(.5)
|
|
except SystemExit:
|
|
if tolerate_error_exit_status:
|
|
exit_status = ExitStatus.ERROR
|
|
else:
|
|
dump_stderr()
|
|
raise
|
|
except Exception:
|
|
stderr.seek(0)
|
|
sys.stderr.write(stderr.read())
|
|
raise
|
|
else:
|
|
if (not tolerate_error_exit_status
|
|
and exit_status != ExitStatus.SUCCESS):
|
|
dump_stderr()
|
|
raise ExitStatusError(
|
|
'httpie.core.main() unexpectedly returned'
|
|
f' a non-zero exit status: {exit_status}'
|
|
)
|
|
|
|
stdout.seek(0)
|
|
stderr.seek(0)
|
|
devnull.seek(0)
|
|
output = stdout.read()
|
|
devnull_output = devnull.read()
|
|
try:
|
|
output = output.decode('utf8')
|
|
except UnicodeDecodeError:
|
|
r = BytesCLIResponse(output)
|
|
else:
|
|
r = StrCLIResponse(output)
|
|
|
|
try:
|
|
devnull_output = devnull_output.decode('utf8')
|
|
except Exception:
|
|
pass
|
|
|
|
r.devnull = devnull_output
|
|
r.stderr = stderr.read()
|
|
r.exit_status = exit_status
|
|
r.args = args
|
|
r.complete_args = ' '.join(complete_args)
|
|
|
|
if r.exit_status != ExitStatus.SUCCESS:
|
|
sys.stderr.write(r.stderr)
|
|
|
|
# print(f'\n\n$ {r.command}\n')
|
|
return r
|
|
|
|
finally:
|
|
devnull.close()
|
|
stdout.close()
|
|
stderr.close()
|
|
env.cleanup()
|