You've already forked httpie-cli
mirror of
https://github.com/httpie/cli.git
synced 2025-08-10 22:42:05 +02:00
Add warnings when there is no incoming data from stdin (#1256)
* Add warnings when there is no incoming data from stdin * Pass os.environ as well * Apply suggestions
This commit is contained in:
@@ -1,10 +1,16 @@
|
||||
import os
|
||||
import json
|
||||
import sys
|
||||
import subprocess
|
||||
import time
|
||||
import contextlib
|
||||
import httpie.__main__ as main
|
||||
|
||||
import pytest
|
||||
|
||||
from httpie.cli.exceptions import ParseError
|
||||
from httpie.client import FORM_CONTENT_TYPE
|
||||
from httpie.compat import is_windows
|
||||
from httpie.status import ExitStatus
|
||||
from .utils import (
|
||||
MockEnvironment, StdinBytesIO, http,
|
||||
@@ -83,6 +89,85 @@ def test_chunked_raw(httpbin_with_chunked_support):
|
||||
assert 'Transfer-Encoding: chunked' in r
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def stdin_processes(httpbin, *args):
|
||||
process_1 = subprocess.Popen(
|
||||
[
|
||||
"cat"
|
||||
],
|
||||
stdin=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE
|
||||
)
|
||||
process_2 = subprocess.Popen(
|
||||
[
|
||||
sys.executable,
|
||||
main.__file__,
|
||||
"POST",
|
||||
httpbin + "/post",
|
||||
*args
|
||||
],
|
||||
stdin=process_1.stdout,
|
||||
stderr=subprocess.PIPE,
|
||||
env={
|
||||
**os.environ,
|
||||
"HTTPIE_STDIN_READ_WARN_THRESHOLD": "0.1"
|
||||
}
|
||||
)
|
||||
try:
|
||||
yield process_1, process_2
|
||||
finally:
|
||||
process_1.terminate()
|
||||
process_2.terminate()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("wait", (True, False))
|
||||
@pytest.mark.skipif(is_windows, reason="Windows doesn't support select() calls into files")
|
||||
def test_reading_from_stdin(httpbin, wait):
|
||||
with stdin_processes(httpbin) as (process_1, process_2):
|
||||
process_1.communicate(timeout=0.1, input=b"bleh")
|
||||
# Since there is data, it doesn't matter if there
|
||||
# you wait or not.
|
||||
if wait:
|
||||
time.sleep(0.75)
|
||||
|
||||
try:
|
||||
_, errs = process_2.communicate(timeout=0.25)
|
||||
except subprocess.TimeoutExpired:
|
||||
errs = b''
|
||||
|
||||
assert b'> warning: no stdin data read in 0.1s' not in errs
|
||||
|
||||
|
||||
@pytest.mark.skipif(is_windows, reason="Windows doesn't support select() calls into files")
|
||||
def test_stdin_read_warning(httpbin):
|
||||
with stdin_processes(httpbin) as (process_1, process_2):
|
||||
# Wait before sending any data
|
||||
time.sleep(0.75)
|
||||
process_1.communicate(timeout=0.1, input=b"bleh\n")
|
||||
|
||||
try:
|
||||
_, errs = process_2.communicate(timeout=0.25)
|
||||
except subprocess.TimeoutExpired:
|
||||
errs = b''
|
||||
|
||||
assert b'> warning: no stdin data read in 0.1s' in errs
|
||||
|
||||
|
||||
@pytest.mark.skipif(is_windows, reason="Windows doesn't support select() calls into files")
|
||||
def test_stdin_read_warning_with_quiet(httpbin):
|
||||
with stdin_processes(httpbin, "-qq") as (process_1, process_2):
|
||||
# Wait before sending any data
|
||||
time.sleep(0.75)
|
||||
process_1.communicate(timeout=0.1, input=b"bleh\n")
|
||||
|
||||
try:
|
||||
_, errs = process_2.communicate(timeout=0.25)
|
||||
except subprocess.TimeoutExpired:
|
||||
errs = b''
|
||||
|
||||
assert b'> warning: no stdin data read in 0.1s' not in errs
|
||||
|
||||
|
||||
class TestMultipartFormDataFileUpload:
|
||||
|
||||
def test_non_existent_file_raises_parse_error(self, httpbin):
|
||||
|
Reference in New Issue
Block a user