Permalink
Browse files

Document tests, replace numerical codes with readable statuses

Do not interpret a NoneType as response content
  • Loading branch information...
1 parent 04b8902 commit b4895d6744fdbc1db6433e64a37e5dad40db8c41 @bruth committed Aug 2, 2012
Showing with 106 additions and 53 deletions.
  1. +1 −1 restlib2/resources.py
  2. +105 −52 restlib2/tests/cases.py
@@ -380,7 +380,7 @@ def process_response(self, request, response):
# If the response already has a `_raw_content` attribute, do not
# bother with the local content.
- if content != '':
+ if content is not None and content != '':
if not isinstance(content, basestring) and serializers.supports_encoding(accept_type):
# Encode the body
content = serializers.encode(accept_type, content)
@@ -1,65 +1,78 @@
import unittest
from calendar import timegm
from django.test.client import RequestFactory
-from django.http import HttpResponse
from restlib2.resources import Resource
from restlib2.http import codes
+
class ResourceTestCase(unittest.TestCase):
def setUp(self):
self.factory = RequestFactory()
def test_default(self):
"Tests for the default Resource which is very limited."
+ # Default resource
resource = Resource()
+ # Populated implicitly via the metaclass..
self.assertEqual(resource.allowed_methods, ('OPTIONS',))
+ # OPTIONS is successful, default response with no content is a 204
request = self.factory.request(REQUEST_METHOD='OPTIONS')
response = resource(request)
- self.assertEqual(response.status_code, 204)
+ self.assertEqual(response.status_code, codes.no_content)
# Try another non-default method
request = self.factory.request()
response = resource(request)
- self.assertEqual(response.status_code, 405)
+ self.assertEqual(response.status_code, codes.method_not_allowed)
self.assertEqual(response['Allow'], 'OPTIONS')
+ def test_default_patch(self):
+ # Resources supporting PATCH requests should have an additional
+ # header in the response from an OPTIONS request
class PatchResource(Resource):
allowed_methods = ('PATCH', 'OPTIONS')
def patch(self, request):
pass
resource = PatchResource()
-
request = self.factory.request(REQUEST_METHOD='OPTIONS')
response = resource(request)
- self.assertEqual(response.status_code, 204)
+ self.assertEqual(response.status_code, codes.no_content)
self.assertEqual(response['Accept-Patch'], 'application/json')
def test_service_unavailable(self):
- "Test service availability."
+ "Test service unavailability."
class IndefiniteUnavailableResource(Resource):
unavailable = True
resource = IndefiniteUnavailableResource()
+ # Simply setting `unavailable` to True will provide a 'Retry-After'
+ # header
request = self.factory.request()
response = resource(request)
- self.assertEqual(response.status_code, 503)
+ self.assertEqual(response.status_code, codes.service_unavailable)
self.assertTrue('Retry-After' not in response)
+ def test_service_unavailable_retry_seconds(self):
+ "Test service unavailability with seconds."
class DeltaUnavailableResource(Resource):
unavailable = 20
resource = DeltaUnavailableResource()
+ # Set unavailable, but with a specific number of seconds to retry
+ # after
request = self.factory.request()
response = resource(request)
- self.assertEqual(response.status_code, 503)
+ self.assertEqual(response.status_code, codes.service_unavailable)
self.assertEqual(response['Retry-After'], '20')
+ def test_service_unavailable_retry_date(self):
+ "Test service unavailability with date."
from datetime import datetime, timedelta
from django.utils.http import http_date
@@ -72,51 +85,57 @@ class DatetimeUnavailableResource(Resource):
request = self.factory.request()
response = resource(request)
- self.assertEqual(response.status_code, 503)
+ self.assertEqual(response.status_code, codes.service_unavailable)
self.assertEqual(response['Retry-After'], http_date(timegm(future.utctimetuple())))
def test_unsupported_media_type(self):
"Test various Content-* combinations."
- class ReadOnlyResource(Resource):
+ class NoOpResource(Resource):
def post(self, request, *args, **kwargs):
pass
- resource = ReadOnlyResource()
+ resource = NoOpResource()
- # Works..
+ # Works.. default accept-type is application/json
request = self.factory.post('/', data='{"message": "hello world"}', content_type='application/json; charset=utf-8')
response = resource(request)
- self.assertEqual(response.status_code, 204)
+ self.assertEqual(response.status_code, codes.no_content)
- # Unsupported Media Type
+ # Does not work.. XML not accepted by default
request = self.factory.post('/', data='<message>hello world</message>', content_type='application/xml')
response = resource(request)
- self.assertEqual(response.status_code, 415)
+ self.assertEqual(response.status_code, codes.unsupported_media_type)
def test_not_acceptable(self):
- "Test various Accept-* combinations."
+ "Test Accept header."
class ReadOnlyResource(Resource):
def get(self, request, *args, **kwargs):
return '{}'
resource = ReadOnlyResource()
- # Non-explicit
+ # No accept-type is specified, defaults to highest priority one
+ # for resource
request = self.factory.request()
response = resource(request)
- self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.status_code, codes.ok)
+ # Explicit accept header, application/json wins since it's equal
+ # priority and supported
request = self.factory.request(HTTP_ACCEPT='application/json,application/xml;q=0.9,*/*;q=0.8')
response = resource(request)
- self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.status_code, codes.ok)
+ # No acceptable type list, */* has an explicit quality of 0 which
+ # does not allow the server to use an alternate content-type
request = self.factory.request(HTTP_ACCEPT='text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0')
response = resource(request)
- self.assertEqual(response.status_code, 406)
+ self.assertEqual(response.status_code, codes.not_acceptable)
+ # Like the first one, but an explicit "anything goes"
request = self.factory.request(HTTP_ACCEPT='*/*')
response = resource(request)
- self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.status_code, codes.ok)
def test_request_entity_too_large(self):
"Test request entity too large."
@@ -128,15 +147,17 @@ def post(self, request, *args, **kwargs):
resource = TinyResource()
+ # No problem..
request = self.factory.post('/', data='{"message": "hello"}', content_type='application/json')
response = resource(request)
- self.assertEqual(response.status_code, 204)
+ self.assertEqual(response.status_code, codes.no_content)
+ # Too large
request = self.factory.post('/', data='{"message": "hello world"}', content_type='application/json')
response = resource(request)
- self.assertEqual(response.status_code, 413)
+ self.assertEqual(response.status_code, codes.request_entity_too_large)
- def test_rate_limit(self):
+ def test_too_many_requests(self):
"""Test a global rate limiting implementation.
This test will take 3 seconds to run to mimic request handling over
@@ -146,6 +167,7 @@ def test_rate_limit(self):
from datetime import datetime
class RateLimitResource(Resource):
+ # Maximum of 10 requests within a 2 second window
rate_limit_count = 10
rate_limit_seconds = 2
@@ -154,18 +176,21 @@ class RateLimitResource(Resource):
request_frame_start = datetime.now()
request_count = 0
+ # Implement rate-limiting logic
def is_too_many_requests(self, request, *args, **kwargs):
+ # Since the start of the frame, calculate the amount of time
+ # that has passed
interval = (datetime.now() - self.request_frame_start).seconds
+ # Increment the request count
self.request_count += 1
# Reset frame if the interval is greater than the rate limit seconds,
- # i.e on the 3 second in this test
+ # i.e on the 3rd second in this test
if interval > self.rate_limit_seconds:
self.request_frame_start = datetime.now()
self.request_count = 1
-
- # Throttle
- elif self.request_count > self.rate_limit_count and interval <= self.rate_limit_seconds:
+ # ..otherwise throttle if the count is greater than the limit
+ elif self.request_count > self.rate_limit_count:
return True
return False
@@ -174,27 +199,38 @@ def is_too_many_requests(self, request, *args, **kwargs):
request = self.factory.request(REQUEST_METHOD='OPTIONS')
+ # First ten requests are ok
for _ in xrange(0, 10):
response = resource(request)
- self.assertEqual(response.status_code, 204)
+ self.assertEqual(response.status_code, codes.no_content)
+ # Mimic a slight delay
time.sleep(1)
+ # Another 10 all get throttled..
for _ in xrange(0, 10):
response = resource(request)
- self.assertEqual(response.status_code, 429)
+ self.assertEqual(response.status_code, codes.too_many_requests)
+ # Another two seconds exceeds the frame, should be good to go
time.sleep(2)
for _ in xrange(0, 10):
response = resource(request)
- self.assertEqual(response.status_code, 204)
+ self.assertEqual(response.status_code, codes.no_content)
def test_precondition_required(self):
+ "Reject non-idempotent requests without the use of a conditional header."
+
class PreconditionResource(Resource):
- require_conditional_request = True
+ # Either etags or last-modified must be used otherwise it
+ # is not enforced
use_etags = True
+ require_conditional_request = True
+
+ def patch(self, request):
+ pass
def put(self, request):
pass
@@ -208,47 +244,61 @@ def get_etag(self, request, *args, **kwargs):
resource = PreconditionResource()
+ # Non-idempotent requests fail without a conditional header, these
+ # responses should not be cached
request = self.factory.put('/', data='{"message": "hello world"}', content_type='application/json')
response = resource(request)
- self.assertEqual(response.status_code, 428)
+ self.assertEqual(response.status_code, codes.precondition_required)
self.assertEqual(response['Cache-Control'], 'no-cache')
self.assertEqual(response['Pragma'], 'no-cache')
+ # Add the correct header for testing the Etag
request = self.factory.put('/', data='{"message": "hello world"}', content_type='application/json',
HTTP_IF_MATCH='abc123')
response = resource(request)
- self.assertEqual(response.status_code, 204)
+ self.assertEqual(response.status_code, codes.no_content)
- def test_precondition_failed(self):
+ # Idempotent requests, such as DELETE, succeed..
+ request = self.factory.delete('/')
+ response = resource(request)
+ self.assertEqual(response.status_code, codes.no_content)
+
+ def test_precondition_failed_etag(self):
+ "Test precondition using etags."
class PreconditionResource(Resource):
use_etags = True
def put(self, request):
pass
def get(self, request):
- pass
+ return '{}'
def get_etag(self, request, *args, **kwargs):
return 'abc123'
resource = PreconditionResource()
- request = self.factory.put('/', data='{"message": "hello world"}', content_type='application/json',
- HTTP_IF_MATCH='"def456"')
+ # Send a non-safe request with an incorrect Etag.. fail
+ request = self.factory.put('/', data='{"message": "hello world"}',
+ content_type='application/json', HTTP_IF_MATCH='"def456"')
response = resource(request)
- self.assertEqual(response.status_code, 412)
+ self.assertEqual(response.status_code, codes.precondition_failed)
self.assertEqual(response['Cache-Control'], 'no-cache')
self.assertEqual(response['Pragma'], 'no-cache')
+ # Incorrect Etag match on GET, updated content is returned
request = self.factory.get('/', HTTP_IF_NONE_MATCH='"def456"')
response = resource(request)
- self.assertEqual(response.status_code, 204)
+ self.assertEqual(response.status_code, codes.ok)
+ # Successful Etag match on GET, resource not modified
request = self.factory.get('/', HTTP_IF_NONE_MATCH='"abc123"')
response = resource(request)
- self.assertEqual(response.status_code, 304)
+ self.assertEqual(response.status_code, codes.not_modified)
+ def test_precondition_failed_last_modified(self):
+ "Test precondition using last-modified dates."
from datetime import datetime, timedelta
from django.utils.http import http_date
@@ -262,27 +312,30 @@ def put(self, request):
pass
def get(self, request):
- pass
+ return '{}'
def get_last_modified(self, request, *args, **kwargs):
return last_modified_date
resource = PreconditionResource()
-
- past_time = datetime.now() - timedelta(seconds=-10)
- past_seconds = timegm(past_time.utctimetuple())
+
+ # Send non-safe request with a old last-modified date.. fail
+ if_modified_since = http_date(timegm((last_modified_date - timedelta(seconds=10)).utctimetuple()))
request = self.factory.put('/', data='{"message": "hello world"}',
- content_type='application/json', HTTP_IF_UNMODIFIED_SINCE=http_date(past_seconds))
+ content_type='application/json', HTTP_IF_UNMODIFIED_SINCE=if_modified_since)
response = resource(request)
- self.assertEqual(response.status_code, 412)
+ self.assertEqual(response.status_code, codes.precondition_failed)
self.assertEqual(response['Cache-Control'], 'no-cache')
self.assertEqual(response['Pragma'], 'no-cache')
- request = self.factory.get('/', HTTP_IF_MODIFIED_SINCE=http_date(timegm(datetime.now().utctimetuple())))
+ # Old last-modified on GET, updated content is returned
+ if_modified_since = http_date(timegm((last_modified_date - timedelta(seconds=10)).utctimetuple()))
+ request = self.factory.get('/', HTTP_IF_MODIFIED_SINCE=if_modified_since)
response = resource(request)
- self.assertEqual(response.status_code, 204)
+ self.assertEqual(response.status_code, codes.ok)
- request = self.factory.get('/', HTTP_IF_MODIFIED_SINCE=http_date(timegm((last_modified_date-timedelta(seconds=-20)).utctimetuple())))
+ # Mimic future request on GET, resource not modified
+ if_modified_since = http_date(timegm((last_modified_date + timedelta(seconds=20)).utctimetuple()))
+ request = self.factory.get('/', HTTP_IF_MODIFIED_SINCE=if_modified_since)
response = resource(request)
- self.assertEqual(response.status_code, 304)
-
+ self.assertEqual(response.status_code, codes.not_modified)

0 comments on commit b4895d6

Please sign in to comment.