Merge pull request #350 from mmattozzi/master

Adding a /etag URL for testing If-Match and If-None-Match conditional logic

httpbin/core.py

@@ -23,7 +23,7 @@ from werkzeug.wrappers import BaseResponse
 from raven.contrib.flask import Sentry
 
 from . import filters
-from .helpers import get_headers, status_code, get_dict, get_request_range, check_basic_auth, check_digest_auth, secure_cookie, H, ROBOT_TXT, ANGRY_ASCII
+from .helpers import get_headers, status_code, get_dict, get_request_range, check_basic_auth, check_digest_auth, secure_cookie, H, ROBOT_TXT, ANGRY_ASCII, parse_multi_value_header
 from .utils import weighted_choice
 from .structures import CaseInsensitiveDict
 
@@ -533,6 +533,23 @@ def cache():
     else:
         return status_code(304)
 
+@app.route('/etag/<etag>', methods=('GET',))
+def etag(etag):
+    """Assumes the resource has the given etag and responds to If-None-Match and If-Match headers appropriately."""
+    if_none_match = parse_multi_value_header(request.headers.get('If-None-Match'))
+    if_match = parse_multi_value_header(request.headers.get('If-Match'))
+
+    if if_none_match:
+        if etag in if_none_match or '*' in if_none_match:
+            return status_code(304)
+    elif if_match:
+        if etag not in if_match and '*' not in if_match:
+            return status_code(412)
+
+    # Special cases don't apply, return normal response
+    response = view_get()
+    response.headers['ETag'] = etag
+    return response
 
 @app.route('/cache/<int:value>')
 def cache_control(value):

httpbin/helpers.py

@@ -9,6 +9,7 @@ This module provides helper functions for httpbin.
 
 import json
 import base64
+import re
 from hashlib import md5, sha256
 from werkzeug.http import parse_authorization_header
 
@@ -420,3 +421,13 @@ def get_request_range(request_headers, upper_bound):
 
     return first_byte_pos, last_byte_pos
 
+def parse_multi_value_header(header_str):
+    """Break apart an HTTP header string that is potentially a quoted, comma separated list as used in entity headers in RFC2616."""
+    parsed_parts = []
+    if header_str:
+        parts = header_str.split(',')
+        for part in parts:
+            match = re.search('\s*(W/)?\"?([^"]*)\"?\s*', part)
+            if match is not None:
+                parsed_parts.append(match.group(2))
+    return parsed_parts

httpbin/templates/httpbin.1.html

@@ -40,6 +40,7 @@
 <li><a href="{{ url_for('view_robots_page') }}" data-bare-link="true"><code>/robots.txt</code></a> Returns some robots.txt rules.</li>
 <li><a href="{{ url_for('view_deny_page') }}" data-bare-link="true"><code>/deny</code></a> Denied by robots.txt file.</li>
 <li><a href="{{ url_for('cache') }}" data-bare-link="true"><code>/cache</code></a> Returns 200 unless an If-Modified-Since or If-None-Match header is provided, when it returns a 304.</li>
+<li><a href="{{ url_for('etag', etag='etag') }}"><code>/etag/:etag</code></a> Assumes the resource has the given etag and responds to If-None-Match header with a 200 or 304 and If-Match with a 200 or 412 as appropriate.</li>
 <li><a href="{{ url_for('cache_control', value=60) }}"><code>/cache/:n</code></a> Sets a Cache-Control header for <em>n</em> seconds.</li>
 <li><a href="{{ url_for('random_bytes', n=1024) }}"><code>/bytes/:n</code></a> Generates <em>n</em> random bytes of binary data, accepts optional <em>seed</em> integer parameter.</li>
 <li><a href="{{ url_for('stream_random_bytes', n=1024) }}"><code>/stream-bytes/:n</code></a> Streams <em>n</em> random bytes of binary data, accepts optional <em>seed</em> and <em>chunk_size</em> integer parameters.</li>

test_httpbin.py

@@ -11,6 +11,7 @@ from hashlib import md5, sha256
 from six import BytesIO
 
 import httpbin
+from httpbin.helpers import parse_multi_value_header
 
 
 @contextlib.contextmanager
@@ -564,6 +565,83 @@ class HttpbinTestCase(unittest.TestCase):
         data = response.data.decode('utf-8')
         self.assertIn('perfectaudience', data)
 
+    def test_etag_if_none_match_matches(self):
+        response = self.app.get(
+            '/etag/abc',
+            headers={ 'If-None-Match': 'abc' }
+        )
+        self.assertEqual(response.status_code, 304)
+
+    def test_etag_if_none_match_matches_list(self):
+        response = self.app.get(
+            '/etag/abc',
+            headers={ 'If-None-Match': '"123", "abc"' }
+        )
+        self.assertEqual(response.status_code, 304)
+
+    def test_etag_if_none_match_matches_star(self):
+        response = self.app.get(
+            '/etag/abc',
+            headers={ 'If-None-Match': '*' }
+        )
+        self.assertEqual(response.status_code, 304)
+
+    def test_etag_if_none_match_w_prefix(self):
+        response = self.app.get(
+            '/etag/c3piozzzz',
+            headers={ 'If-None-Match': 'W/"xyzzy", W/"r2d2xxxx", W/"c3piozzzz"' }
+        )
+        self.assertEqual(response.status_code, 304)
+
+    def test_etag_if_none_match_has_no_match(self):
+        response = self.app.get(
+            '/etag/abc',
+            headers={ 'If-None-Match': '123' }
+        )
+        self.assertEqual(response.status_code, 200)
+
+    def test_etag_if_match_matches(self):
+        response = self.app.get(
+            '/etag/abc',
+            headers={ 'If-Match': 'abc' }
+        )
+        self.assertEqual(response.status_code, 200)
+
+    def test_etag_if_match_matches_list(self):
+        response = self.app.get(
+            '/etag/abc',
+            headers={ 'If-Match': '"123", "abc"' }
+        )
+        self.assertEqual(response.status_code, 200)
+
+    def test_etag_if_match_matches_star(self):
+        response = self.app.get(
+            '/etag/abc',
+            headers={ 'If-Match': '*' }
+        )
+        self.assertEqual(response.status_code, 200)
+
+    def test_etag_if_match_has_no_match(self):
+        response = self.app.get(
+            '/etag/abc',
+            headers={ 'If-Match': '123' }
+        )
+        self.assertEqual(response.status_code, 412)
+
+    def test_etag_with_no_headers(self):
+        response = self.app.get(
+            '/etag/abc'
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.headers.get('ETag'), 'abc')
+
+    def test_parse_multi_value_header(self):
+        self.assertEqual(parse_multi_value_header('xyzzy'), [ "xyzzy" ])
+        self.assertEqual(parse_multi_value_header('"xyzzy"'), [ "xyzzy" ])
+        self.assertEqual(parse_multi_value_header('W/"xyzzy"'), [ "xyzzy" ])
+        self.assertEqual(parse_multi_value_header('"xyzzy", "r2d2xxxx", "c3piozzzz"'), [ "xyzzy", "r2d2xxxx", "c3piozzzz" ])
+        self.assertEqual(parse_multi_value_header('W/"xyzzy", W/"r2d2xxxx", W/"c3piozzzz"'), [ "xyzzy", "r2d2xxxx", "c3piozzzz" ])
+        self.assertEqual(parse_multi_value_header('*'), [ "*" ])
 
 if __name__ == '__main__':
     unittest.main()