Skip to content

Instantly share code, notes, and snippets.

@CaselIT
Created January 26, 2020 10:10
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save CaselIT/16792866e7dd2c545b186207976e060e to your computer and use it in GitHub Desktop.
Save CaselIT/16792866e7dd2c545b186207976e060e to your computer and use it in GitHub Desktop.
Falcon 3 dev errors on windows
server_port = testing.get_unused_port()
base_url = 'http://{}:{}/'.format(_SERVER_HOST, server_port)
with _run_server_isolated(process_factory, _SERVER_HOST, server_port) as server:
# NOTE(kgriffs): Let the server start up. Give up after 5 seconds.
start_ts = time.time()
while (time.time() - start_ts) < 5:
try:
requests.get(base_url, timeout=0.2)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
time.sleep(0.2)
else:
break
else:
if server.poll() is None:
pytest.fail('Server is not responding to requests')
else:
# NOTE(kgriffs): The server did not start up; probably due to
# the port being in use. We could check the output but
# capsys fixture may not have buffered the error output
# yet, so we just retry.
continue
yield base_url
> assert server.returncode == 0
E assert 1 == 0
E + where 1 = <subprocess.Popen object at 0x000002A2F3D3F9A0>.returncode
tests\asgi\test_asgi_servers.py:178: AssertionError
------------------------------------------------ Captured stdout setup ------------------------------------------------
[Starting server process...]
------------------------------------------------ Captured stderr setup ------------------------------------------------
INFO: Started server process [5748]
INFO: Waiting for application startup.
WARNING: Executing <Task pending name='Task-1' coro=<Server.serve() running at c:\users\federico\miniconda3\envs\falcon\lib\site-packages\uvicorn\main.py:399> wait_for=<Future
pending cb=[<TaskWakeupMethWrapper object at 0x000001DF5272FDF0>()] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:422> cb=[_run_until_complete_cb() at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:184] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:591> took 1.188 seconds
INFO: Application startup complete.
INFO: <Server sockets=(<asyncio.TransportSocket fd=336, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('127.0.0.1', 60289)>,)> is serving
INFO: Uvicorn running on http://127.0.0.1:60289 (Press CTRL+C to quit)
INFO: 127.0.0.1:60296 - "GET / HTTP/1.1" 200
------------------------------------------------ Captured stderr call -------------------------------------------------
INFO: 127.0.0.1:60297 - "GET / HTTP/1.1" 200
---------------------------------------------- Captured stdout teardown -----------------------------------------------
[Sending SIGTERM to server process...]
_____________________________ ERROR at setup of TestASGIServer.test_get[_daphne_factory] ______________________________
request = <SubRequest 'server_base_url' for <Function test_get[_daphne_factory]>>
@pytest.fixture(params=[_uvicorn_factory, _daphne_factory])
def server_base_url(request):
process_factory = request.param
for i in range(3):
server_port = testing.get_unused_port()
base_url = 'http://{}:{}/'.format(_SERVER_HOST, server_port)
with _run_server_isolated(process_factory, _SERVER_HOST, server_port) as server:
# NOTE(kgriffs): Let the server start up. Give up after 5 seconds.
start_ts = time.time()
while (time.time() - start_ts) < 5:
try:
requests.get(base_url, timeout=0.2)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
time.sleep(0.2)
else:
break
else:
if server.poll() is None:
pytest.fail('Server is not responding to requests')
else:
# NOTE(kgriffs): The server did not start up; probably due to
# the port being in use. We could check the output but
# capsys fixture may not have buffered the error output
# yet, so we just retry.
continue
yield base_url
assert server.returncode == 0
break
else:
> pytest.fail('Could not start server')
E Failed: Could not start server
tests\asgi\test_asgi_servers.py:182: Failed
------------------------------------------------ Captured stdout setup ------------------------------------------------
[Starting server process...]
[Sending SIGTERM to server process...]
[Starting server process...]
[Sending SIGTERM to server process...]
[Starting server process...]
[Sending SIGTERM to server process...]
------------------------------------------------ Captured stderr setup ------------------------------------------------
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
___________________________ ERROR at teardown of TestASGIServer.test_put[_uvicorn_factory] ____________________________
request = <SubRequest 'server_base_url' for <Function test_put[_uvicorn_factory]>>
@pytest.fixture(params=[_uvicorn_factory, _daphne_factory])
def server_base_url(request):
process_factory = request.param
for i in range(3):
server_port = testing.get_unused_port()
base_url = 'http://{}:{}/'.format(_SERVER_HOST, server_port)
with _run_server_isolated(process_factory, _SERVER_HOST, server_port) as server:
# NOTE(kgriffs): Let the server start up. Give up after 5 seconds.
start_ts = time.time()
while (time.time() - start_ts) < 5:
try:
requests.get(base_url, timeout=0.2)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
time.sleep(0.2)
else:
break
else:
if server.poll() is None:
pytest.fail('Server is not responding to requests')
else:
# NOTE(kgriffs): The server did not start up; probably due to
# the port being in use. We could check the output but
# capsys fixture may not have buffered the error output
# yet, so we just retry.
continue
yield base_url
> assert server.returncode == 0
E assert 1 == 0
E + where 1 = <subprocess.Popen object at 0x000002A2F3DB0220>.returncode
tests\asgi\test_asgi_servers.py:178: AssertionError
------------------------------------------------ Captured stdout setup ------------------------------------------------
[Starting server process...]
------------------------------------------------ Captured stderr setup ------------------------------------------------
INFO: Started server process [5924]
INFO: Waiting for application startup.
WARNING: Executing <Task pending name='Task-1' coro=<Server.serve() running at c:\users\federico\miniconda3\envs\falcon\lib\site-packages\uvicorn\main.py:399> wait_for=<Future
pending cb=[<TaskWakeupMethWrapper object at 0x0000023B494CEDF0>()] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:422> cb=[_run_until_complete_cb() at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:184] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:591> took 1.187 seconds
INFO: Application startup complete.
INFO: <Server sockets=(<asyncio.TransportSocket fd=716, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('127.0.0.1', 60350)>,)> is serving
INFO: Uvicorn running on http://127.0.0.1:60350 (Press CTRL+C to quit)
INFO: 127.0.0.1:60358 - "GET / HTTP/1.1" 200
------------------------------------------------ Captured stderr call -------------------------------------------------
WARNING: Executing <Task pending name='Task-7' coro=<RequestResponseCycle.run_asgi() running at c:\users\federico\miniconda3\envs\falcon\lib\site-packages\uvicorn\protocols\http\h11_impl.py:385> wait_for=<_GatheringFuture pending cb=[<TaskWakeupMethWrapper object at 0x0000023B49C4A310>()] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\tasks.py:698> cb=[set.discard()] created at c:\users\federico\miniconda3\envs\falcon\lib\site-packages\uvicorn\protocols\http\h11_impl.py:234> took 1.578 seconds
---------------------------------------------- Captured stdout teardown -----------------------------------------------
[Sending SIGTERM to server process...]
_____________________________ ERROR at setup of TestASGIServer.test_put[_daphne_factory] ______________________________
request = <SubRequest 'server_base_url' for <Function test_put[_daphne_factory]>>
@pytest.fixture(params=[_uvicorn_factory, _daphne_factory])
def server_base_url(request):
process_factory = request.param
for i in range(3):
server_port = testing.get_unused_port()
base_url = 'http://{}:{}/'.format(_SERVER_HOST, server_port)
with _run_server_isolated(process_factory, _SERVER_HOST, server_port) as server:
# NOTE(kgriffs): Let the server start up. Give up after 5 seconds.
start_ts = time.time()
while (time.time() - start_ts) < 5:
try:
requests.get(base_url, timeout=0.2)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
time.sleep(0.2)
else:
break
else:
if server.poll() is None:
pytest.fail('Server is not responding to requests')
else:
# NOTE(kgriffs): The server did not start up; probably due to
# the port being in use. We could check the output but
# capsys fixture may not have buffered the error output
# yet, so we just retry.
continue
yield base_url
assert server.returncode == 0
break
else:
> pytest.fail('Could not start server')
E Failed: Could not start server
tests\asgi\test_asgi_servers.py:182: Failed
------------------------------------------------ Captured stdout setup ------------------------------------------------
[Starting server process...]
[Sending SIGTERM to server process...]
[Starting server process...]
[Sending SIGTERM to server process...]
[Starting server process...]
[Sending SIGTERM to server process...]
------------------------------------------------ Captured stderr setup ------------------------------------------------
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
_________________________ ERROR at teardown of TestASGIServer.test_head_405[_uvicorn_factory] _________________________
request = <SubRequest 'server_base_url' for <Function test_head_405[_uvicorn_factory]>>
@pytest.fixture(params=[_uvicorn_factory, _daphne_factory])
def server_base_url(request):
process_factory = request.param
for i in range(3):
server_port = testing.get_unused_port()
base_url = 'http://{}:{}/'.format(_SERVER_HOST, server_port)
with _run_server_isolated(process_factory, _SERVER_HOST, server_port) as server:
# NOTE(kgriffs): Let the server start up. Give up after 5 seconds.
start_ts = time.time()
while (time.time() - start_ts) < 5:
try:
requests.get(base_url, timeout=0.2)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
time.sleep(0.2)
else:
break
else:
if server.poll() is None:
pytest.fail('Server is not responding to requests')
else:
# NOTE(kgriffs): The server did not start up; probably due to
# the port being in use. We could check the output but
# capsys fixture may not have buffered the error output
# yet, so we just retry.
continue
yield base_url
> assert server.returncode == 0
E assert 1 == 0
E + where 1 = <subprocess.Popen object at 0x000002A2F2559550>.returncode
tests\asgi\test_asgi_servers.py:178: AssertionError
------------------------------------------------ Captured stdout setup ------------------------------------------------
[Starting server process...]
------------------------------------------------ Captured stderr setup ------------------------------------------------
INFO: Started server process [26540]
INFO: Waiting for application startup.
WARNING: Executing <Task pending name='Task-1' coro=<Server.serve() running at c:\users\federico\miniconda3\envs\falcon\lib\site-packages\uvicorn\main.py:399> wait_for=<Future
pending cb=[<TaskWakeupMethWrapper object at 0x000001F81FE0FDF0>()] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:422> cb=[_run_until_complete_cb() at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:184] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:591> took 1.078 seconds
INFO: Application startup complete.
INFO: <Server sockets=(<asyncio.TransportSocket fd=756, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('127.0.0.1', 60419)>,)> is serving
INFO: Uvicorn running on http://127.0.0.1:60419 (Press CTRL+C to quit)
INFO: 127.0.0.1:60426 - "GET / HTTP/1.1" 200
------------------------------------------------ Captured stderr call -------------------------------------------------
INFO: 127.0.0.1:60427 - "HEAD / HTTP/1.1" 405
---------------------------------------------- Captured stdout teardown -----------------------------------------------
[Sending SIGTERM to server process...]
___________________________ ERROR at setup of TestASGIServer.test_head_405[_daphne_factory] ___________________________
request = <SubRequest 'server_base_url' for <Function test_head_405[_daphne_factory]>>
@pytest.fixture(params=[_uvicorn_factory, _daphne_factory])
def server_base_url(request):
process_factory = request.param
for i in range(3):
server_port = testing.get_unused_port()
base_url = 'http://{}:{}/'.format(_SERVER_HOST, server_port)
with _run_server_isolated(process_factory, _SERVER_HOST, server_port) as server:
# NOTE(kgriffs): Let the server start up. Give up after 5 seconds.
start_ts = time.time()
while (time.time() - start_ts) < 5:
try:
requests.get(base_url, timeout=0.2)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
time.sleep(0.2)
else:
break
else:
if server.poll() is None:
pytest.fail('Server is not responding to requests')
else:
# NOTE(kgriffs): The server did not start up; probably due to
# the port being in use. We could check the output but
# capsys fixture may not have buffered the error output
# yet, so we just retry.
continue
yield base_url
assert server.returncode == 0
break
else:
> pytest.fail('Could not start server')
E Failed: Could not start server
tests\asgi\test_asgi_servers.py:182: Failed
------------------------------------------------ Captured stdout setup ------------------------------------------------
[Starting server process...]
[Sending SIGTERM to server process...]
[Starting server process...]
[Sending SIGTERM to server process...]
[Starting server process...]
[Sending SIGTERM to server process...]
------------------------------------------------ Captured stderr setup ------------------------------------------------
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
______________________ ERROR at teardown of TestASGIServer.test_post_multiple[_uvicorn_factory] _______________________
request = <SubRequest 'server_base_url' for <Function test_post_multiple[_uvicorn_factory]>>
@pytest.fixture(params=[_uvicorn_factory, _daphne_factory])
def server_base_url(request):
process_factory = request.param
for i in range(3):
server_port = testing.get_unused_port()
base_url = 'http://{}:{}/'.format(_SERVER_HOST, server_port)
with _run_server_isolated(process_factory, _SERVER_HOST, server_port) as server:
# NOTE(kgriffs): Let the server start up. Give up after 5 seconds.
start_ts = time.time()
while (time.time() - start_ts) < 5:
try:
requests.get(base_url, timeout=0.2)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
time.sleep(0.2)
else:
break
else:
if server.poll() is None:
pytest.fail('Server is not responding to requests')
else:
# NOTE(kgriffs): The server did not start up; probably due to
# the port being in use. We could check the output but
# capsys fixture may not have buffered the error output
# yet, so we just retry.
continue
yield base_url
> assert server.returncode == 0
E assert 1 == 0
E + where 1 = <subprocess.Popen object at 0x000002A2F3B05640>.returncode
tests\asgi\test_asgi_servers.py:178: AssertionError
------------------------------------------------ Captured stdout setup ------------------------------------------------
[Starting server process...]
------------------------------------------------ Captured stderr setup ------------------------------------------------
INFO: Started server process [17576]
INFO: Waiting for application startup.
WARNING: Executing <Task pending name='Task-1' coro=<Server.serve() running at c:\users\federico\miniconda3\envs\falcon\lib\site-packages\uvicorn\main.py:399> wait_for=<Future
pending cb=[<TaskWakeupMethWrapper object at 0x0000023BDFF3FDF0>()] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:422> cb=[_run_until_complete_cb() at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:184] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:591> took 1.031 seconds
INFO: Application startup complete.
INFO: <Server sockets=(<asyncio.TransportSocket fd=88, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('127.0.0.1', 60481)>,)> is serving
INFO: Uvicorn running on http://127.0.0.1:60481 (Press CTRL+C to quit)
INFO: 127.0.0.1:60488 - "GET / HTTP/1.1" 200
------------------------------------------------ Captured stderr call -------------------------------------------------
INFO: 127.0.0.1:60489 - "POST / HTTP/1.1" 200
INFO: 127.0.0.1:60490 - "POST / HTTP/1.1" 200
---------------------------------------------- Captured stdout teardown -----------------------------------------------
[Sending SIGTERM to server process...]
________________________ ERROR at setup of TestASGIServer.test_post_multiple[_daphne_factory] _________________________
request = <SubRequest 'server_base_url' for <Function test_post_multiple[_daphne_factory]>>
@pytest.fixture(params=[_uvicorn_factory, _daphne_factory])
def server_base_url(request):
process_factory = request.param
for i in range(3):
server_port = testing.get_unused_port()
base_url = 'http://{}:{}/'.format(_SERVER_HOST, server_port)
with _run_server_isolated(process_factory, _SERVER_HOST, server_port) as server:
# NOTE(kgriffs): Let the server start up. Give up after 5 seconds.
start_ts = time.time()
while (time.time() - start_ts) < 5:
try:
requests.get(base_url, timeout=0.2)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
time.sleep(0.2)
else:
break
else:
if server.poll() is None:
pytest.fail('Server is not responding to requests')
else:
# NOTE(kgriffs): The server did not start up; probably due to
# the port being in use. We could check the output but
# capsys fixture may not have buffered the error output
# yet, so we just retry.
continue
yield base_url
assert server.returncode == 0
break
else:
> pytest.fail('Could not start server')
E Failed: Could not start server
tests\asgi\test_asgi_servers.py:182: Failed
------------------------------------------------ Captured stdout setup ------------------------------------------------
[Starting server process...]
[Sending SIGTERM to server process...]
[Starting server process...]
[Sending SIGTERM to server process...]
[Starting server process...]
[Sending SIGTERM to server process...]
------------------------------------------------ Captured stderr setup ------------------------------------------------
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
_______________ ERROR at teardown of TestASGIServer.test_post_invalid_content_length[_uvicorn_factory] ________________
request = <SubRequest 'server_base_url' for <Function test_post_invalid_content_length[_uvicorn_factory]>>
@pytest.fixture(params=[_uvicorn_factory, _daphne_factory])
def server_base_url(request):
process_factory = request.param
for i in range(3):
server_port = testing.get_unused_port()
base_url = 'http://{}:{}/'.format(_SERVER_HOST, server_port)
with _run_server_isolated(process_factory, _SERVER_HOST, server_port) as server:
# NOTE(kgriffs): Let the server start up. Give up after 5 seconds.
start_ts = time.time()
while (time.time() - start_ts) < 5:
try:
requests.get(base_url, timeout=0.2)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
time.sleep(0.2)
else:
break
else:
if server.poll() is None:
pytest.fail('Server is not responding to requests')
else:
# NOTE(kgriffs): The server did not start up; probably due to
# the port being in use. We could check the output but
# capsys fixture may not have buffered the error output
# yet, so we just retry.
continue
yield base_url
> assert server.returncode == 0
E assert 1 == 0
E + where 1 = <subprocess.Popen object at 0x000002A2F39719A0>.returncode
tests\asgi\test_asgi_servers.py:178: AssertionError
------------------------------------------------ Captured stdout setup ------------------------------------------------
[Starting server process...]
------------------------------------------------ Captured stderr setup ------------------------------------------------
INFO: Started server process [24240]
INFO: Waiting for application startup.
WARNING: Executing <Task pending name='Task-1' coro=<Server.serve() running at c:\users\federico\miniconda3\envs\falcon\lib\site-packages\uvicorn\main.py:399> wait_for=<Future
pending cb=[<TaskWakeupMethWrapper object at 0x00000239FD9EFDF0>()] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:422> cb=[_run_until_complete_cb() at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:184] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:591> took 1.031 seconds
INFO: Application startup complete.
INFO: <Server sockets=(<asyncio.TransportSocket fd=332, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('127.0.0.1', 60546)>,)> is serving
INFO: Uvicorn running on http://127.0.0.1:60546 (Press CTRL+C to quit)
INFO: 127.0.0.1:60553 - "GET / HTTP/1.1" 200
------------------------------------------------ Captured stderr call -------------------------------------------------
WARNING: Invalid HTTP request received.
---------------------------------------------- Captured stdout teardown -----------------------------------------------
[Sending SIGTERM to server process...]
_________________ ERROR at setup of TestASGIServer.test_post_invalid_content_length[_daphne_factory] __________________
request = <SubRequest 'server_base_url' for <Function test_post_invalid_content_length[_daphne_factory]>>
@pytest.fixture(params=[_uvicorn_factory, _daphne_factory])
def server_base_url(request):
process_factory = request.param
for i in range(3):
server_port = testing.get_unused_port()
base_url = 'http://{}:{}/'.format(_SERVER_HOST, server_port)
with _run_server_isolated(process_factory, _SERVER_HOST, server_port) as server:
# NOTE(kgriffs): Let the server start up. Give up after 5 seconds.
start_ts = time.time()
while (time.time() - start_ts) < 5:
try:
requests.get(base_url, timeout=0.2)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
time.sleep(0.2)
else:
break
else:
if server.poll() is None:
pytest.fail('Server is not responding to requests')
else:
# NOTE(kgriffs): The server did not start up; probably due to
# the port being in use. We could check the output but
# capsys fixture may not have buffered the error output
# yet, so we just retry.
continue
yield base_url
assert server.returncode == 0
break
else:
> pytest.fail('Could not start server')
E Failed: Could not start server
tests\asgi\test_asgi_servers.py:182: Failed
------------------------------------------------ Captured stdout setup ------------------------------------------------
[Starting server process...]
[Sending SIGTERM to server process...]
[Starting server process...]
[Sending SIGTERM to server process...]
[Starting server process...]
[Sending SIGTERM to server process...]
------------------------------------------------ Captured stderr setup ------------------------------------------------
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
_________________ ERROR at teardown of TestASGIServer.test_post_read_bounded_stream[_uvicorn_factory] _________________
request = <SubRequest 'server_base_url' for <Function test_post_read_bounded_stream[_uvicorn_factory]>>
@pytest.fixture(params=[_uvicorn_factory, _daphne_factory])
def server_base_url(request):
process_factory = request.param
for i in range(3):
server_port = testing.get_unused_port()
base_url = 'http://{}:{}/'.format(_SERVER_HOST, server_port)
with _run_server_isolated(process_factory, _SERVER_HOST, server_port) as server:
# NOTE(kgriffs): Let the server start up. Give up after 5 seconds.
start_ts = time.time()
while (time.time() - start_ts) < 5:
try:
requests.get(base_url, timeout=0.2)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
time.sleep(0.2)
else:
break
else:
if server.poll() is None:
pytest.fail('Server is not responding to requests')
else:
# NOTE(kgriffs): The server did not start up; probably due to
# the port being in use. We could check the output but
# capsys fixture may not have buffered the error output
# yet, so we just retry.
continue
yield base_url
> assert server.returncode == 0
E assert 1 == 0
E + where 1 = <subprocess.Popen object at 0x000002A2F38C1E80>.returncode
tests\asgi\test_asgi_servers.py:178: AssertionError
------------------------------------------------ Captured stdout setup ------------------------------------------------
[Starting server process...]
------------------------------------------------ Captured stderr setup ------------------------------------------------
INFO: Started server process [6116]
INFO: Waiting for application startup.
WARNING: Executing <Task pending name='Task-1' coro=<Server.serve() running at c:\users\federico\miniconda3\envs\falcon\lib\site-packages\uvicorn\main.py:399> wait_for=<Future
pending cb=[<TaskWakeupMethWrapper object at 0x000002966BFFFDF0>()] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:422> cb=[_run_until_complete_cb() at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:184] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:591> took 1.047 seconds
INFO: Application startup complete.
INFO: <Server sockets=(<asyncio.TransportSocket fd=340, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('127.0.0.1', 60607)>,)> is serving
INFO: Uvicorn running on http://127.0.0.1:60607 (Press CTRL+C to quit)
INFO: 127.0.0.1:60614 - "GET / HTTP/1.1" 200
------------------------------------------------ Captured stderr call -------------------------------------------------
INFO: 127.0.0.1:60615 - "POST /bucket HTTP/1.1" 200
---------------------------------------------- Captured stdout teardown -----------------------------------------------
[Sending SIGTERM to server process...]
___________________ ERROR at setup of TestASGIServer.test_post_read_bounded_stream[_daphne_factory] ___________________
request = <SubRequest 'server_base_url' for <Function test_post_read_bounded_stream[_daphne_factory]>>
@pytest.fixture(params=[_uvicorn_factory, _daphne_factory])
def server_base_url(request):
process_factory = request.param
for i in range(3):
server_port = testing.get_unused_port()
base_url = 'http://{}:{}/'.format(_SERVER_HOST, server_port)
with _run_server_isolated(process_factory, _SERVER_HOST, server_port) as server:
# NOTE(kgriffs): Let the server start up. Give up after 5 seconds.
start_ts = time.time()
while (time.time() - start_ts) < 5:
try:
requests.get(base_url, timeout=0.2)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
time.sleep(0.2)
else:
break
else:
if server.poll() is None:
pytest.fail('Server is not responding to requests')
else:
# NOTE(kgriffs): The server did not start up; probably due to
# the port being in use. We could check the output but
# capsys fixture may not have buffered the error output
# yet, so we just retry.
continue
yield base_url
assert server.returncode == 0
break
else:
> pytest.fail('Could not start server')
E Failed: Could not start server
tests\asgi\test_asgi_servers.py:182: Failed
------------------------------------------------ Captured stdout setup ------------------------------------------------
[Starting server process...]
[Sending SIGTERM to server process...]
[Starting server process...]
[Sending SIGTERM to server process...]
[Starting server process...]
[Sending SIGTERM to server process...]
------------------------------------------------ Captured stderr setup ------------------------------------------------
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
_____________ ERROR at teardown of TestASGIServer.test_post_read_bounded_stream_no_body[_uvicorn_factory] _____________
request = <SubRequest 'server_base_url' for <Function test_post_read_bounded_stream_no_body[_uvicorn_factory]>>
@pytest.fixture(params=[_uvicorn_factory, _daphne_factory])
def server_base_url(request):
process_factory = request.param
for i in range(3):
server_port = testing.get_unused_port()
base_url = 'http://{}:{}/'.format(_SERVER_HOST, server_port)
with _run_server_isolated(process_factory, _SERVER_HOST, server_port) as server:
# NOTE(kgriffs): Let the server start up. Give up after 5 seconds.
start_ts = time.time()
while (time.time() - start_ts) < 5:
try:
requests.get(base_url, timeout=0.2)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
time.sleep(0.2)
else:
break
else:
if server.poll() is None:
pytest.fail('Server is not responding to requests')
else:
# NOTE(kgriffs): The server did not start up; probably due to
# the port being in use. We could check the output but
# capsys fixture may not have buffered the error output
# yet, so we just retry.
continue
yield base_url
> assert server.returncode == 0
E assert 1 == 0
E + where 1 = <subprocess.Popen object at 0x000002A2F3AB9D60>.returncode
tests\asgi\test_asgi_servers.py:178: AssertionError
------------------------------------------------ Captured stdout setup ------------------------------------------------
[Starting server process...]
------------------------------------------------ Captured stderr setup ------------------------------------------------
INFO: Started server process [10380]
INFO: Waiting for application startup.
WARNING: Executing <Task pending name='Task-1' coro=<Server.serve() running at c:\users\federico\miniconda3\envs\falcon\lib\site-packages\uvicorn\main.py:399> wait_for=<Future
pending cb=[<TaskWakeupMethWrapper object at 0x0000024B9BD3FDF0>()] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:422> cb=[_run_until_complete_cb() at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:184] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:591> took 1.047 seconds
INFO: Application startup complete.
INFO: <Server sockets=(<asyncio.TransportSocket fd=332, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('127.0.0.1', 60667)>,)> is serving
INFO: Uvicorn running on http://127.0.0.1:60667 (Press CTRL+C to quit)
INFO: 127.0.0.1:60675 - "GET / HTTP/1.1" 200
------------------------------------------------ Captured stderr call -------------------------------------------------
INFO: 127.0.0.1:60676 - "POST /bucket HTTP/1.1" 200
---------------------------------------------- Captured stdout teardown -----------------------------------------------
[Sending SIGTERM to server process...]
_______________ ERROR at setup of TestASGIServer.test_post_read_bounded_stream_no_body[_daphne_factory] _______________
request = <SubRequest 'server_base_url' for <Function test_post_read_bounded_stream_no_body[_daphne_factory]>>
@pytest.fixture(params=[_uvicorn_factory, _daphne_factory])
def server_base_url(request):
process_factory = request.param
for i in range(3):
server_port = testing.get_unused_port()
base_url = 'http://{}:{}/'.format(_SERVER_HOST, server_port)
with _run_server_isolated(process_factory, _SERVER_HOST, server_port) as server:
# NOTE(kgriffs): Let the server start up. Give up after 5 seconds.
start_ts = time.time()
while (time.time() - start_ts) < 5:
try:
requests.get(base_url, timeout=0.2)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
time.sleep(0.2)
else:
break
else:
if server.poll() is None:
pytest.fail('Server is not responding to requests')
else:
# NOTE(kgriffs): The server did not start up; probably due to
# the port being in use. We could check the output but
# capsys fixture may not have buffered the error output
# yet, so we just retry.
continue
yield base_url
assert server.returncode == 0
break
else:
> pytest.fail('Could not start server')
E Failed: Could not start server
tests\asgi\test_asgi_servers.py:182: Failed
------------------------------------------------ Captured stdout setup ------------------------------------------------
[Starting server process...]
[Sending SIGTERM to server process...]
[Starting server process...]
[Sending SIGTERM to server process...]
[Starting server process...]
[Sending SIGTERM to server process...]
------------------------------------------------ Captured stderr setup ------------------------------------------------
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
___________________________ ERROR at teardown of TestASGIServer.test_sse[_uvicorn_factory] ____________________________
request = <SubRequest 'server_base_url' for <Function test_sse[_uvicorn_factory]>>
@pytest.fixture(params=[_uvicorn_factory, _daphne_factory])
def server_base_url(request):
process_factory = request.param
for i in range(3):
server_port = testing.get_unused_port()
base_url = 'http://{}:{}/'.format(_SERVER_HOST, server_port)
with _run_server_isolated(process_factory, _SERVER_HOST, server_port) as server:
# NOTE(kgriffs): Let the server start up. Give up after 5 seconds.
start_ts = time.time()
while (time.time() - start_ts) < 5:
try:
requests.get(base_url, timeout=0.2)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
time.sleep(0.2)
else:
break
else:
if server.poll() is None:
pytest.fail('Server is not responding to requests')
else:
# NOTE(kgriffs): The server did not start up; probably due to
# the port being in use. We could check the output but
# capsys fixture may not have buffered the error output
# yet, so we just retry.
continue
yield base_url
> assert server.returncode == 0
E assert 1 == 0
E + where 1 = <subprocess.Popen object at 0x000002A2F3969610>.returncode
tests\asgi\test_asgi_servers.py:178: AssertionError
------------------------------------------------ Captured stdout setup ------------------------------------------------
[Starting server process...]
------------------------------------------------ Captured stderr setup ------------------------------------------------
INFO: Started server process [13696]
INFO: Waiting for application startup.
WARNING: Executing <Task pending name='Task-1' coro=<Server.serve() running at c:\users\federico\miniconda3\envs\falcon\lib\site-packages\uvicorn\main.py:399> wait_for=<Future
pending cb=[<TaskWakeupMethWrapper object at 0x000002439FB6FDF0>()] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:422> cb=[_run_until_complete_cb() at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:184] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:591> took 1.063 seconds
INFO: Application startup complete.
INFO: <Server sockets=(<asyncio.TransportSocket fd=84, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('127.0.0.1', 60729)>,)> is serving
INFO: Uvicorn running on http://127.0.0.1:60729 (Press CTRL+C to quit)
INFO: 127.0.0.1:60736 - "GET / HTTP/1.1" 200
------------------------------------------------ Captured stderr call -------------------------------------------------
INFO: 127.0.0.1:60737 - "GET /events HTTP/1.1" 200
---------------------------------------------- Captured stdout teardown -----------------------------------------------
[Sending SIGTERM to server process...]
_____________________________ ERROR at setup of TestASGIServer.test_sse[_daphne_factory] ______________________________
request = <SubRequest 'server_base_url' for <Function test_sse[_daphne_factory]>>
@pytest.fixture(params=[_uvicorn_factory, _daphne_factory])
def server_base_url(request):
process_factory = request.param
for i in range(3):
server_port = testing.get_unused_port()
base_url = 'http://{}:{}/'.format(_SERVER_HOST, server_port)
with _run_server_isolated(process_factory, _SERVER_HOST, server_port) as server:
# NOTE(kgriffs): Let the server start up. Give up after 5 seconds.
start_ts = time.time()
while (time.time() - start_ts) < 5:
try:
requests.get(base_url, timeout=0.2)
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError):
time.sleep(0.2)
else:
break
else:
if server.poll() is None:
pytest.fail('Server is not responding to requests')
else:
# NOTE(kgriffs): The server did not start up; probably due to
# the port being in use. We could check the output but
# capsys fixture may not have buffered the error output
# yet, so we just retry.
continue
yield base_url
assert server.returncode == 0
break
else:
> pytest.fail('Could not start server')
E Failed: Could not start server
tests\asgi\test_asgi_servers.py:182: Failed
------------------------------------------------ Captured stdout setup ------------------------------------------------
[Starting server process...]
[Sending SIGTERM to server process...]
[Starting server process...]
[Sending SIGTERM to server process...]
[Starting server process...]
[Sending SIGTERM to server process...]
------------------------------------------------ Captured stderr setup ------------------------------------------------
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
Traceback (most recent call last):
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 193, in _run_module_as_main
return _run_code(code, main_globals, None,
File "c:\users\federico\miniconda3\envs\falcon\lib\runpy.py", line 86, in _run_code
exec(code, run_globals)
File "C:\Users\Federico\Miniconda3\envs\falcon\Scripts\daphne.exe\__main__.py", line 4, in <module>
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\cli.py", line 11, in <module>
from .server import Server
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\daphne\server.py", line 20, in <module>
asyncioreactor.install(twisted_loop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 320, in install
reactor = AsyncioSelectorReactor(eventloop)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 69, in __init__
super().__init__()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\base.py", line 571, in __init__
self.installWaker()
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\posixbase.py", line 286, in installWaker
self.addReader(self.waker)
File "c:\users\federico\miniconda3\envs\falcon\lib\site-packages\twisted\internet\asyncioreactor.py", line 151, in addReader
self._asyncioEventloop.add_reader(fd, callWithLogger, reader,
File "c:\users\federico\miniconda3\envs\falcon\lib\asyncio\events.py", line 501, in add_reader
raise NotImplementedError
NotImplementedError
====================================================== FAILURES =======================================================
__________________________________________ test_traverse_with_verbose[True] ___________________________________________
app = <falcon.asgi.app.App object at 0x000002A2F23A0680>
def test_traverse_with_verbose(app):
"""Ensure traverse() finds the proper routes and outputs verbose info."""
output = io.StringIO()
with redirected(stdout=output):
print_routes.traverse(app._router._roots, verbose=True)
route, get_info, options_info = output.getvalue().strip().split('\n')
assert '-> /test' == route
# NOTE(kgriffs) We might receive these in either order, since the
# method map is not ordered, so check and swap if necessary.
if options_info.startswith('-->GET'):
get_info, options_info = options_info, get_info
assert options_info.startswith('-->OPTIONS')
> assert 'falcon/responders.py:' in options_info
E AssertionError: assert 'falcon/responders.py:' in '-->OPTIONS c:\\users\\federico\\miniconda3\\envs\\falcon\\lib\\site-packages\\falcon\\responders.py:77'
tests\test_cmd_print_api.py:49: AssertionError
__________________________________________ test_traverse_with_verbose[False] __________________________________________
app = <falcon.app.App object at 0x000002A2F23FA8B0>
def test_traverse_with_verbose(app):
"""Ensure traverse() finds the proper routes and outputs verbose info."""
output = io.StringIO()
with redirected(stdout=output):
print_routes.traverse(app._router._roots, verbose=True)
route, get_info, options_info = output.getvalue().strip().split('\n')
assert '-> /test' == route
# NOTE(kgriffs) We might receive these in either order, since the
# method map is not ordered, so check and swap if necessary.
if options_info.startswith('-->GET'):
get_info, options_info = options_info, get_info
assert options_info.startswith('-->OPTIONS')
> assert 'falcon/responders.py:' in options_info
E AssertionError: assert 'falcon/responders.py:' in '-->OPTIONS c:\\users\\federico\\miniconda3\\envs\\falcon\\lib\\site-packages\\falcon\\responders.py:84'
tests\test_cmd_print_api.py:49: AssertionError
_________________________ test_good_path[True-/static/-/css/test.css-/css/test.css-text/css] __________________________
asgi = True, uri_prefix = '/static/', uri_path = '/css/test.css', expected_path = '/css/test.css', mtype = 'text/css'
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3B3A430>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
assert resp.content_type == mtype
> assert body.decode() == '/var/www/statics' + expected_path
E AssertionError: assert '/var/www/sta...css\\test.css' == '/var/www/sta.../css/test.css'
E - /var/www/statics\css\test.css
E ? ^ ^
E + /var/www/statics/css/test.css
E ? ^ ^
tests\test_static.py:184: AssertionError
__________________________ test_good_path[True-/static-/css/test.css-/css/test.css-text/css] __________________________
asgi = True, uri_prefix = '/static', uri_path = '/css/test.css', expected_path = '/css/test.css', mtype = 'text/css'
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3B30BE0>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
assert resp.content_type == mtype
> assert body.decode() == '/var/www/statics' + expected_path
E AssertionError: assert '/var/www/sta...css\\test.css' == '/var/www/sta.../css/test.css'
E - /var/www/statics\css\test.css
E ? ^ ^
E + /var/www/statics/css/test.css
E ? ^ ^
tests\test_static.py:184: AssertionError
_ test_good_path[True-/static-/tttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttt-/tttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttt-application/octet-stream] _
asgi = True, uri_prefix = '/static'
uri_path = '/tttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttt...tttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttt'
expected_path = '/tttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttt...tttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttt'
mtype = 'application/octet-stream', monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3B38F40>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
assert resp.content_type == mtype
> assert body.decode() == '/var/www/statics' + expected_path
E AssertionError: assert '/var/www/sta...ttttttttttttt' == '/var/www/sta...ttttttttttttt'
E Skipping 503 identical trailing characters in diff, use -v to show
E - /var/www/statics\ttttttttt
E ? ^
E + /var/www/statics/ttttttttt
E ? ^
tests\test_static.py:184: AssertionError
_____________________________ test_good_path[True-/static-/.test.css-/.test.css-text/css] _____________________________
asgi = True, uri_prefix = '/static', uri_path = '/.test.css', expected_path = '/.test.css', mtype = 'text/css'
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3B24B80>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
assert resp.content_type == mtype
> assert body.decode() == '/var/www/statics' + expected_path
E AssertionError: assert '/var/www/statics\\.test.css' == '/var/www/statics/.test.css'
E - /var/www/statics\.test.css
E ? ^
E + /var/www/statics/.test.css
E ? ^
tests\test_static.py:184: AssertionError
____________________ test_good_path[True-/some/download/-/report.pdf-/report.pdf-application/pdf] _____________________
asgi = True, uri_prefix = '/some/download/', uri_path = '/report.pdf', expected_path = '/report.pdf'
mtype = 'application/pdf', monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3B41C10>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
assert resp.content_type == mtype
> assert body.decode() == '/var/www/statics' + expected_path
E AssertionError: assert '/var/www/statics\\report.pdf' == '/var/www/statics/report.pdf'
E - /var/www/statics\report.pdf
E ? ^
E + /var/www/statics/report.pdf
E ? ^
tests\test_static.py:184: AssertionError
______________ test_good_path[True-/some/download/-/Fancy Report.pdf-/Fancy Report.pdf-application/pdf] _______________
asgi = True, uri_prefix = '/some/download/', uri_path = '/Fancy Report.pdf', expected_path = '/Fancy Report.pdf'
mtype = 'application/pdf', monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3B38A30>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
assert resp.content_type == mtype
> assert body.decode() == '/var/www/statics' + expected_path
E AssertionError: assert '/var/www/sta...cy Report.pdf' == '/var/www/sta...cy Report.pdf'
E - /var/www/statics\Fancy Report.pdf
E ? ^
E + /var/www/statics/Fancy Report.pdf
E ? ^
tests\test_static.py:184: AssertionError
_____________________ test_good_path[True-/some/download-/report.zip-/report.zip-application/zip] _____________________
asgi = True, uri_prefix = '/some/download', uri_path = '/report.zip', expected_path = '/report.zip'
mtype = 'application/zip', monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3BD3970>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
> assert resp.content_type == mtype
E AssertionError: assert 'application/x-zip-compressed' == 'application/zip'
E - application/x-zip-compressed
E + application/zip
tests\test_static.py:183: AssertionError
_________________ test_good_path[True-/some/download-/foo/../report.zip-/report.zip-application/zip] __________________
asgi = True, uri_prefix = '/some/download', uri_path = '/foo/../report.zip', expected_path = '/report.zip'
mtype = 'application/zip', monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3BD85B0>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
> assert resp.content_type == mtype
E AssertionError: assert 'application/x-zip-compressed' == 'application/zip'
E - application/x-zip-compressed
E + application/zip
tests\test_static.py:183: AssertionError
______________ test_good_path[True-/some/download-/foo/../bar/../report.zip-/report.zip-application/zip] ______________
asgi = True, uri_prefix = '/some/download', uri_path = '/foo/../bar/../report.zip', expected_path = '/report.zip'
mtype = 'application/zip', monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3B3A3D0>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
> assert resp.content_type == mtype
E AssertionError: assert 'application/x-zip-compressed' == 'application/zip'
E - application/x-zip-compressed
E + application/zip
tests\test_static.py:183: AssertionError
______________ test_good_path[True-/some/download-/foo/bar/../../report.zip-/report.zip-application/zip] ______________
asgi = True, uri_prefix = '/some/download', uri_path = '/foo/bar/../../report.zip', expected_path = '/report.zip'
mtype = 'application/zip', monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3B4E880>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
> assert resp.content_type == mtype
E AssertionError: assert 'application/x-zip-compressed' == 'application/zip'
E - application/x-zip-compressed
E + application/zip
tests\test_static.py:183: AssertionError
_________________________ test_good_path[False-/static/-/css/test.css-/css/test.css-text/css] _________________________
asgi = False, uri_prefix = '/static/', uri_path = '/css/test.css', expected_path = '/css/test.css', mtype = 'text/css'
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3BF0340>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
assert resp.content_type == mtype
> assert body.decode() == '/var/www/statics' + expected_path
E AssertionError: assert '/var/www/sta...css\\test.css' == '/var/www/sta.../css/test.css'
E - /var/www/statics\css\test.css
E ? ^ ^
E + /var/www/statics/css/test.css
E ? ^ ^
tests\test_static.py:184: AssertionError
_________________________ test_good_path[False-/static-/css/test.css-/css/test.css-text/css] __________________________
asgi = False, uri_prefix = '/static', uri_path = '/css/test.css', expected_path = '/css/test.css', mtype = 'text/css'
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F1900790>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
assert resp.content_type == mtype
> assert body.decode() == '/var/www/statics' + expected_path
E AssertionError: assert '/var/www/sta...css\\test.css' == '/var/www/sta.../css/test.css'
E - /var/www/statics\css\test.css
E ? ^ ^
E + /var/www/statics/css/test.css
E ? ^ ^
tests\test_static.py:184: AssertionError
_ test_good_path[False-/static-/tttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttt-/tttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttt-application/octet-stream] _
asgi = False, uri_prefix = '/static'
uri_path = '/tttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttt...tttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttt'
expected_path = '/tttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttt...tttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttt'
mtype = 'application/octet-stream', monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F1B97C40>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
assert resp.content_type == mtype
> assert body.decode() == '/var/www/statics' + expected_path
E AssertionError: assert '/var/www/sta...ttttttttttttt' == '/var/www/sta...ttttttttttttt'
E Skipping 503 identical trailing characters in diff, use -v to show
E - /var/www/statics\ttttttttt
E ? ^
E + /var/www/statics/ttttttttt
E ? ^
tests\test_static.py:184: AssertionError
____________________________ test_good_path[False-/static-/.test.css-/.test.css-text/css] _____________________________
asgi = False, uri_prefix = '/static', uri_path = '/.test.css', expected_path = '/.test.css', mtype = 'text/css'
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3840C70>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
assert resp.content_type == mtype
> assert body.decode() == '/var/www/statics' + expected_path
E AssertionError: assert '/var/www/statics\\.test.css' == '/var/www/statics/.test.css'
E - /var/www/statics\.test.css
E ? ^
E + /var/www/statics/.test.css
E ? ^
tests\test_static.py:184: AssertionError
____________________ test_good_path[False-/some/download/-/report.pdf-/report.pdf-application/pdf] ____________________
asgi = False, uri_prefix = '/some/download/', uri_path = '/report.pdf', expected_path = '/report.pdf'
mtype = 'application/pdf', monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3907460>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
assert resp.content_type == mtype
> assert body.decode() == '/var/www/statics' + expected_path
E AssertionError: assert '/var/www/statics\\report.pdf' == '/var/www/statics/report.pdf'
E - /var/www/statics\report.pdf
E ? ^
E + /var/www/statics/report.pdf
E ? ^
tests\test_static.py:184: AssertionError
______________ test_good_path[False-/some/download/-/Fancy Report.pdf-/Fancy Report.pdf-application/pdf] ______________
asgi = False, uri_prefix = '/some/download/', uri_path = '/Fancy Report.pdf', expected_path = '/Fancy Report.pdf'
mtype = 'application/pdf', monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F2452850>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
assert resp.content_type == mtype
> assert body.decode() == '/var/www/statics' + expected_path
E AssertionError: assert '/var/www/sta...cy Report.pdf' == '/var/www/sta...cy Report.pdf'
E - /var/www/statics\Fancy Report.pdf
E ? ^
E + /var/www/statics/Fancy Report.pdf
E ? ^
tests\test_static.py:184: AssertionError
____________________ test_good_path[False-/some/download-/report.zip-/report.zip-application/zip] _____________________
asgi = False, uri_prefix = '/some/download', uri_path = '/report.zip', expected_path = '/report.zip'
mtype = 'application/zip', monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F38D8EE0>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
> assert resp.content_type == mtype
E AssertionError: assert 'application/x-zip-compressed' == 'application/zip'
E - application/x-zip-compressed
E + application/zip
tests\test_static.py:183: AssertionError
_________________ test_good_path[False-/some/download-/foo/../report.zip-/report.zip-application/zip] _________________
asgi = False, uri_prefix = '/some/download', uri_path = '/foo/../report.zip', expected_path = '/report.zip'
mtype = 'application/zip', monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F399A970>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
> assert resp.content_type == mtype
E AssertionError: assert 'application/x-zip-compressed' == 'application/zip'
E - application/x-zip-compressed
E + application/zip
tests\test_static.py:183: AssertionError
_____________ test_good_path[False-/some/download-/foo/../bar/../report.zip-/report.zip-application/zip] ______________
asgi = False, uri_prefix = '/some/download', uri_path = '/foo/../bar/../report.zip', expected_path = '/report.zip'
mtype = 'application/zip', monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F39711F0>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
> assert resp.content_type == mtype
E AssertionError: assert 'application/x-zip-compressed' == 'application/zip'
E - application/x-zip-compressed
E + application/zip
tests\test_static.py:183: AssertionError
_____________ test_good_path[False-/some/download-/foo/bar/../../report.zip-/report.zip-application/zip] ______________
asgi = False, uri_prefix = '/some/download', uri_path = '/foo/bar/../../report.zip', expected_path = '/report.zip'
mtype = 'application/zip', monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F38B1EE0>
@pytest.mark.parametrize('uri_prefix, uri_path, expected_path, mtype', [
('/static/', '/css/test.css', '/css/test.css', 'text/css'),
('/static', '/css/test.css', '/css/test.css', 'text/css'),
(
'/static',
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'/' + ('t' * StaticRoute._MAX_NON_PREFIXED_LEN),
'application/octet-stream',
),
('/static', '/.test.css', '/.test.css', 'text/css'),
('/some/download/', '/report.pdf', '/report.pdf', 'application/pdf'),
('/some/download/', '/Fancy Report.pdf', '/Fancy Report.pdf', 'application/pdf'),
('/some/download', '/report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/../bar/../report.zip', '/report.zip', 'application/zip'),
('/some/download', '/foo/bar/../../report.zip', '/report.zip', 'application/zip'),
])
def test_good_path(asgi, uri_prefix, uri_path, expected_path, mtype, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
sr = create_sr(asgi, uri_prefix, '/var/www/statics')
req_path = uri_prefix[:-1] if uri_prefix.endswith('/') else uri_prefix
req_path += uri_path
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
> assert resp.content_type == mtype
E AssertionError: assert 'application/x-zip-compressed' == 'application/zip'
E - application/x-zip-compressed
E + application/zip
tests\test_static.py:183: AssertionError
___________________________________________________ test_lifo[True] ___________________________________________________
client = <falcon.testing.client.TestClient object at 0x000002A2F39B9CA0>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3907490>
def test_lifo(client, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
client.app.add_static_route('/downloads', '/opt/somesite/downloads')
client.app.add_static_route('/downloads/archive', '/opt/somesite/x')
response = client.simulate_request(path='/downloads/thing.zip')
assert response.status == falcon.HTTP_200
> assert response.text == '/opt/somesite/downloads/thing.zip'
E AssertionError: assert '/opt/somesit...ds\\thing.zip' == '/opt/somesit...ads/thing.zip'
E - /opt/somesite/downloads\thing.zip
E ? ^
E + /opt/somesite/downloads/thing.zip
E ? ^
tests\test_static.py:195: AssertionError
__________________________________________________ test_lifo[False] ___________________________________________________
client = <falcon.testing.client.TestClient object at 0x000002A2F3979790>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3A28730>
def test_lifo(client, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
client.app.add_static_route('/downloads', '/opt/somesite/downloads')
client.app.add_static_route('/downloads/archive', '/opt/somesite/x')
response = client.simulate_request(path='/downloads/thing.zip')
assert response.status == falcon.HTTP_200
> assert response.text == '/opt/somesite/downloads/thing.zip'
E AssertionError: assert '/opt/somesit...ds\\thing.zip' == '/opt/somesit...ads/thing.zip'
E - /opt/somesite/downloads\thing.zip
E ? ^
E + /opt/somesite/downloads/thing.zip
E ? ^
tests\test_static.py:195: AssertionError
______________________________________________ test_lifo_negative[True] _______________________________________________
client = <falcon.testing.client.TestClient object at 0x000002A2F38E8FA0>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F38C1E80>
def test_lifo_negative(client, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
client.app.add_static_route('/downloads/archive', '/opt/somesite/x')
client.app.add_static_route('/downloads', '/opt/somesite/downloads')
response = client.simulate_request(path='/downloads/thing.zip')
assert response.status == falcon.HTTP_200
> assert response.text == '/opt/somesite/downloads/thing.zip'
E AssertionError: assert '/opt/somesit...ds\\thing.zip' == '/opt/somesit...ads/thing.zip'
E - /opt/somesite/downloads\thing.zip
E ? ^
E + /opt/somesite/downloads/thing.zip
E ? ^
tests\test_static.py:210: AssertionError
______________________________________________ test_lifo_negative[False] ______________________________________________
client = <falcon.testing.client.TestClient object at 0x000002A2F38E3880>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F399AE50>
def test_lifo_negative(client, monkeypatch):
monkeypatch.setattr(io, 'open', lambda path, mode: io.BytesIO(path.encode()))
client.app.add_static_route('/downloads/archive', '/opt/somesite/x')
client.app.add_static_route('/downloads', '/opt/somesite/downloads')
response = client.simulate_request(path='/downloads/thing.zip')
assert response.status == falcon.HTTP_200
> assert response.text == '/opt/somesite/downloads/thing.zip'
E AssertionError: assert '/opt/somesit...ds\\thing.zip' == '/opt/somesit...ads/thing.zip'
E - /opt/somesite/downloads\thing.zip
E ? ^
E + /opt/somesite/downloads/thing.zip
E ? ^
tests\test_static.py:210: AssertionError
_____ test_fallback_filename[True-True-index.html_files/test.txt-index.html-index.html_files/test.txt-text/plain] _____
asgi = True, uri = 'index.html_files/test.txt', default = 'index.html', expected = 'index.html_files/test.txt'
content_type = 'text/plain', downloadable = True
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F390FD60>
@pytest.mark.parametrize('uri, default, expected, content_type', [
('', 'default', 'default', 'application/octet-stream'),
('other', 'default.html', 'default.html', 'text/html'),
('index2', 'index', 'index2', 'application/octet-stream'),
('absolute', '/foo/bar/index', '/foo/bar/index', 'application/octet-stream'),
('docs/notes/test.txt', 'index.html', 'index.html', 'text/html'),
('index.html_files/test.txt', 'index.html', 'index.html_files/test.txt', 'text/plain'),
])
@pytest.mark.parametrize('downloadable', [True, False])
def test_fallback_filename(asgi, uri, default, expected, content_type, downloadable,
monkeypatch):
def mock_open(path, mode):
if default in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mock_open)
monkeypatch.setattr('os.path.isfile', lambda file: default in file)
sr = create_sr(
asgi,
'/static',
'/var/www/statics',
downloadable=downloadable,
fallback_filename=default
)
req_path = '/static/' + uri
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
assert sr.match(req.path)
> assert body.decode() == os.path.join('/var/www/statics', expected)
E AssertionError: assert '/var/www/sta...les\\test.txt' == '/var/www/sta...iles/test.txt'
E - /var/www/statics\index.html_files\test.txt
E ? ^
E + /var/www/statics\index.html_files/test.txt
E ? ^
tests\test_static.py:293: AssertionError
____ test_fallback_filename[True-False-index.html_files/test.txt-index.html-index.html_files/test.txt-text/plain] _____
asgi = True, uri = 'index.html_files/test.txt', default = 'index.html', expected = 'index.html_files/test.txt'
content_type = 'text/plain', downloadable = False
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F251D2B0>
@pytest.mark.parametrize('uri, default, expected, content_type', [
('', 'default', 'default', 'application/octet-stream'),
('other', 'default.html', 'default.html', 'text/html'),
('index2', 'index', 'index2', 'application/octet-stream'),
('absolute', '/foo/bar/index', '/foo/bar/index', 'application/octet-stream'),
('docs/notes/test.txt', 'index.html', 'index.html', 'text/html'),
('index.html_files/test.txt', 'index.html', 'index.html_files/test.txt', 'text/plain'),
])
@pytest.mark.parametrize('downloadable', [True, False])
def test_fallback_filename(asgi, uri, default, expected, content_type, downloadable,
monkeypatch):
def mock_open(path, mode):
if default in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mock_open)
monkeypatch.setattr('os.path.isfile', lambda file: default in file)
sr = create_sr(
asgi,
'/static',
'/var/www/statics',
downloadable=downloadable,
fallback_filename=default
)
req_path = '/static/' + uri
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
assert sr.match(req.path)
> assert body.decode() == os.path.join('/var/www/statics', expected)
E AssertionError: assert '/var/www/sta...les\\test.txt' == '/var/www/sta...iles/test.txt'
E - /var/www/statics\index.html_files\test.txt
E ? ^
E + /var/www/statics\index.html_files/test.txt
E ? ^
tests\test_static.py:293: AssertionError
____ test_fallback_filename[False-True-index.html_files/test.txt-index.html-index.html_files/test.txt-text/plain] _____
asgi = False, uri = 'index.html_files/test.txt', default = 'index.html', expected = 'index.html_files/test.txt'
content_type = 'text/plain', downloadable = True
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F38DF6D0>
@pytest.mark.parametrize('uri, default, expected, content_type', [
('', 'default', 'default', 'application/octet-stream'),
('other', 'default.html', 'default.html', 'text/html'),
('index2', 'index', 'index2', 'application/octet-stream'),
('absolute', '/foo/bar/index', '/foo/bar/index', 'application/octet-stream'),
('docs/notes/test.txt', 'index.html', 'index.html', 'text/html'),
('index.html_files/test.txt', 'index.html', 'index.html_files/test.txt', 'text/plain'),
])
@pytest.mark.parametrize('downloadable', [True, False])
def test_fallback_filename(asgi, uri, default, expected, content_type, downloadable,
monkeypatch):
def mock_open(path, mode):
if default in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mock_open)
monkeypatch.setattr('os.path.isfile', lambda file: default in file)
sr = create_sr(
asgi,
'/static',
'/var/www/statics',
downloadable=downloadable,
fallback_filename=default
)
req_path = '/static/' + uri
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
assert sr.match(req.path)
> assert body.decode() == os.path.join('/var/www/statics', expected)
E AssertionError: assert '/var/www/sta...les\\test.txt' == '/var/www/sta...iles/test.txt'
E - /var/www/statics\index.html_files\test.txt
E ? ^
E + /var/www/statics\index.html_files/test.txt
E ? ^
tests\test_static.py:293: AssertionError
____ test_fallback_filename[False-False-index.html_files/test.txt-index.html-index.html_files/test.txt-text/plain] ____
asgi = False, uri = 'index.html_files/test.txt', default = 'index.html', expected = 'index.html_files/test.txt'
content_type = 'text/plain', downloadable = False
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F10FFF40>
@pytest.mark.parametrize('uri, default, expected, content_type', [
('', 'default', 'default', 'application/octet-stream'),
('other', 'default.html', 'default.html', 'text/html'),
('index2', 'index', 'index2', 'application/octet-stream'),
('absolute', '/foo/bar/index', '/foo/bar/index', 'application/octet-stream'),
('docs/notes/test.txt', 'index.html', 'index.html', 'text/html'),
('index.html_files/test.txt', 'index.html', 'index.html_files/test.txt', 'text/plain'),
])
@pytest.mark.parametrize('downloadable', [True, False])
def test_fallback_filename(asgi, uri, default, expected, content_type, downloadable,
monkeypatch):
def mock_open(path, mode):
if default in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mock_open)
monkeypatch.setattr('os.path.isfile', lambda file: default in file)
sr = create_sr(
asgi,
'/static',
'/var/www/statics',
downloadable=downloadable,
fallback_filename=default
)
req_path = '/static/' + uri
req = _util.create_req(
asgi,
host='test.com',
path=req_path,
root_path='statics'
)
resp = _util.create_resp(asgi)
if asgi:
async def run():
await sr(req, resp)
return await resp.stream.read()
body = testing.invoke_coroutine_sync(run)
else:
sr(req, resp)
body = resp.stream.read()
assert sr.match(req.path)
> assert body.decode() == os.path.join('/var/www/statics', expected)
E AssertionError: assert '/var/www/sta...les\\test.txt' == '/var/www/sta...iles/test.txt'
E - /var/www/statics\index.html_files\test.txt
E ? ^
E + /var/www/statics\index.html_files/test.txt
E ? ^
tests\test_static.py:293: AssertionError
_________________________ test_e2e_fallback_filename[True-/index-index.html-index-index-True] _________________________
client = <falcon.testing.client.TestClient object at 0x000002A2F39B0D90>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F384C040>, strip_slash = True, path = '/index'
fallback = 'index.html', static_exp = 'index', assert_axp = 'index'
@pytest.mark.parametrize('strip_slash', [True, False])
@pytest.mark.parametrize('path, fallback, static_exp, assert_axp', [
('/index', 'index.html', 'index', 'index'),
('', 'index.html', 'index.html', None),
('/', 'index.html', 'index.html', None),
('/other', 'index.html', 'index.html', None),
('/other', 'index.raise', None, None)
])
def test_e2e_fallback_filename(client, monkeypatch, strip_slash, path, fallback,
static_exp, assert_axp):
def mockOpen(path, mode):
if 'index' in path and 'raise' not in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mockOpen)
monkeypatch.setattr('os.path.isfile', lambda file: 'index' in file)
client.app.req_options.strip_url_path_trailing_slash = strip_slash
client.app.add_static_route('/static', '/opt/somesite/static',
fallback_filename=fallback)
client.app.add_static_route('/assets/', '/opt/somesite/assets')
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
assert response.text == directory + expected
> test('/static', '/opt/somesite/static/', static_exp)
tests\test_static.py:334:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
prefix = '/static', directory = '/opt/somesite/static/', expected = 'index'
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
> assert response.text == directory + expected
E AssertionError: assert '/opt/somesite/static\\index' == '/opt/somesite/static/index'
E - /opt/somesite/static\index
E ? ^
E + /opt/somesite/static/index
E ? ^
tests\test_static.py:332: AssertionError
________________________ test_e2e_fallback_filename[True-/index-index.html-index-index-False] _________________________
client = <falcon.testing.client.TestClient object at 0x000002A2F3AB80A0>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3B21E20>, strip_slash = False, path = '/index'
fallback = 'index.html', static_exp = 'index', assert_axp = 'index'
@pytest.mark.parametrize('strip_slash', [True, False])
@pytest.mark.parametrize('path, fallback, static_exp, assert_axp', [
('/index', 'index.html', 'index', 'index'),
('', 'index.html', 'index.html', None),
('/', 'index.html', 'index.html', None),
('/other', 'index.html', 'index.html', None),
('/other', 'index.raise', None, None)
])
def test_e2e_fallback_filename(client, monkeypatch, strip_slash, path, fallback,
static_exp, assert_axp):
def mockOpen(path, mode):
if 'index' in path and 'raise' not in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mockOpen)
monkeypatch.setattr('os.path.isfile', lambda file: 'index' in file)
client.app.req_options.strip_url_path_trailing_slash = strip_slash
client.app.add_static_route('/static', '/opt/somesite/static',
fallback_filename=fallback)
client.app.add_static_route('/assets/', '/opt/somesite/assets')
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
assert response.text == directory + expected
> test('/static', '/opt/somesite/static/', static_exp)
tests\test_static.py:334:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
prefix = '/static', directory = '/opt/somesite/static/', expected = 'index'
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
> assert response.text == directory + expected
E AssertionError: assert '/opt/somesite/static\\index' == '/opt/somesite/static/index'
E - /opt/somesite/static\index
E ? ^
E + /opt/somesite/static/index
E ? ^
tests\test_static.py:332: AssertionError
__________________________ test_e2e_fallback_filename[True--index.html-index.html-None-True] __________________________
client = <falcon.testing.client.TestClient object at 0x000002A2F3AB8BB0>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F39334F0>, strip_slash = True, path = ''
fallback = 'index.html', static_exp = 'index.html', assert_axp = None
@pytest.mark.parametrize('strip_slash', [True, False])
@pytest.mark.parametrize('path, fallback, static_exp, assert_axp', [
('/index', 'index.html', 'index', 'index'),
('', 'index.html', 'index.html', None),
('/', 'index.html', 'index.html', None),
('/other', 'index.html', 'index.html', None),
('/other', 'index.raise', None, None)
])
def test_e2e_fallback_filename(client, monkeypatch, strip_slash, path, fallback,
static_exp, assert_axp):
def mockOpen(path, mode):
if 'index' in path and 'raise' not in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mockOpen)
monkeypatch.setattr('os.path.isfile', lambda file: 'index' in file)
client.app.req_options.strip_url_path_trailing_slash = strip_slash
client.app.add_static_route('/static', '/opt/somesite/static',
fallback_filename=fallback)
client.app.add_static_route('/assets/', '/opt/somesite/assets')
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
assert response.text == directory + expected
> test('/static', '/opt/somesite/static/', static_exp)
tests\test_static.py:334:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
prefix = '/static', directory = '/opt/somesite/static/', expected = 'index.html'
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
> assert response.text == directory + expected
E AssertionError: assert '/opt/somesit...c\\index.html' == '/opt/somesit...ic/index.html'
E - /opt/somesite/static\index.html
E ? ^
E + /opt/somesite/static/index.html
E ? ^
tests\test_static.py:332: AssertionError
_________________________ test_e2e_fallback_filename[True--index.html-index.html-None-False] __________________________
client = <falcon.testing.client.TestClient object at 0x000002A2F3B4E1C0>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3983D00>, strip_slash = False, path = ''
fallback = 'index.html', static_exp = 'index.html', assert_axp = None
@pytest.mark.parametrize('strip_slash', [True, False])
@pytest.mark.parametrize('path, fallback, static_exp, assert_axp', [
('/index', 'index.html', 'index', 'index'),
('', 'index.html', 'index.html', None),
('/', 'index.html', 'index.html', None),
('/other', 'index.html', 'index.html', None),
('/other', 'index.raise', None, None)
])
def test_e2e_fallback_filename(client, monkeypatch, strip_slash, path, fallback,
static_exp, assert_axp):
def mockOpen(path, mode):
if 'index' in path and 'raise' not in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mockOpen)
monkeypatch.setattr('os.path.isfile', lambda file: 'index' in file)
client.app.req_options.strip_url_path_trailing_slash = strip_slash
client.app.add_static_route('/static', '/opt/somesite/static',
fallback_filename=fallback)
client.app.add_static_route('/assets/', '/opt/somesite/assets')
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
assert response.text == directory + expected
> test('/static', '/opt/somesite/static/', static_exp)
tests\test_static.py:334:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
prefix = '/static', directory = '/opt/somesite/static/', expected = 'index.html'
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
> assert response.text == directory + expected
E AssertionError: assert '/opt/somesit...c\\index.html' == '/opt/somesit...ic/index.html'
E - /opt/somesite/static\index.html
E ? ^
E + /opt/somesite/static/index.html
E ? ^
tests\test_static.py:332: AssertionError
_________________________ test_e2e_fallback_filename[True-/-index.html-index.html-None-True] __________________________
client = <falcon.testing.client.TestClient object at 0x000002A2F3B4ECD0>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F38D2190>, strip_slash = True, path = '/'
fallback = 'index.html', static_exp = 'index.html', assert_axp = None
@pytest.mark.parametrize('strip_slash', [True, False])
@pytest.mark.parametrize('path, fallback, static_exp, assert_axp', [
('/index', 'index.html', 'index', 'index'),
('', 'index.html', 'index.html', None),
('/', 'index.html', 'index.html', None),
('/other', 'index.html', 'index.html', None),
('/other', 'index.raise', None, None)
])
def test_e2e_fallback_filename(client, monkeypatch, strip_slash, path, fallback,
static_exp, assert_axp):
def mockOpen(path, mode):
if 'index' in path and 'raise' not in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mockOpen)
monkeypatch.setattr('os.path.isfile', lambda file: 'index' in file)
client.app.req_options.strip_url_path_trailing_slash = strip_slash
client.app.add_static_route('/static', '/opt/somesite/static',
fallback_filename=fallback)
client.app.add_static_route('/assets/', '/opt/somesite/assets')
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
assert response.text == directory + expected
> test('/static', '/opt/somesite/static/', static_exp)
tests\test_static.py:334:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
prefix = '/static', directory = '/opt/somesite/static/', expected = 'index.html'
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
> assert response.text == directory + expected
E AssertionError: assert '/opt/somesit...c\\index.html' == '/opt/somesit...ic/index.html'
E - /opt/somesite/static\index.html
E ? ^
E + /opt/somesite/static/index.html
E ? ^
tests\test_static.py:332: AssertionError
_________________________ test_e2e_fallback_filename[True-/-index.html-index.html-None-False] _________________________
client = <falcon.testing.client.TestClient object at 0x000002A2F3B21C70>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F26A4EE0>, strip_slash = False, path = '/'
fallback = 'index.html', static_exp = 'index.html', assert_axp = None
@pytest.mark.parametrize('strip_slash', [True, False])
@pytest.mark.parametrize('path, fallback, static_exp, assert_axp', [
('/index', 'index.html', 'index', 'index'),
('', 'index.html', 'index.html', None),
('/', 'index.html', 'index.html', None),
('/other', 'index.html', 'index.html', None),
('/other', 'index.raise', None, None)
])
def test_e2e_fallback_filename(client, monkeypatch, strip_slash, path, fallback,
static_exp, assert_axp):
def mockOpen(path, mode):
if 'index' in path and 'raise' not in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mockOpen)
monkeypatch.setattr('os.path.isfile', lambda file: 'index' in file)
client.app.req_options.strip_url_path_trailing_slash = strip_slash
client.app.add_static_route('/static', '/opt/somesite/static',
fallback_filename=fallback)
client.app.add_static_route('/assets/', '/opt/somesite/assets')
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
assert response.text == directory + expected
> test('/static', '/opt/somesite/static/', static_exp)
tests\test_static.py:334:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
prefix = '/static', directory = '/opt/somesite/static/', expected = 'index.html'
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
> assert response.text == directory + expected
E AssertionError: assert '/opt/somesit...c\\index.html' == '/opt/somesit...ic/index.html'
E - /opt/somesite/static\index.html
E ? ^
E + /opt/somesite/static/index.html
E ? ^
tests\test_static.py:332: AssertionError
_______________________ test_e2e_fallback_filename[True-/other-index.html-index.html-None-True] _______________________
client = <falcon.testing.client.TestClient object at 0x000002A2F39B0D90>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3B24CA0>, strip_slash = True, path = '/other'
fallback = 'index.html', static_exp = 'index.html', assert_axp = None
@pytest.mark.parametrize('strip_slash', [True, False])
@pytest.mark.parametrize('path, fallback, static_exp, assert_axp', [
('/index', 'index.html', 'index', 'index'),
('', 'index.html', 'index.html', None),
('/', 'index.html', 'index.html', None),
('/other', 'index.html', 'index.html', None),
('/other', 'index.raise', None, None)
])
def test_e2e_fallback_filename(client, monkeypatch, strip_slash, path, fallback,
static_exp, assert_axp):
def mockOpen(path, mode):
if 'index' in path and 'raise' not in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mockOpen)
monkeypatch.setattr('os.path.isfile', lambda file: 'index' in file)
client.app.req_options.strip_url_path_trailing_slash = strip_slash
client.app.add_static_route('/static', '/opt/somesite/static',
fallback_filename=fallback)
client.app.add_static_route('/assets/', '/opt/somesite/assets')
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
assert response.text == directory + expected
> test('/static', '/opt/somesite/static/', static_exp)
tests\test_static.py:334:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
prefix = '/static', directory = '/opt/somesite/static/', expected = 'index.html'
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
> assert response.text == directory + expected
E AssertionError: assert '/opt/somesit...c\\index.html' == '/opt/somesit...ic/index.html'
E - /opt/somesite/static\index.html
E ? ^
E + /opt/somesite/static/index.html
E ? ^
tests\test_static.py:332: AssertionError
______________________ test_e2e_fallback_filename[True-/other-index.html-index.html-None-False] _______________________
client = <falcon.testing.client.TestClient object at 0x000002A2F2577B20>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3A8D610>, strip_slash = False, path = '/other'
fallback = 'index.html', static_exp = 'index.html', assert_axp = None
@pytest.mark.parametrize('strip_slash', [True, False])
@pytest.mark.parametrize('path, fallback, static_exp, assert_axp', [
('/index', 'index.html', 'index', 'index'),
('', 'index.html', 'index.html', None),
('/', 'index.html', 'index.html', None),
('/other', 'index.html', 'index.html', None),
('/other', 'index.raise', None, None)
])
def test_e2e_fallback_filename(client, monkeypatch, strip_slash, path, fallback,
static_exp, assert_axp):
def mockOpen(path, mode):
if 'index' in path and 'raise' not in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mockOpen)
monkeypatch.setattr('os.path.isfile', lambda file: 'index' in file)
client.app.req_options.strip_url_path_trailing_slash = strip_slash
client.app.add_static_route('/static', '/opt/somesite/static',
fallback_filename=fallback)
client.app.add_static_route('/assets/', '/opt/somesite/assets')
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
assert response.text == directory + expected
> test('/static', '/opt/somesite/static/', static_exp)
tests\test_static.py:334:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
prefix = '/static', directory = '/opt/somesite/static/', expected = 'index.html'
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
> assert response.text == directory + expected
E AssertionError: assert '/opt/somesit...c\\index.html' == '/opt/somesit...ic/index.html'
E - /opt/somesite/static\index.html
E ? ^
E + /opt/somesite/static/index.html
E ? ^
tests\test_static.py:332: AssertionError
________________________ test_e2e_fallback_filename[False-/index-index.html-index-index-True] _________________________
client = <falcon.testing.client.TestClient object at 0x000002A2F3BD9940>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3BD92B0>, strip_slash = True, path = '/index'
fallback = 'index.html', static_exp = 'index', assert_axp = 'index'
@pytest.mark.parametrize('strip_slash', [True, False])
@pytest.mark.parametrize('path, fallback, static_exp, assert_axp', [
('/index', 'index.html', 'index', 'index'),
('', 'index.html', 'index.html', None),
('/', 'index.html', 'index.html', None),
('/other', 'index.html', 'index.html', None),
('/other', 'index.raise', None, None)
])
def test_e2e_fallback_filename(client, monkeypatch, strip_slash, path, fallback,
static_exp, assert_axp):
def mockOpen(path, mode):
if 'index' in path and 'raise' not in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mockOpen)
monkeypatch.setattr('os.path.isfile', lambda file: 'index' in file)
client.app.req_options.strip_url_path_trailing_slash = strip_slash
client.app.add_static_route('/static', '/opt/somesite/static',
fallback_filename=fallback)
client.app.add_static_route('/assets/', '/opt/somesite/assets')
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
assert response.text == directory + expected
> test('/static', '/opt/somesite/static/', static_exp)
tests\test_static.py:334:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
prefix = '/static', directory = '/opt/somesite/static/', expected = 'index'
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
> assert response.text == directory + expected
E AssertionError: assert '/opt/somesite/static\\index' == '/opt/somesite/static/index'
E - /opt/somesite/static\index
E ? ^
E + /opt/somesite/static/index
E ? ^
tests\test_static.py:332: AssertionError
________________________ test_e2e_fallback_filename[False-/index-index.html-index-index-False] ________________________
client = <falcon.testing.client.TestClient object at 0x000002A2F3A00730>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3A44E20>, strip_slash = False, path = '/index'
fallback = 'index.html', static_exp = 'index', assert_axp = 'index'
@pytest.mark.parametrize('strip_slash', [True, False])
@pytest.mark.parametrize('path, fallback, static_exp, assert_axp', [
('/index', 'index.html', 'index', 'index'),
('', 'index.html', 'index.html', None),
('/', 'index.html', 'index.html', None),
('/other', 'index.html', 'index.html', None),
('/other', 'index.raise', None, None)
])
def test_e2e_fallback_filename(client, monkeypatch, strip_slash, path, fallback,
static_exp, assert_axp):
def mockOpen(path, mode):
if 'index' in path and 'raise' not in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mockOpen)
monkeypatch.setattr('os.path.isfile', lambda file: 'index' in file)
client.app.req_options.strip_url_path_trailing_slash = strip_slash
client.app.add_static_route('/static', '/opt/somesite/static',
fallback_filename=fallback)
client.app.add_static_route('/assets/', '/opt/somesite/assets')
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
assert response.text == directory + expected
> test('/static', '/opt/somesite/static/', static_exp)
tests\test_static.py:334:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
prefix = '/static', directory = '/opt/somesite/static/', expected = 'index'
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
> assert response.text == directory + expected
E AssertionError: assert '/opt/somesite/static\\index' == '/opt/somesite/static/index'
E - /opt/somesite/static\index
E ? ^
E + /opt/somesite/static/index
E ? ^
tests\test_static.py:332: AssertionError
_________________________ test_e2e_fallback_filename[False--index.html-index.html-None-True] __________________________
client = <falcon.testing.client.TestClient object at 0x000002A2F3840C70>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F25DF910>, strip_slash = True, path = ''
fallback = 'index.html', static_exp = 'index.html', assert_axp = None
@pytest.mark.parametrize('strip_slash', [True, False])
@pytest.mark.parametrize('path, fallback, static_exp, assert_axp', [
('/index', 'index.html', 'index', 'index'),
('', 'index.html', 'index.html', None),
('/', 'index.html', 'index.html', None),
('/other', 'index.html', 'index.html', None),
('/other', 'index.raise', None, None)
])
def test_e2e_fallback_filename(client, monkeypatch, strip_slash, path, fallback,
static_exp, assert_axp):
def mockOpen(path, mode):
if 'index' in path and 'raise' not in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mockOpen)
monkeypatch.setattr('os.path.isfile', lambda file: 'index' in file)
client.app.req_options.strip_url_path_trailing_slash = strip_slash
client.app.add_static_route('/static', '/opt/somesite/static',
fallback_filename=fallback)
client.app.add_static_route('/assets/', '/opt/somesite/assets')
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
assert response.text == directory + expected
> test('/static', '/opt/somesite/static/', static_exp)
tests\test_static.py:334:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
prefix = '/static', directory = '/opt/somesite/static/', expected = 'index.html'
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
> assert response.text == directory + expected
E AssertionError: assert '/opt/somesit...c\\index.html' == '/opt/somesit...ic/index.html'
E - /opt/somesite/static\index.html
E ? ^
E + /opt/somesite/static/index.html
E ? ^
tests\test_static.py:332: AssertionError
_________________________ test_e2e_fallback_filename[False--index.html-index.html-None-False] _________________________
client = <falcon.testing.client.TestClient object at 0x000002A2F268CEB0>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F24F3F10>, strip_slash = False, path = ''
fallback = 'index.html', static_exp = 'index.html', assert_axp = None
@pytest.mark.parametrize('strip_slash', [True, False])
@pytest.mark.parametrize('path, fallback, static_exp, assert_axp', [
('/index', 'index.html', 'index', 'index'),
('', 'index.html', 'index.html', None),
('/', 'index.html', 'index.html', None),
('/other', 'index.html', 'index.html', None),
('/other', 'index.raise', None, None)
])
def test_e2e_fallback_filename(client, monkeypatch, strip_slash, path, fallback,
static_exp, assert_axp):
def mockOpen(path, mode):
if 'index' in path and 'raise' not in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mockOpen)
monkeypatch.setattr('os.path.isfile', lambda file: 'index' in file)
client.app.req_options.strip_url_path_trailing_slash = strip_slash
client.app.add_static_route('/static', '/opt/somesite/static',
fallback_filename=fallback)
client.app.add_static_route('/assets/', '/opt/somesite/assets')
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
assert response.text == directory + expected
> test('/static', '/opt/somesite/static/', static_exp)
tests\test_static.py:334:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
prefix = '/static', directory = '/opt/somesite/static/', expected = 'index.html'
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
> assert response.text == directory + expected
E AssertionError: assert '/opt/somesit...c\\index.html' == '/opt/somesit...ic/index.html'
E - /opt/somesite/static\index.html
E ? ^
E + /opt/somesite/static/index.html
E ? ^
tests\test_static.py:332: AssertionError
_________________________ test_e2e_fallback_filename[False-/-index.html-index.html-None-True] _________________________
client = <falcon.testing.client.TestClient object at 0x000002A2F3A44E20>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3858EB0>, strip_slash = True, path = '/'
fallback = 'index.html', static_exp = 'index.html', assert_axp = None
@pytest.mark.parametrize('strip_slash', [True, False])
@pytest.mark.parametrize('path, fallback, static_exp, assert_axp', [
('/index', 'index.html', 'index', 'index'),
('', 'index.html', 'index.html', None),
('/', 'index.html', 'index.html', None),
('/other', 'index.html', 'index.html', None),
('/other', 'index.raise', None, None)
])
def test_e2e_fallback_filename(client, monkeypatch, strip_slash, path, fallback,
static_exp, assert_axp):
def mockOpen(path, mode):
if 'index' in path and 'raise' not in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mockOpen)
monkeypatch.setattr('os.path.isfile', lambda file: 'index' in file)
client.app.req_options.strip_url_path_trailing_slash = strip_slash
client.app.add_static_route('/static', '/opt/somesite/static',
fallback_filename=fallback)
client.app.add_static_route('/assets/', '/opt/somesite/assets')
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
assert response.text == directory + expected
> test('/static', '/opt/somesite/static/', static_exp)
tests\test_static.py:334:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
prefix = '/static', directory = '/opt/somesite/static/', expected = 'index.html'
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
> assert response.text == directory + expected
E AssertionError: assert '/opt/somesit...c\\index.html' == '/opt/somesit...ic/index.html'
E - /opt/somesite/static\index.html
E ? ^
E + /opt/somesite/static/index.html
E ? ^
tests\test_static.py:332: AssertionError
________________________ test_e2e_fallback_filename[False-/-index.html-index.html-None-False] _________________________
client = <falcon.testing.client.TestClient object at 0x000002A2F3A32730>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3907A30>, strip_slash = False, path = '/'
fallback = 'index.html', static_exp = 'index.html', assert_axp = None
@pytest.mark.parametrize('strip_slash', [True, False])
@pytest.mark.parametrize('path, fallback, static_exp, assert_axp', [
('/index', 'index.html', 'index', 'index'),
('', 'index.html', 'index.html', None),
('/', 'index.html', 'index.html', None),
('/other', 'index.html', 'index.html', None),
('/other', 'index.raise', None, None)
])
def test_e2e_fallback_filename(client, monkeypatch, strip_slash, path, fallback,
static_exp, assert_axp):
def mockOpen(path, mode):
if 'index' in path and 'raise' not in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mockOpen)
monkeypatch.setattr('os.path.isfile', lambda file: 'index' in file)
client.app.req_options.strip_url_path_trailing_slash = strip_slash
client.app.add_static_route('/static', '/opt/somesite/static',
fallback_filename=fallback)
client.app.add_static_route('/assets/', '/opt/somesite/assets')
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
assert response.text == directory + expected
> test('/static', '/opt/somesite/static/', static_exp)
tests\test_static.py:334:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
prefix = '/static', directory = '/opt/somesite/static/', expected = 'index.html'
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
> assert response.text == directory + expected
E AssertionError: assert '/opt/somesit...c\\index.html' == '/opt/somesit...ic/index.html'
E - /opt/somesite/static\index.html
E ? ^
E + /opt/somesite/static/index.html
E ? ^
tests\test_static.py:332: AssertionError
______________________ test_e2e_fallback_filename[False-/other-index.html-index.html-None-True] _______________________
client = <falcon.testing.client.TestClient object at 0x000002A2F2559E20>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3BD90A0>, strip_slash = True, path = '/other'
fallback = 'index.html', static_exp = 'index.html', assert_axp = None
@pytest.mark.parametrize('strip_slash', [True, False])
@pytest.mark.parametrize('path, fallback, static_exp, assert_axp', [
('/index', 'index.html', 'index', 'index'),
('', 'index.html', 'index.html', None),
('/', 'index.html', 'index.html', None),
('/other', 'index.html', 'index.html', None),
('/other', 'index.raise', None, None)
])
def test_e2e_fallback_filename(client, monkeypatch, strip_slash, path, fallback,
static_exp, assert_axp):
def mockOpen(path, mode):
if 'index' in path and 'raise' not in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mockOpen)
monkeypatch.setattr('os.path.isfile', lambda file: 'index' in file)
client.app.req_options.strip_url_path_trailing_slash = strip_slash
client.app.add_static_route('/static', '/opt/somesite/static',
fallback_filename=fallback)
client.app.add_static_route('/assets/', '/opt/somesite/assets')
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
assert response.text == directory + expected
> test('/static', '/opt/somesite/static/', static_exp)
tests\test_static.py:334:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
prefix = '/static', directory = '/opt/somesite/static/', expected = 'index.html'
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
> assert response.text == directory + expected
E AssertionError: assert '/opt/somesit...c\\index.html' == '/opt/somesit...ic/index.html'
E - /opt/somesite/static\index.html
E ? ^
E + /opt/somesite/static/index.html
E ? ^
tests\test_static.py:332: AssertionError
______________________ test_e2e_fallback_filename[False-/other-index.html-index.html-None-False] ______________________
client = <falcon.testing.client.TestClient object at 0x000002A2F24F3100>
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x000002A2F3BD9790>, strip_slash = False, path = '/other'
fallback = 'index.html', static_exp = 'index.html', assert_axp = None
@pytest.mark.parametrize('strip_slash', [True, False])
@pytest.mark.parametrize('path, fallback, static_exp, assert_axp', [
('/index', 'index.html', 'index', 'index'),
('', 'index.html', 'index.html', None),
('/', 'index.html', 'index.html', None),
('/other', 'index.html', 'index.html', None),
('/other', 'index.raise', None, None)
])
def test_e2e_fallback_filename(client, monkeypatch, strip_slash, path, fallback,
static_exp, assert_axp):
def mockOpen(path, mode):
if 'index' in path and 'raise' not in path:
return io.BytesIO(path.encode())
raise IOError()
monkeypatch.setattr(io, 'open', mockOpen)
monkeypatch.setattr('os.path.isfile', lambda file: 'index' in file)
client.app.req_options.strip_url_path_trailing_slash = strip_slash
client.app.add_static_route('/static', '/opt/somesite/static',
fallback_filename=fallback)
client.app.add_static_route('/assets/', '/opt/somesite/assets')
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
assert response.text == directory + expected
> test('/static', '/opt/somesite/static/', static_exp)
tests\test_static.py:334:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
prefix = '/static', directory = '/opt/somesite/static/', expected = 'index.html'
def test(prefix, directory, expected):
response = client.simulate_request(path=prefix + path)
if expected is None:
assert response.status == falcon.HTTP_404
else:
assert response.status == falcon.HTTP_200
> assert response.text == directory + expected
E AssertionError: assert '/opt/somesit...c\\index.html' == '/opt/somesit...ic/index.html'
E - /opt/somesite/static\index.html
E ? ^
E + /opt/somesite/static/index.html
E ? ^
tests\test_static.py:332: AssertionError
______________________________________ TestASGIServer.test_put[_uvicorn_factory] ______________________________________
self = <urllib3.connectionpool.HTTPConnectionPool object at 0x000002A2F3D15C10>
conn = <urllib3.connection.HTTPConnection object at 0x000002A2F2022D60>, method = 'PUT', url = '/'
timeout = <urllib3.util.timeout.Timeout object at 0x000002A2F2022220>, chunked = False
httplib_request_kw = {'body': '{}', 'headers': {'User-Agent': 'python-requests/2.22.0', 'Accept-Encoding': 'gzip, deflate', 'Accept': '*/*', 'Connection': 'keep-alive', 'Content-Length': '2'}}
timeout_obj = <urllib3.util.timeout.Timeout object at 0x000002A2F2022E20>, read_timeout = 10
def _make_request(
self, conn, method, url, timeout=_Default, chunked=False, **httplib_request_kw
):
"""
Perform a request on a given urllib connection object taken from our
pool.
:param conn:
a connection from one of our connection pools
:param timeout:
Socket timeout in seconds for the request. This can be a
float or integer, which will set the same timeout value for
the socket connect and the socket read, or an instance of
:class:`urllib3.util.Timeout`, which gives you more fine-grained
control over your timeouts.
"""
self.num_requests += 1
timeout_obj = self._get_timeout(timeout)
timeout_obj.start_connect()
conn.timeout = timeout_obj.connect_timeout
# Trigger any extra validation we need to do.
try:
self._validate_conn(conn)
except (SocketTimeout, BaseSSLError) as e:
# Py2 raises this as a BaseSSLError, Py3 raises it as socket timeout.
self._raise_timeout(err=e, url=url, timeout_value=conn.timeout)
raise
# conn.request() calls httplib.*.request, not the method in
# urllib3.request. It also calls makefile (recv) on the socket.
if chunked:
conn.request_chunked(method, url, **httplib_request_kw)
else:
conn.request(method, url, **httplib_request_kw)
# Reset the timeout for the recv() on the socket
read_timeout = timeout_obj.read_timeout
# App Engine doesn't have a sock attr
if getattr(conn, "sock", None):
# In Python 3 socket.py will catch EAGAIN and return None when you
# try and read into the file pointer created by http.client, which
# instead raises a BadStatusLine exception. Instead of catching
# the exception and assuming all BadStatusLine exceptions are read
# timeouts, check for a zero timeout before making the request.
if read_timeout == 0:
raise ReadTimeoutError(
self, url, "Read timed out. (read timeout=%s)" % read_timeout
)
if read_timeout is Timeout.DEFAULT_TIMEOUT:
conn.sock.settimeout(socket.getdefaulttimeout())
else: # None or a value
conn.sock.settimeout(read_timeout)
# Receive the response from the server
try:
try:
# Python 2.7, use buffering of HTTP responses
httplib_response = conn.getresponse(buffering=True)
except TypeError:
# Python 3
try:
httplib_response = conn.getresponse()
except BaseException as e:
# Remove the TypeError from the exception chain in
# Python 3 (including for exceptions like SystemExit).
# Otherwise it looks like a bug in the code.
> six.raise_from(e, None)
..\..\..\miniconda3\envs\falcon\lib\site-packages\urllib3\connectionpool.py:421:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
value = None, from_value = None
> ???
<string>:3:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <urllib3.connectionpool.HTTPConnectionPool object at 0x000002A2F3D15C10>
conn = <urllib3.connection.HTTPConnection object at 0x000002A2F2022D60>, method = 'PUT', url = '/'
timeout = <urllib3.util.timeout.Timeout object at 0x000002A2F2022220>, chunked = False
httplib_request_kw = {'body': '{}', 'headers': {'User-Agent': 'python-requests/2.22.0', 'Accept-Encoding': 'gzip, deflate', 'Accept': '*/*', 'Connection': 'keep-alive', 'Content-Length': '2'}}
timeout_obj = <urllib3.util.timeout.Timeout object at 0x000002A2F2022E20>, read_timeout = 10
def _make_request(
self, conn, method, url, timeout=_Default, chunked=False, **httplib_request_kw
):
"""
Perform a request on a given urllib connection object taken from our
pool.
:param conn:
a connection from one of our connection pools
:param timeout:
Socket timeout in seconds for the request. This can be a
float or integer, which will set the same timeout value for
the socket connect and the socket read, or an instance of
:class:`urllib3.util.Timeout`, which gives you more fine-grained
control over your timeouts.
"""
self.num_requests += 1
timeout_obj = self._get_timeout(timeout)
timeout_obj.start_connect()
conn.timeout = timeout_obj.connect_timeout
# Trigger any extra validation we need to do.
try:
self._validate_conn(conn)
except (SocketTimeout, BaseSSLError) as e:
# Py2 raises this as a BaseSSLError, Py3 raises it as socket timeout.
self._raise_timeout(err=e, url=url, timeout_value=conn.timeout)
raise
# conn.request() calls httplib.*.request, not the method in
# urllib3.request. It also calls makefile (recv) on the socket.
if chunked:
conn.request_chunked(method, url, **httplib_request_kw)
else:
conn.request(method, url, **httplib_request_kw)
# Reset the timeout for the recv() on the socket
read_timeout = timeout_obj.read_timeout
# App Engine doesn't have a sock attr
if getattr(conn, "sock", None):
# In Python 3 socket.py will catch EAGAIN and return None when you
# try and read into the file pointer created by http.client, which
# instead raises a BadStatusLine exception. Instead of catching
# the exception and assuming all BadStatusLine exceptions are read
# timeouts, check for a zero timeout before making the request.
if read_timeout == 0:
raise ReadTimeoutError(
self, url, "Read timed out. (read timeout=%s)" % read_timeout
)
if read_timeout is Timeout.DEFAULT_TIMEOUT:
conn.sock.settimeout(socket.getdefaulttimeout())
else: # None or a value
conn.sock.settimeout(read_timeout)
# Receive the response from the server
try:
try:
# Python 2.7, use buffering of HTTP responses
httplib_response = conn.getresponse(buffering=True)
except TypeError:
# Python 3
try:
> httplib_response = conn.getresponse()
..\..\..\miniconda3\envs\falcon\lib\site-packages\urllib3\connectionpool.py:416:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <urllib3.connection.HTTPConnection object at 0x000002A2F2022D60>
def getresponse(self):
"""Get the response from the server.
If the HTTPConnection is in the correct state, returns an
instance of HTTPResponse or of whatever object is returned by
the response_class variable.
If a request has not been sent or if a previous response has
not be handled, ResponseNotReady is raised. If the HTTP
response indicates that the connection should be closed, then
it will be closed before the response is returned. When the
connection is closed, the underlying socket is closed.
"""
# if a prior response has been completed, then forget about it.
if self.__response and self.__response.isclosed():
self.__response = None
# if a prior response exists, then it must be completed (otherwise, we
# cannot read this response's header to determine the connection-close
# behavior)
#
# note: if a prior response existed, but was connection-close, then the
# socket and response were made independent of this HTTPConnection
# object since a new request requires that we open a whole new
# connection
#
# this means the prior response had one of two states:
# 1) will_close: this connection was reset and the prior socket and
# response operate independently
# 2) persistent: the response was retained and we await its
# isclosed() status to become true.
#
if self.__state != _CS_REQ_SENT or self.__response:
raise ResponseNotReady(self.__state)
if self.debuglevel > 0:
response = self.response_class(self.sock, self.debuglevel,
method=self._method)
else:
response = self.response_class(self.sock, method=self._method)
try:
try:
> response.begin()
..\..\..\miniconda3\envs\falcon\lib\http\client.py:1322:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <http.client.HTTPResponse object at 0x000002A2F3D22370>
def begin(self):
if self.headers is not None:
# we've already started reading the response
return
# read until we get a non-100 response
while True:
> version, status, reason = self._read_status()
..\..\..\miniconda3\envs\falcon\lib\http\client.py:303:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <http.client.HTTPResponse object at 0x000002A2F3D22370>
def _read_status(self):
> line = str(self.fp.readline(_MAXLINE + 1), "iso-8859-1")
..\..\..\miniconda3\envs\falcon\lib\http\client.py:264:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <socket.SocketIO object at 0x000002A2F3D22D30>, b = <memory at 0x000002A2F3DA6940>
def readinto(self, b):
"""Read up to len(b) bytes into the writable buffer *b* and return
the number of bytes read. If the socket is non-blocking and no bytes
are available, None is returned.
If *b* is non-empty, a 0 return value indicates that the connection
was shutdown at the other end.
"""
self._checkClosed()
self._checkReadable()
if self._timeout_occurred:
raise OSError("cannot read from timed out object")
while True:
try:
> return self._sock.recv_into(b)
E socket.timeout: timed out
..\..\..\miniconda3\envs\falcon\lib\socket.py:669: timeout
During handling of the above exception, another exception occurred:
self = <requests.adapters.HTTPAdapter object at 0x000002A2F3D70040>, request = <PreparedRequest [PUT]>, stream = False
timeout = <urllib3.util.timeout.Timeout object at 0x000002A2F3D152E0>, verify = True, cert = None
proxies = OrderedDict()
def send(self, request, stream=False, timeout=None, verify=True, cert=None, proxies=None):
"""Sends PreparedRequest object. Returns Response object.
:param request: The :class:`PreparedRequest <PreparedRequest>` being sent.
:param stream: (optional) Whether to stream the request content.
:param timeout: (optional) How long to wait for the server to send
data before giving up, as a float, or a :ref:`(connect timeout,
read timeout) <timeouts>` tuple.
:type timeout: float or tuple or urllib3 Timeout object
:param verify: (optional) Either a boolean, in which case it controls whether
we verify the server's TLS certificate, or a string, in which case it
must be a path to a CA bundle to use
:param cert: (optional) Any user-provided SSL certificate to be trusted.
:param proxies: (optional) The proxies dictionary to apply to the request.
:rtype: requests.Response
"""
try:
conn = self.get_connection(request.url, proxies)
except LocationValueError as e:
raise InvalidURL(e, request=request)
self.cert_verify(conn, request.url, verify, cert)
url = self.request_url(request, proxies)
self.add_headers(request, stream=stream, timeout=timeout, verify=verify, cert=cert, proxies=proxies)
chunked = not (request.body is None or 'Content-Length' in request.headers)
if isinstance(timeout, tuple):
try:
connect, read = timeout
timeout = TimeoutSauce(connect=connect, read=read)
except ValueError as e:
# this may raise a string formatting error.
err = ("Invalid timeout {}. Pass a (connect, read) "
"timeout tuple, or a single float to set "
"both timeouts to the same value".format(timeout))
raise ValueError(err)
elif isinstance(timeout, TimeoutSauce):
pass
else:
timeout = TimeoutSauce(connect=timeout, read=timeout)
try:
if not chunked:
> resp = conn.urlopen(
method=request.method,
url=url,
body=request.body,
headers=request.headers,
redirect=False,
assert_same_host=False,
preload_content=False,
decode_content=False,
retries=self.max_retries,
timeout=timeout
)
..\..\..\miniconda3\envs\falcon\lib\site-packages\requests\adapters.py:439:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <urllib3.connectionpool.HTTPConnectionPool object at 0x000002A2F3D15C10>, method = 'PUT', url = '/', body = '{}'
headers = {'User-Agent': 'python-requests/2.22.0', 'Accept-Encoding': 'gzip, deflate', 'Accept': '*/*', 'Connection': 'keep-alive', 'Content-Length': '2'}
retries = Retry(total=0, connect=None, read=False, redirect=None, status=None), redirect = False
assert_same_host = False, timeout = <urllib3.util.timeout.Timeout object at 0x000002A2F3D152E0>, pool_timeout = None
release_conn = False, chunked = False, body_pos = None
response_kw = {'decode_content': False, 'preload_content': False}, conn = None, release_this_conn = True, err = None
clean_exit = False, timeout_obj = <urllib3.util.timeout.Timeout object at 0x000002A2F2022220>, is_new_proxy_conn = False
def urlopen(
self,
method,
url,
body=None,
headers=None,
retries=None,
redirect=True,
assert_same_host=True,
timeout=_Default,
pool_timeout=None,
release_conn=None,
chunked=False,
body_pos=None,
**response_kw
):
"""
Get a connection from the pool and perform an HTTP request. This is the
lowest level call for making a request, so you'll need to specify all
the raw details.
.. note::
More commonly, it's appropriate to use a convenience method provided
by :class:`.RequestMethods`, such as :meth:`request`.
.. note::
`release_conn` will only behave as expected if
`preload_content=False` because we want to make
`preload_content=False` the default behaviour someday soon without
breaking backwards compatibility.
:param method:
HTTP request method (such as GET, POST, PUT, etc.)
:param body:
Data to send in the request body (useful for creating
POST requests, see HTTPConnectionPool.post_url for
more convenience).
:param headers:
Dictionary of custom headers to send, such as User-Agent,
If-None-Match, etc. If None, pool headers are used. If provided,
these headers completely replace any pool-specific headers.
:param retries:
Configure the number of retries to allow before raising a
:class:`~urllib3.exceptions.MaxRetryError` exception.
Pass ``None`` to retry until you receive a response. Pass a
:class:`~urllib3.util.retry.Retry` object for fine-grained control
over different types of retries.
Pass an integer number to retry connection errors that many times,
but no other types of errors. Pass zero to never retry.
If ``False``, then retries are disabled and any exception is raised
immediately. Also, instead of raising a MaxRetryError on redirects,
the redirect response will be returned.
:type retries: :class:`~urllib3.util.retry.Retry`, False, or an int.
:param redirect:
If True, automatically handle redirects (status codes 301, 302,
303, 307, 308). Each redirect counts as a retry. Disabling retries
will disable redirect, too.
:param assert_same_host:
If ``True``, will make sure that the host of the pool requests is
consistent else will raise HostChangedError. When False, you can
use the pool on an HTTP proxy and request foreign hosts.
:param timeout:
If specified, overrides the default timeout for this one
request. It may be a float (in seconds) or an instance of
:class:`urllib3.util.Timeout`.
:param pool_timeout:
If set and the pool is set to block=True, then this method will
block for ``pool_timeout`` seconds and raise EmptyPoolError if no
connection is available within the time period.
:param release_conn:
If False, then the urlopen call will not release the connection
back into the pool once a response is received (but will release if
you read the entire contents of the response such as when
`preload_content=True`). This is useful if you're not preloading
the response's content immediately. You will need to call
``r.release_conn()`` on the response ``r`` to return the connection
back into the pool. If None, it takes the value of
``response_kw.get('preload_content', True)``.
:param chunked:
If True, urllib3 will send the body using chunked transfer
encoding. Otherwise, urllib3 will send the body using the standard
content-length form. Defaults to False.
:param int body_pos:
Position to seek to in file-like body in the event of a retry or
redirect. Typically this won't need to be set because urllib3 will
auto-populate the value when needed.
:param \\**response_kw:
Additional parameters are passed to
:meth:`urllib3.response.HTTPResponse.from_httplib`
"""
if headers is None:
headers = self.headers
if not isinstance(retries, Retry):
retries = Retry.from_int(retries, redirect=redirect, default=self.retries)
if release_conn is None:
release_conn = response_kw.get("preload_content", True)
# Check host
if assert_same_host and not self.is_same_host(url):
raise HostChangedError(self, url, retries)
# Ensure that the URL we're connecting to is properly encoded
if url.startswith("/"):
url = six.ensure_str(_encode_target(url))
else:
url = six.ensure_str(parse_url(url).url)
conn = None
# Track whether `conn` needs to be released before
# returning/raising/recursing. Update this variable if necessary, and
# leave `release_conn` constant throughout the function. That way, if
# the function recurses, the original value of `release_conn` will be
# passed down into the recursive call, and its value will be respected.
#
# See issue #651 [1] for details.
#
# [1] <https://github.com/urllib3/urllib3/issues/651>
release_this_conn = release_conn
# Merge the proxy headers. Only do this in HTTP. We have to copy the
# headers dict so we can safely change it without those changes being
# reflected in anyone else's copy.
if self.scheme == "http":
headers = headers.copy()
headers.update(self.proxy_headers)
# Must keep the exception bound to a separate variable or else Python 3
# complains about UnboundLocalError.
err = None
# Keep track of whether we cleanly exited the except block. This
# ensures we do proper cleanup in finally.
clean_exit = False
# Rewind body position, if needed. Record current position
# for future rewinds in the event of a redirect/retry.
body_pos = set_file_position(body, body_pos)
try:
# Request a connection from the queue.
timeout_obj = self._get_timeout(timeout)
conn = self._get_conn(timeout=pool_timeout)
conn.timeout = timeout_obj.connect_timeout
is_new_proxy_conn = self.proxy is not None and not getattr(
conn, "sock", None
)
if is_new_proxy_conn:
self._prepare_proxy(conn)
# Make the request on the httplib connection object.
httplib_response = self._make_request(
conn,
method,
url,
timeout=timeout_obj,
body=body,
headers=headers,
chunked=chunked,
)
# If we're going to release the connection in ``finally:``, then
# the response doesn't need to know about the connection. Otherwise
# it will also try to release it and we'll have a double-release
# mess.
response_conn = conn if not release_conn else None
# Pass method to Response for length checking
response_kw["request_method"] = method
# Import httplib's response into our own wrapper object
response = self.ResponseCls.from_httplib(
httplib_response,
pool=self,
connection=response_conn,
retries=retries,
**response_kw
)
# Everything went great!
clean_exit = True
except queue.Empty:
# Timed out by queue.
raise EmptyPoolError(self, "No pool connections are available.")
except (
TimeoutError,
HTTPException,
SocketError,
ProtocolError,
BaseSSLError,
SSLError,
CertificateError,
) as e:
# Discard the connection for these exceptions. It will be
# replaced during the next _get_conn() call.
clean_exit = False
if isinstance(e, (BaseSSLError, CertificateError)):
e = SSLError(e)
elif isinstance(e, (SocketError, NewConnectionError)) and self.proxy:
e = ProxyError("Cannot connect to proxy.", e)
elif isinstance(e, (SocketError, HTTPException)):
e = ProtocolError("Connection aborted.", e)
> retries = retries.increment(
method, url, error=e, _pool=self, _stacktrace=sys.exc_info()[2]
)
..\..\..\miniconda3\envs\falcon\lib\site-packages\urllib3\connectionpool.py:719:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = Retry(total=0, connect=None, read=False, redirect=None, status=None), method = 'PUT', url = '/', response = None
error = ReadTimeoutError("HTTPConnectionPool(host='127.0.0.1', port=60350): Read timed out. (read timeout=10)")
_pool = <urllib3.connectionpool.HTTPConnectionPool object at 0x000002A2F3D15C10>
_stacktrace = <traceback object at 0x000002A2F3D12900>
def increment(
self,
method=None,
url=None,
response=None,
error=None,
_pool=None,
_stacktrace=None,
):
""" Return a new Retry object with incremented retry counters.
:param response: A response object, or None, if the server did not
return a response.
:type response: :class:`~urllib3.response.HTTPResponse`
:param Exception error: An error encountered during the request, or
None if the response was received successfully.
:return: A new ``Retry`` object.
"""
if self.total is False and error:
# Disabled, indicate to re-raise the error.
raise six.reraise(type(error), error, _stacktrace)
total = self.total
if total is not None:
total -= 1
connect = self.connect
read = self.read
redirect = self.redirect
status_count = self.status
cause = "unknown"
status = None
redirect_location = None
if error and self._is_connection_error(error):
# Connect retry?
if connect is False:
raise six.reraise(type(error), error, _stacktrace)
elif connect is not None:
connect -= 1
elif error and self._is_read_error(error):
# Read retry?
if read is False or not self._is_method_retryable(method):
> raise six.reraise(type(error), error, _stacktrace)
..\..\..\miniconda3\envs\falcon\lib\site-packages\urllib3\util\retry.py:400:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
tp = <class 'urllib3.exceptions.ReadTimeoutError'>, value = None, tb = None
def reraise(tp, value, tb=None):
try:
if value is None:
value = tp()
if value.__traceback__ is not tb:
raise value.with_traceback(tb)
> raise value
..\..\..\miniconda3\envs\falcon\lib\site-packages\urllib3\packages\six.py:735:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <urllib3.connectionpool.HTTPConnectionPool object at 0x000002A2F3D15C10>, method = 'PUT', url = '/', body = '{}'
headers = {'User-Agent': 'python-requests/2.22.0', 'Accept-Encoding': 'gzip, deflate', 'Accept': '*/*', 'Connection': 'keep-alive', 'Content-Length': '2'}
retries = Retry(total=0, connect=None, read=False, redirect=None, status=None), redirect = False
assert_same_host = False, timeout = <urllib3.util.timeout.Timeout object at 0x000002A2F3D152E0>, pool_timeout = None
release_conn = False, chunked = False, body_pos = None
response_kw = {'decode_content': False, 'preload_content': False}, conn = None, release_this_conn = True, err = None
clean_exit = False, timeout_obj = <urllib3.util.timeout.Timeout object at 0x000002A2F2022220>, is_new_proxy_conn = False
def urlopen(
self,
method,
url,
body=None,
headers=None,
retries=None,
redirect=True,
assert_same_host=True,
timeout=_Default,
pool_timeout=None,
release_conn=None,
chunked=False,
body_pos=None,
**response_kw
):
"""
Get a connection from the pool and perform an HTTP request. This is the
lowest level call for making a request, so you'll need to specify all
the raw details.
.. note::
More commonly, it's appropriate to use a convenience method provided
by :class:`.RequestMethods`, such as :meth:`request`.
.. note::
`release_conn` will only behave as expected if
`preload_content=False` because we want to make
`preload_content=False` the default behaviour someday soon without
breaking backwards compatibility.
:param method:
HTTP request method (such as GET, POST, PUT, etc.)
:param body:
Data to send in the request body (useful for creating
POST requests, see HTTPConnectionPool.post_url for
more convenience).
:param headers:
Dictionary of custom headers to send, such as User-Agent,
If-None-Match, etc. If None, pool headers are used. If provided,
these headers completely replace any pool-specific headers.
:param retries:
Configure the number of retries to allow before raising a
:class:`~urllib3.exceptions.MaxRetryError` exception.
Pass ``None`` to retry until you receive a response. Pass a
:class:`~urllib3.util.retry.Retry` object for fine-grained control
over different types of retries.
Pass an integer number to retry connection errors that many times,
but no other types of errors. Pass zero to never retry.
If ``False``, then retries are disabled and any exception is raised
immediately. Also, instead of raising a MaxRetryError on redirects,
the redirect response will be returned.
:type retries: :class:`~urllib3.util.retry.Retry`, False, or an int.
:param redirect:
If True, automatically handle redirects (status codes 301, 302,
303, 307, 308). Each redirect counts as a retry. Disabling retries
will disable redirect, too.
:param assert_same_host:
If ``True``, will make sure that the host of the pool requests is
consistent else will raise HostChangedError. When False, you can
use the pool on an HTTP proxy and request foreign hosts.
:param timeout:
If specified, overrides the default timeout for this one
request. It may be a float (in seconds) or an instance of
:class:`urllib3.util.Timeout`.
:param pool_timeout:
If set and the pool is set to block=True, then this method will
block for ``pool_timeout`` seconds and raise EmptyPoolError if no
connection is available within the time period.
:param release_conn:
If False, then the urlopen call will not release the connection
back into the pool once a response is received (but will release if
you read the entire contents of the response such as when
`preload_content=True`). This is useful if you're not preloading
the response's content immediately. You will need to call
``r.release_conn()`` on the response ``r`` to return the connection
back into the pool. If None, it takes the value of
``response_kw.get('preload_content', True)``.
:param chunked:
If True, urllib3 will send the body using chunked transfer
encoding. Otherwise, urllib3 will send the body using the standard
content-length form. Defaults to False.
:param int body_pos:
Position to seek to in file-like body in the event of a retry or
redirect. Typically this won't need to be set because urllib3 will
auto-populate the value when needed.
:param \\**response_kw:
Additional parameters are passed to
:meth:`urllib3.response.HTTPResponse.from_httplib`
"""
if headers is None:
headers = self.headers
if not isinstance(retries, Retry):
retries = Retry.from_int(retries, redirect=redirect, default=self.retries)
if release_conn is None:
release_conn = response_kw.get("preload_content", True)
# Check host
if assert_same_host and not self.is_same_host(url):
raise HostChangedError(self, url, retries)
# Ensure that the URL we're connecting to is properly encoded
if url.startswith("/"):
url = six.ensure_str(_encode_target(url))
else:
url = six.ensure_str(parse_url(url).url)
conn = None
# Track whether `conn` needs to be released before
# returning/raising/recursing. Update this variable if necessary, and
# leave `release_conn` constant throughout the function. That way, if
# the function recurses, the original value of `release_conn` will be
# passed down into the recursive call, and its value will be respected.
#
# See issue #651 [1] for details.
#
# [1] <https://github.com/urllib3/urllib3/issues/651>
release_this_conn = release_conn
# Merge the proxy headers. Only do this in HTTP. We have to copy the
# headers dict so we can safely change it without those changes being
# reflected in anyone else's copy.
if self.scheme == "http":
headers = headers.copy()
headers.update(self.proxy_headers)
# Must keep the exception bound to a separate variable or else Python 3
# complains about UnboundLocalError.
err = None
# Keep track of whether we cleanly exited the except block. This
# ensures we do proper cleanup in finally.
clean_exit = False
# Rewind body position, if needed. Record current position
# for future rewinds in the event of a redirect/retry.
body_pos = set_file_position(body, body_pos)
try:
# Request a connection from the queue.
timeout_obj = self._get_timeout(timeout)
conn = self._get_conn(timeout=pool_timeout)
conn.timeout = timeout_obj.connect_timeout
is_new_proxy_conn = self.proxy is not None and not getattr(
conn, "sock", None
)
if is_new_proxy_conn:
self._prepare_proxy(conn)
# Make the request on the httplib connection object.
> httplib_response = self._make_request(
conn,
method,
url,
timeout=timeout_obj,
body=body,
headers=headers,
chunked=chunked,
)
..\..\..\miniconda3\envs\falcon\lib\site-packages\urllib3\connectionpool.py:665:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <urllib3.connectionpool.HTTPConnectionPool object at 0x000002A2F3D15C10>
conn = <urllib3.connection.HTTPConnection object at 0x000002A2F2022D60>, method = 'PUT', url = '/'
timeout = <urllib3.util.timeout.Timeout object at 0x000002A2F2022220>, chunked = False
httplib_request_kw = {'body': '{}', 'headers': {'User-Agent': 'python-requests/2.22.0', 'Accept-Encoding': 'gzip, deflate', 'Accept': '*/*', 'Connection': 'keep-alive', 'Content-Length': '2'}}
timeout_obj = <urllib3.util.timeout.Timeout object at 0x000002A2F2022E20>, read_timeout = 10
def _make_request(
self, conn, method, url, timeout=_Default, chunked=False, **httplib_request_kw
):
"""
Perform a request on a given urllib connection object taken from our
pool.
:param conn:
a connection from one of our connection pools
:param timeout:
Socket timeout in seconds for the request. This can be a
float or integer, which will set the same timeout value for
the socket connect and the socket read, or an instance of
:class:`urllib3.util.Timeout`, which gives you more fine-grained
control over your timeouts.
"""
self.num_requests += 1
timeout_obj = self._get_timeout(timeout)
timeout_obj.start_connect()
conn.timeout = timeout_obj.connect_timeout
# Trigger any extra validation we need to do.
try:
self._validate_conn(conn)
except (SocketTimeout, BaseSSLError) as e:
# Py2 raises this as a BaseSSLError, Py3 raises it as socket timeout.
self._raise_timeout(err=e, url=url, timeout_value=conn.timeout)
raise
# conn.request() calls httplib.*.request, not the method in
# urllib3.request. It also calls makefile (recv) on the socket.
if chunked:
conn.request_chunked(method, url, **httplib_request_kw)
else:
conn.request(method, url, **httplib_request_kw)
# Reset the timeout for the recv() on the socket
read_timeout = timeout_obj.read_timeout
# App Engine doesn't have a sock attr
if getattr(conn, "sock", None):
# In Python 3 socket.py will catch EAGAIN and return None when you
# try and read into the file pointer created by http.client, which
# instead raises a BadStatusLine exception. Instead of catching
# the exception and assuming all BadStatusLine exceptions are read
# timeouts, check for a zero timeout before making the request.
if read_timeout == 0:
raise ReadTimeoutError(
self, url, "Read timed out. (read timeout=%s)" % read_timeout
)
if read_timeout is Timeout.DEFAULT_TIMEOUT:
conn.sock.settimeout(socket.getdefaulttimeout())
else: # None or a value
conn.sock.settimeout(read_timeout)
# Receive the response from the server
try:
try:
# Python 2.7, use buffering of HTTP responses
httplib_response = conn.getresponse(buffering=True)
except TypeError:
# Python 3
try:
httplib_response = conn.getresponse()
except BaseException as e:
# Remove the TypeError from the exception chain in
# Python 3 (including for exceptions like SystemExit).
# Otherwise it looks like a bug in the code.
six.raise_from(e, None)
except (SocketTimeout, BaseSSLError, SocketError) as e:
> self._raise_timeout(err=e, url=url, timeout_value=read_timeout)
..\..\..\miniconda3\envs\falcon\lib\site-packages\urllib3\connectionpool.py:423:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <urllib3.connectionpool.HTTPConnectionPool object at 0x000002A2F3D15C10>, err = timeout('timed out'), url = '/'
timeout_value = 10
def _raise_timeout(self, err, url, timeout_value):
"""Is the error actually a timeout? Will raise a ReadTimeout or pass"""
if isinstance(err, SocketTimeout):
> raise ReadTimeoutError(
self, url, "Read timed out. (read timeout=%s)" % timeout_value
)
E urllib3.exceptions.ReadTimeoutError: HTTPConnectionPool(host='127.0.0.1', port=60350): Read timed out. (read timeout=10)
..\..\..\miniconda3\envs\falcon\lib\site-packages\urllib3\connectionpool.py:330: ReadTimeoutError
During handling of the above exception, another exception occurred:
self = <asgi.test_asgi_servers.TestASGIServer object at 0x000002A2F1F9B6A0>, server_base_url = 'http://127.0.0.1:60350/'
def test_put(self, server_base_url):
body = '{}'
> resp = requests.put(server_base_url, data=body, timeout=_REQUEST_TIMEOUT)
tests\asgi\test_asgi_servers.py:34:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
..\..\..\miniconda3\envs\falcon\lib\site-packages\requests\api.py:131: in put
return request('put', url, data=data, **kwargs)
..\..\..\miniconda3\envs\falcon\lib\site-packages\requests\api.py:60: in request
return session.request(method=method, url=url, **kwargs)
..\..\..\miniconda3\envs\falcon\lib\site-packages\requests\sessions.py:533: in request
resp = self.send(prep, **send_kwargs)
..\..\..\miniconda3\envs\falcon\lib\site-packages\requests\sessions.py:646: in send
r = adapter.send(request, **kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <requests.adapters.HTTPAdapter object at 0x000002A2F3D70040>, request = <PreparedRequest [PUT]>, stream = False
timeout = <urllib3.util.timeout.Timeout object at 0x000002A2F3D152E0>, verify = True, cert = None
proxies = OrderedDict()
def send(self, request, stream=False, timeout=None, verify=True, cert=None, proxies=None):
"""Sends PreparedRequest object. Returns Response object.
:param request: The :class:`PreparedRequest <PreparedRequest>` being sent.
:param stream: (optional) Whether to stream the request content.
:param timeout: (optional) How long to wait for the server to send
data before giving up, as a float, or a :ref:`(connect timeout,
read timeout) <timeouts>` tuple.
:type timeout: float or tuple or urllib3 Timeout object
:param verify: (optional) Either a boolean, in which case it controls whether
we verify the server's TLS certificate, or a string, in which case it
must be a path to a CA bundle to use
:param cert: (optional) Any user-provided SSL certificate to be trusted.
:param proxies: (optional) The proxies dictionary to apply to the request.
:rtype: requests.Response
"""
try:
conn = self.get_connection(request.url, proxies)
except LocationValueError as e:
raise InvalidURL(e, request=request)
self.cert_verify(conn, request.url, verify, cert)
url = self.request_url(request, proxies)
self.add_headers(request, stream=stream, timeout=timeout, verify=verify, cert=cert, proxies=proxies)
chunked = not (request.body is None or 'Content-Length' in request.headers)
if isinstance(timeout, tuple):
try:
connect, read = timeout
timeout = TimeoutSauce(connect=connect, read=read)
except ValueError as e:
# this may raise a string formatting error.
err = ("Invalid timeout {}. Pass a (connect, read) "
"timeout tuple, or a single float to set "
"both timeouts to the same value".format(timeout))
raise ValueError(err)
elif isinstance(timeout, TimeoutSauce):
pass
else:
timeout = TimeoutSauce(connect=timeout, read=timeout)
try:
if not chunked:
resp = conn.urlopen(
method=request.method,
url=url,
body=request.body,
headers=request.headers,
redirect=False,
assert_same_host=False,
preload_content=False,
decode_content=False,
retries=self.max_retries,
timeout=timeout
)
# Send the request.
else:
if hasattr(conn, 'proxy_pool'):
conn = conn.proxy_pool
low_conn = conn._get_conn(timeout=DEFAULT_POOL_TIMEOUT)
try:
low_conn.putrequest(request.method,
url,
skip_accept_encoding=True)
for header, value in request.headers.items():
low_conn.putheader(header, value)
low_conn.endheaders()
for i in request.body:
low_conn.send(hex(len(i))[2:].encode('utf-8'))
low_conn.send(b'\r\n')
low_conn.send(i)
low_conn.send(b'\r\n')
low_conn.send(b'0\r\n\r\n')
# Receive the response from the server
try:
# For Python 2.7, use buffering of HTTP responses
r = low_conn.getresponse(buffering=True)
except TypeError:
# For compatibility with Python 3.3+
r = low_conn.getresponse()
resp = HTTPResponse.from_httplib(
r,
pool=conn,
connection=low_conn,
preload_content=False,
decode_content=False
)
except:
# If we hit any problems here, clean up the connection.
# Then, reraise so that we can handle the actual exception.
low_conn.close()
raise
except (ProtocolError, socket.error) as err:
raise ConnectionError(err, request=request)
except MaxRetryError as e:
if isinstance(e.reason, ConnectTimeoutError):
# TODO: Remove this in 3.0.0: see #2811
if not isinstance(e.reason, NewConnectionError):
raise ConnectTimeout(e, request=request)
if isinstance(e.reason, ResponseError):
raise RetryError(e, request=request)
if isinstance(e.reason, _ProxyError):
raise ProxyError(e, request=request)
if isinstance(e.reason, _SSLError):
# This branch is for urllib3 v1.22 and later.
raise SSLError(e, request=request)
raise ConnectionError(e, request=request)
except ClosedPoolError as e:
raise ConnectionError(e, request=request)
except _ProxyError as e:
raise ProxyError(e)
except (_SSLError, _HTTPError) as e:
if isinstance(e, _SSLError):
# This branch is for urllib3 versions earlier than v1.22
raise SSLError(e, request=request)
elif isinstance(e, ReadTimeoutError):
> raise ReadTimeout(e, request=request)
E requests.exceptions.ReadTimeout: HTTPConnectionPool(host='127.0.0.1', port=60350): Read timed out. (read timeout=10)
..\..\..\miniconda3\envs\falcon\lib\site-packages\requests\adapters.py:529: ReadTimeout
------------------------------------------------ Captured stdout setup ------------------------------------------------
[Starting server process...]
------------------------------------------------ Captured stderr setup ------------------------------------------------
INFO: Started server process [5924]
INFO: Waiting for application startup.
WARNING: Executing <Task pending name='Task-1' coro=<Server.serve() running at c:\users\federico\miniconda3\envs\falcon\lib\site-packages\uvicorn\main.py:399> wait_for=<Future
pending cb=[<TaskWakeupMethWrapper object at 0x0000023B494CEDF0>()] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:422> cb=[_run_until_complete_cb() at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:184] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\base_events.py:591> took 1.187 seconds
INFO: Application startup complete.
INFO: <Server sockets=(<asyncio.TransportSocket fd=716, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('127.0.0.1', 60350)>,)> is serving
INFO: Uvicorn running on http://127.0.0.1:60350 (Press CTRL+C to quit)
INFO: 127.0.0.1:60358 - "GET / HTTP/1.1" 200
------------------------------------------------ Captured stderr call -------------------------------------------------
WARNING: Executing <Task pending name='Task-7' coro=<RequestResponseCycle.run_asgi() running at c:\users\federico\miniconda3\envs\falcon\lib\site-packages\uvicorn\protocols\http\h11_impl.py:385> wait_for=<_GatheringFuture pending cb=[<TaskWakeupMethWrapper object at 0x0000023B49C4A310>()] created at c:\users\federico\miniconda3\envs\falcon\lib\asyncio\tasks.py:698> cb=[set.discard()] created at c:\users\federico\miniconda3\envs\falcon\lib\site-packages\uvicorn\protocols\http\h11_impl.py:234> took 1.578 seconds
================================================== warnings summary ===================================================
c:\users\federico\miniconda3\envs\falcon\lib\site-packages\win32\lib\pywintypes.py:2
c:\users\federico\miniconda3\envs\falcon\lib\site-packages\win32\lib\pywintypes.py:2: DeprecationWarning: the imp module is deprecated in favour of importlib; see the module's documentation for alternative uses
import imp, sys, os
-- Docs: https://docs.pytest.org/en/latest/warnings.html
=================== 47 failed, 2094 passed, 79 skipped, 1 warning, 16 errors in 1885.17s (0:31:25) ====================
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment