Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fixed bug in serving static directory #803

Merged
merged 6 commits into from
Feb 27, 2016
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions aiohttp/web_urldispatcher.py
Original file line number Diff line number Diff line change
Expand Up @@ -569,6 +569,10 @@ def handle(self, request):
request.logger.exception(error)
raise HTTPNotFound() from error

# Make sure that filepath is a file
if not filepath.is_file():
raise HTTPNotFound()
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

flake8 tool blames that you have trailing whitespaces in the line.


st = filepath.stat()

modsince = request.if_modified_since
Expand Down
103 changes: 103 additions & 0 deletions tests/test_web_urldispatcher.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
import pytest
import tempfile
import aiohttp
from aiohttp import web
import os
import shutil
import asyncio

SERVER_HOST = '127.0.0.1'
SERVER_PORT = 8080
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.


# Timeout in seconds for an asynchronous test:
ASYNC_TEST_TIMEOUT = 1

class ExceptAsyncTestTimeout(Exception): pass

def run_timeout(cor,loop,timeout=ASYNC_TEST_TIMEOUT):
"""
Run a given coroutine with timeout.
"""
task_with_timeout = asyncio.wait_for(cor,timeout,loop=loop)
try:
return loop.run_until_complete(task_with_timeout)
except asyncio.futures.TimeoutError:
# Timeout:
raise ExceptAsyncTestTimeout()


@pytest.fixture(scope='function')
def tloop(request):
"""
Obtain a test loop. We want each test case to have its own loop.
"""
# Create a new test loop:
tloop = asyncio.new_event_loop()
asyncio.set_event_loop(None)

def teardown():
# Close the test loop:
tloop.close()

request.addfinalizer(teardown)
return tloop


@pytest.fixture(scope='function')
def tmp_dir_path(request):
"""
Give a path for a temporary directory
The directory is destroyed at the end of the test.
"""
# Temporary directory.
tmp_dir = tempfile.mkdtemp()

def teardown():
# Delete the whole directory:
shutil.rmtree(tmp_dir)

request.addfinalizer(teardown)
return tmp_dir


def test_access_root_of_static_handler(tloop,tmp_dir_path):
"""
Tests the operation of static file server.
Try to access the root of static file server, and make
sure that a proper not found error is returned.
"""
# Put a file inside tmp_dir_path:
my_file_path = os.path.join(tmp_dir_path,'my_file')
with open(my_file_path,'w') as fw:
fw.write('hello')

asyncio.set_event_loop(None)
app = web.Application(loop=tloop)
# Register global static route:
app.router.add_static('/', tmp_dir_path)

@asyncio.coroutine
def inner_cor():
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I suggest to get rid of inner function.

Declare test_access_root_of_static_handler as

def test_access_root_of_static_handler(...):

It allows to use yield from statements inside the test function.

handler = app.make_handler()
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Use create_app_and_client fixture https://github.com/KeepSafe/aiohttp/blob/master/tests/conftest.py#L232-L250 for test server starting/stopping.

srv = yield from tloop.create_server(handler,\
SERVER_HOST,SERVER_PORT ,reuse_address=True)

# Request the root of the static directory.
# Expect an 404 error page.
url = 'http://{}:{}/'.format(\
SERVER_HOST,SERVER_PORT)

r = ( yield from aiohttp.get(url,loop=tloop) )
assert r.status == 404
# data = (yield from r.read())
yield from r.release()

srv.close()
yield from srv.wait_closed()

yield from app.shutdown()
yield from handler.finish_connections(10.0)
yield from app.cleanup()


run_timeout(inner_cor(),tloop,timeout=5)