Created
January 26, 2020 10:10
-
-
Save CaselIT/16792866e7dd2c545b186207976e060e to your computer and use it in GitHub Desktop.
Falcon 3 dev errors on windows
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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