Make StaticRoute support Last-Modified and If-Modified-Since headers

aiohttp/web_urldispatcher.py

@@ -11,11 +11,12 @@
 import inspect
 
 from urllib.parse import urlencode
+from wsgiref.handlers import format_date_time
 
 from . import hdrs
 from .abc import AbstractRouter, AbstractMatchInfo
 from .protocol import HttpVersion11
-from .web_exceptions import HTTPMethodNotAllowed, HTTPNotFound
+from .web_exceptions import HTTPMethodNotAllowed, HTTPNotFound, HTTPNotModified
 from .web_reqrep import StreamResponse
 
 
@@ -169,22 +170,30 @@ def url(self, *, filename, query=None):
 
     @asyncio.coroutine
     def handle(self, request):
-        resp = StreamResponse()
         filename = request.match_info['filename']
         filepath = os.path.abspath(os.path.join(self._directory, filename))
         if not filepath.startswith(self._directory):
             raise HTTPNotFound()
         if not os.path.exists(filepath) or not os.path.isfile(filepath):
             raise HTTPNotFound()
 
+        st = os.stat(filepath)
+        mtime = format_date_time(st.st_mtime)
+
+        if request.headers.get(hdrs.IF_MODIFIED_SINCE) == mtime:
+            raise HTTPNotModified()
+
         ct, encoding = mimetypes.guess_type(filepath)
         if not ct:
             ct = 'application/octet-stream'
+
+        resp = StreamResponse()
         resp.content_type = ct
         if encoding:
-            resp.headers['content-encoding'] = encoding
+            resp.headers[hdrs.CONTENT_ENCODING] = encoding
+        resp.headers[hdrs.LAST_MODIFIED] = mtime
 
-        file_size = os.stat(filepath).st_size
+        file_size = st.st_size
         single_chunk = file_size < self._chunk_size
 
         if single_chunk:

tests/test_web_functional.py

@@ -374,6 +374,48 @@ def go(dirname, relpath):
         filename = '../README.rst'
         self.loop.run_until_complete(go(here, filename))
 
+    def test_static_file_if_modified_since(self):
+
+        @asyncio.coroutine
+        def go(dirname, filename):
+            app, _, url = yield from self.create_server(
+                'GET', '/static/' + filename
+            )
+            app.router.add_static('/static', dirname)
+
+            resp = yield from request('GET', url, loop=self.loop)
+            self.assertEqual(200, resp.status)
+            lastmod = resp.headers.get('Last-Modified')
+            self.assertIsNotNone(lastmod)
+            resp.close()
+
+            resp = yield from request('GET', url, loop=self.loop,
+                                      headers={'If-Modified-Since': lastmod})
+            self.assertEqual(304, resp.status)
+            resp.close()
+
+            lastmod = 'Mon, 1 Jan 1990 01:01:01 GMT'
+            resp = yield from request('GET', url, loop=self.loop,
+                                      headers={'If-Modified-Since': lastmod})
+            self.assertIn(resp.status, (200, 304))
+            resp.close()
+
+            lastmod = 'Fri, 31 Dec 9999 23:59:59 GMT'
+            resp = yield from request('GET', url, loop=self.loop,
+                                      headers={'If-Modified-Since': lastmod})
+            self.assertEqual(200, resp.status)
+            resp.close()
+
+            lastmod = 'not a valid HTTP-date'
+            resp = yield from request('GET', url, loop=self.loop,
+                                      headers={'If-Modified-Since': lastmod})
+            self.assertEqual(200, resp.status)
+            resp.close()
+
+        here = os.path.dirname(__file__)
+        filename = 'data.unknown_mime_type'
+        self.loop.run_until_complete(go(here, filename))
+
     def test_static_route_path_existence_check(self):
         directory = os.path.dirname(__file__)
         web.StaticRoute(None, "/", directory)