- added Widget and forms.VersionField tests

- added ConcurrencyMiddleware

Author: saxix

concurrency/__init__.py

@@ -2,7 +2,7 @@
 import datetime
 import os
 
-VERSION = __version__ = (0, 4, 0, 'beta', 9)
+VERSION = __version__ = (0, 4, 0, 'rc', 1)
 __author__ = 'sax'
 
 

concurrency/core.py

@@ -1,16 +1,20 @@
 from functools import update_wrapper
-from django.core.exceptions import ImproperlyConfigured
+from django.core.exceptions import ImproperlyConfigured, ValidationError
 from django.db import DatabaseError, connections, router
 from django.utils.translation import ugettext as _
 
 __all__ = []
 
 
-class RecordModifiedError(DatabaseError):
+class VersionChangedError(ValidationError):
     pass
 
-class Http409(Exception):
-    pass
+
+class RecordModifiedError(DatabaseError):
+    def __init__(self, *args, **kwargs):
+        self.target = kwargs.pop('target')
+        super(RecordModifiedError, self).__init__(*args, **kwargs)
+
 
 class InconsistencyError(DatabaseError):
     pass
@@ -59,7 +63,7 @@ def _select_lock(model_instance, version_value=None):
         NOWAIT = connections[alias].features.has_select_for_update_nowait
         entry = model_instance.__class__.objects.select_for_update(nowait=NOWAIT).filter(**kwargs)
         if not entry:
-            raise RecordModifiedError(_('Record has been modified'))
+            raise RecordModifiedError(_('Record has been modified'), target=model_instance)
     elif is_versioned:
         raise InconsistencyError(_('Version field is set (%s) but record has not `pk`.' % value))
 

concurrency/forms.py

@@ -39,6 +39,8 @@ class VersionWidget(HiddenInput):
 
     def render(self, name, value, attrs=None):
         ret = super(VersionWidget, self).render(name, value, attrs)
+        if value is None:
+            value = ''
         return mark_safe("%s<div>%s</div>" % (ret, value))
 
 
@@ -56,31 +58,21 @@ def __init__(self, *args, **kwargs):
         kwargs.pop('max_value', None)
         kwargs['required'] = True
         kwargs['initial'] = None
-        kwargs['widget'] = None
+        kwargs.setdefault('widget', HiddenInput)
         super(VersionField, self).__init__(*args, **kwargs)
 
-    def clean(self, value):
-        try:
-            # this check is here because some badly written test
-            if value not in (None, '', 'None'):
-                return int(self._signer.unsign(value))
-            return 0
-        except (BadSignature, ValueError):
-            raise SuspiciousOperation(_('Version number seems tampered'))
-
     def prepare_value(self, value):
         if value:
             return self._signer.sign(value)
         return None
 
     def to_python(self, value):
-        if value is None:
-            return 0
         try:
-            value = int(str(value))
-        except (ValueError, TypeError):
-            value = 0
-        return value
+            if value not in (None, '', 'None'):
+                return int(self._signer.unsign(value))
+            return 0
+        except (BadSignature, ValueError):
+            raise SuspiciousOperation(_('Version number seems tampered'))
 
     def widget_attrs(self, widget):
         return {}

concurrency/http.py

@@ -1,9 +1,12 @@
 # -*- coding: utf-8 -*-
-from concurrency.core import Http409
-from concurrency.http import HttpResponseConflict
+from django.core.urlresolvers import get_callable
+from concurrency.core import RecordModifiedError
+from concurrency.views import handler409
 
 
-class Http409Middleware(object):
+class ConcurrencyMiddleware(object):
     def process_exception(self, request, exception):
-        if isinstance(exception, Http409):
-            return HttpResponseConflict()
+        if isinstance(exception, RecordModifiedError):
+            callback = get_callable(handler409)
+
+            return callback(request, target=exception.target)

concurrency/middleware.py

@@ -1,7 +1,7 @@
 from concurrency.tests.all import *
 from concurrency.tests.contrib_admin import TestDjangoAdmin
 from concurrency.tests.forms import *
-
+from concurrency.tests.middleware import *
 
 try:
     from concurrency.tests.south_test import *

concurrency/tests/__init__.py

@@ -77,7 +77,7 @@ def test_creation(self):
                 '_continue': 1,
                 'date_field': '2010-09-01'}
 
-        response = self.client.post(url, data, follow=True)
+        self.client.post(url, data, follow=True)
         self.assertTrue(TestModel0.objects.filter(username='new_username').exists())
         self.assertGreater(TestModel0.objects.get(username='new_username').version, 0)
 

concurrency/tests/contrib_admin.py

@@ -1,10 +1,12 @@
+import warnings
 from django.core.exceptions import SuspiciousOperation
 from django.forms.models import modelform_factory
 from django.forms.widgets import HiddenInput, TextInput
 from django.utils.encoding import smart_str
-from django.utils.unittest.case import TestCase
-from concurrency.forms import ConcurrentForm, VersionField, VersionFieldSigner
+from django.test import TestCase
+from concurrency.forms import ConcurrentForm, VersionField, VersionFieldSigner, VersionWidget
 from concurrency.tests import TestModel0, TestIssue3Model
+from django.test.testcases import SimpleTestCase
 
 
 class DummySigner():
@@ -14,59 +16,73 @@ def sign(self, value):
     def unsign(self, signed_value):
         return smart_str(signed_value)
 
+
+class WidgetTest(TestCase):
+    def test(self):
+        w = VersionWidget()
+        self.assertHTMLEqual(w.render('ver', None),
+                             u'<input name="ver" type="hidden"/><div></div>')
+        self.assertHTMLEqual(w.render('ver', 100),
+                             u'<input name="ver" type="hidden" value="100"/><div>100</div>')
+
+
+class FieldTest(SimpleTestCase):
+
+    def setUp(self):
+        self.save_warnings_state()
+        warnings.filterwarnings('ignore', category=DeprecationWarning,
+                                module='django.core.validators')
+
+    def tearDown(self):
+        self.restore_warnings_state()
+
+    def test_with_dummy_signer(self):
+        f = VersionField(signer=DummySigner())
+        self.assertEqual(1, f.clean(1))
+        self.assertEqual(1, f.clean('1'))
+        self.assertEqual(0, f.clean(None))
+        self.assertEqual(0, f.clean(''))
+        self.assertRaisesMessage(SuspiciousOperation,
+                                "Version number seems tampered", f.clean, 'aa:bb')
+        self.assertRaisesMessage(SuspiciousOperation,
+                                 "Version number seems tampered", f.clean, 1.5)
+
+    def test(self):
+        f = VersionField()
+        self.assertEqual(1, f.clean(VersionFieldSigner().sign(1)))
+        self.assertEqual(1, f.clean(VersionFieldSigner().sign('1')))
+        self.assertEqual(0, f.clean(None))
+        self.assertEqual(0, f.clean(''))
+        self.assertRaisesMessage(SuspiciousOperation,
+                                "Version number seems tampered", f.clean, '100')
+        self.assertRaisesMessage(SuspiciousOperation,
+                                 "Version number seems tampered",
+                                 f.clean,
+                                 VersionFieldSigner().sign(1.5))
+
+
 class ConcurrentFormTest(TestCase):
     def test_version(self):
         Form = modelform_factory(TestModel0, ConcurrentForm)
         form = Form()
         self.assertIsInstance(form.fields['version'].widget, HiddenInput)
 
-    def test_signer(self):
+    def test_clean(self):
+        pass
+
+    def test_dummy_signer(self):
         obj, __ = TestIssue3Model.objects.get_or_create(username='aaa')
-        Form = modelform_factory(TestIssue3Model, type('xxx',(ConcurrentForm,), {'revision': VersionField(signer=DummySigner())}))
+        Form = modelform_factory(TestIssue3Model, type('xxx', (ConcurrentForm,), {'revision': VersionField(signer=DummySigner())}))
         data = {'id': 1,
                 'revision': obj.revision}
         form = Form(data, instance=obj)
         self.assertTrue(form.is_valid(), form.non_field_errors())
 
-    def test_signer2(self):
+    def test_signer(self):
         Form = modelform_factory(TestIssue3Model, ConcurrentForm)
         form = Form({'username': 'aaa'})
         self.assertTrue(form.is_valid(), form.non_field_errors())
 
-    def test_creation_None_version(self):
-        Form = modelform_factory(TestIssue3Model, type('xxx',(ConcurrentForm,), {}))
-        data = {'username': 'aaaq',
-                'last_name': None,
-                'revision': VersionField().prepare_value(None)}
-        form = Form(data)
-        self.assertTrue(form.is_valid())
-        obj = form.save()
-
-    def test_creation_empty_version(self):
-        Form = modelform_factory(TestIssue3Model, type('xxx',(ConcurrentForm,), {}))
-        data = {'username': 'aaaq',
-                'last_name': None,
-                'revision': VersionField().prepare_value('')}
-        form = Form(data)
-        self.assertTrue(form.is_valid())
-        obj = form.save()
-
-    def test_creation_0_version(self):
-        Form = modelform_factory(TestIssue3Model, type('xxx',(ConcurrentForm,), {}))
-        data = {'username': 'aaa',
-                'revision': VersionField().prepare_value(0)}
-        form = Form(data)
-        self.assertTrue(form.is_valid())
-        obj = form.save()
-
-    def test_creation_empty_version2(self):
-        obj, __ = TestIssue3Model.objects.get_or_create(username='aaa')
-        Form = modelform_factory(TestIssue3Model, type('xxx',(ConcurrentForm,), {'revision': VersionField(signer=DummySigner())}))
-        data = {'id': 1,
-                'revision': ''}
-        form = Form(data, instance=obj)
-        self.assertTrue(form.is_valid(), form.non_field_errors())
-
     def test_tamperig(self):
         obj, __ = TestIssue3Model.objects.get_or_create(username='aaa')
         Form = modelform_factory(TestIssue3Model, ConcurrentForm)

concurrency/tests/forms.py

@@ -0,0 +1,34 @@
+# -*- coding: utf-8 -*-
+from django.http import HttpRequest
+from django.test import TestCase
+from concurrency.core import RecordModifiedError
+from concurrency.middleware import ConcurrencyMiddleware
+from concurrency.tests import TestModel0
+
+
+class ConcurrencyMiddlewareTest(TestCase):
+    def setUp(self):
+        pass
+
+    def tearDown(self):
+        pass
+
+    def _get_request(self, path):
+        request = HttpRequest()
+        request.META = {
+            'SERVER_NAME': 'testserver',
+            'SERVER_PORT': 80,
+        }
+        request.path = request.path_info = "/middleware/%s" % path
+        return request
+
+    def test_process_exception(self):
+        """
+        Tests that RecordModifiedError is handled correctly.
+        """
+        m, __ = TestModel0.objects.get_or_create(username="New", last_name="1")
+        copy = TestModel0.objects.get(pk=m.pk)
+        copy.save()
+        request = self._get_request('/')
+        r = ConcurrencyMiddleware().process_exception(request, RecordModifiedError(target=m))
+        self.assertEqual(r.status_code, 409)

concurrency/tests/middleware.py

@@ -1,5 +1,3 @@
-import os
-
 SITE_ID = 1
 ROOT_URLCONF = 'demoproject.urls'
 SECRET_KEY = ';klkj;okj;lkn;lklj;lkj;kjmlliuewhy2ioqwjdkh'