Ticket #2259: 2259_r13817.diff

django/contrib/admin/options.py

@@ -600 +600 @@
         """
         Given a model instance save it to the database.
         """
-        obj.save()
+        obj.save(original_pk=form.initial.get(self.model._meta.pk.name))
 
     def save_formset(self, request, form, formset, change):
         """
@@ -680 +680 @@
         """
         opts = obj._meta
         pk_value = obj._get_pk_val()
+        same_obj_url = "../%s/" % pk_value
 
         msg = _('The %(name)s "%(obj)s" was changed successfully.') % {'name': force_unicode(opts.verbose_name), 'obj': force_unicode(obj)}
         if request.POST.has_key("_continue"):
             self.message_user(request, msg + ' ' + _("You may edit it again below."))
             if request.REQUEST.has_key('_popup'):
-                return HttpResponseRedirect(request.path + "?_popup=1")
+                return HttpResponseRedirect(same_obj_url + "?_popup=1")
             else:
-                return HttpResponseRedirect(request.path)
+                return HttpResponseRedirect(same_obj_url)
         elif request.POST.has_key("_saveasnew"):
             msg = _('The %(name)s "%(obj)s" was added successfully. You may edit it again below.') % {'name': force_unicode(opts.verbose_name), 'obj': obj}
             self.message_user(request, msg)
-            return HttpResponseRedirect("../%s/" % pk_value)
+            return HttpResponseRedirect(same_obj_url)
         elif request.POST.has_key("_addanother"):
             self.message_user(request, msg + ' ' + (_("You may add another %s below.") % force_unicode(opts.verbose_name)))
             return HttpResponseRedirect("../add/")
@@ -875 +876 @@
             return self.add_view(request, form_url='../add/')
 
         ModelForm = self.get_form(request, obj)
+
         formsets = []
         if request.method == 'POST':
             form = ModelForm(request.POST, request.FILES, instance=obj)

django/db/models/base.py

@@ -420 +420 @@
             return getattr(self, field_name)
         return getattr(self, field.attname)
 
-    def save(self, force_insert=False, force_update=False, using=None):
+    def save(self, force_insert=False, force_update=False, using=None, original_pk=None):
         """
         Saves the current instance. Override this in a subclass if you want to
         control the saving process.
@@ -428 +428 @@
         The 'force_insert' and 'force_update' parameters can be used to insist
         that the "save" must be an SQL insert or update (or equivalent for
         non-SQL backends), respectively. Normally, they should not be set.
+
+        The 'original_pk' argument allows updating the primary key of an
+        existing row. Assuming the actual instance pk attribute has already
+        been updated to the new value, passing in the original pk will update
+        that row.
         """
         if force_insert and force_update:
             raise ValueError("Cannot force both insert and updating in model saving.")
-        self.save_base(using=using, force_insert=force_insert, force_update=force_update)
+        if force_insert and original_pk is not None:
+            raise ValueError("Cannot force insert and provide an original primary key.")
+        self.save_base(using=using, force_insert=force_insert, force_update=force_update, original_pk=original_pk)
 
     save.alters_data = True
 
     def save_base(self, raw=False, cls=None, origin=None, force_insert=False,
-            force_update=False, using=None):
+            force_update=False, using=None, original_pk=None):
         """
         Does the heavy-lifting involved in saving. Subclasses shouldn't need to
         override this method. It's separate from save() in order to hide the
@@ -483 +490 @@
                 return
 
         if not meta.proxy:
-            non_pks = [f for f in meta.local_fields if not f.primary_key]
+            fields_to_update = meta.local_fields
 
             # First, try an UPDATE. If that doesn't update anything, do an INSERT.
             pk_val = self._get_pk_val(meta)
             pk_set = pk_val is not None
+
+            if original_pk is not None:
+                if not pk_set:
+                    raise ValueError("Cannot update existing row primary key to None")
+                pk_val = original_pk
+            else:
+                fields_to_update = [f for f in fields_to_update if not f.primary_key]
+
             record_exists = True
             manager = cls._base_manager
             if pk_set:
@@ -495 +510 @@
                 if (force_update or (not force_insert and
                         manager.using(using).filter(pk=pk_val).exists())):
                     # It does already exist, so do an UPDATE.
-                    if force_update or non_pks:
-                        values = [(f, None, (raw and getattr(self, f.attname) or f.pre_save(self, False))) for f in non_pks]
+                    if force_update or fields_to_update:
+                        values = [
+                            (f, None, (raw and getattr(self, f.attname) or f.pre_save(self, False)))
+                            for f in fields_to_update
+                        ]
                         rows = manager.using(using).filter(pk=pk_val)._update(values)
                         if force_update and not rows:
                             raise DatabaseError("Forced update did not affect any rows.")

tests/regressiontests/admin_views/tests.py

@@ -326 +326 @@
         response = self.client.post('/test_admin/admin/admin_views/person/1/', post_data)
         self.assertEqual(response.context['form_url'], '../add/')
 
+class UpdatePrimaryKeyTests(TestCase):
+    fixtures = ['admin-views-users.xml', 'string-primary-key.xml']
+    model_class = ModelWithStringPrimaryKey
+    model_pk = """abcdefghijklmnopqrstuvwxyz ABCDEFGHIJKLMNOPQRSTUVWXYZ 1234567890 -_.!~*'() ;/?:@&=+$, <>#%" {}|\^[]`"""
+
+    def setUp(self):
+        self.client.login(username='super', password='secret')
+        self.instance = self._get_by_pk(self.model_pk)
+
+    def tearDown(self):
+        self.client.logout()
+
+    def _get_by_pk(self, pk_val):
+        return self._manager.get(pk=pk_val)
+
+    @property
+    def _manager(self):
+        return self.model_class._default_manager
+
+    def _pk_name(self):
+        return self.model_class._meta.pk.attname
+
+    def _change_url(self, pk=None):
+        pk = pk or self.instance.pk
+        return ('/test_admin/admin/admin_views/%s/%s/'
+                % (self.model_class._meta.module_name,
+                   iri_to_uri(quote(pk))))
+
+    def test_update_pk(self):
+        post_data = {
+            self._pk_name(): 'new pk value',
+            }
+        count = self._manager.count()
+        response = self.client.post(self._change_url(), data=post_data)
+        self.assertEqual(response.status_code, 302)
+        self.assertEqual(self._manager.filter(pk='new pk value').count(), 1)
+        self.assertEqual(self._manager.count(), count)
+        self.assertFalse(self._manager.filter(pk=self.model_pk).exists())
+
+    def test_update_pk_continue_redirect(self):
+        post_data = {
+            self._pk_name(): 'new pk value',
+            '_continue': 'Save and continue editing',
+            }
+        response = self.client.post(self._change_url(), data=post_data)
+        self.assertRedirects(response, self._change_url('new pk value'))
+
 class CustomModelAdminTest(AdminViewBasicTest):
     urlbit = "admin2"
 

new file tests/regressiontests/model_save/models.py

@@ +1 @@
+from django.db import models
+
+class Beverage(models.Model):
+    id = models.IntegerField(primary_key=True)
+    name = models.CharField(max_length=100)
+
+class Train(models.Model):
+    name = models.CharField(max_length=100, primary_key=True)
+
+class TrainCar(models.Model):
+    train = models.ForeignKey(Train)
+    number = models.IntegerField()

new file tests/regressiontests/model_save/tests.py

@@ +1 @@
+from django.db import DatabaseError
+from django.test import TestCase, TransactionTestCase
+
+from models import Beverage, Train, TrainCar
+
+class UpdatePKTests(TestCase):
+    def setUp(self):
+        self.b = Beverage.objects.create(id=1, name='Water')
+
+    def test_update_pk(self):
+        self.b.id = 2
+        self.b.save(original_pk=1)
+        self.assertEqual(Beverage.objects.count(), 1)
+        self.assertEqual(Beverage.objects.filter(id=2).count(), 1)
+        self.assertEqual(Beverage.objects.filter(id=1).count(), 0)
+
+    def test_update_pk_to_none(self):
+        """
+        Trying to update a row to a PK of None is totally not legit.
+
+        """
+        self.b.id = None
+        self.assertRaises(ValueError, self.b.save, original_pk=1)
+
+    def test_bogus_original_pk(self):
+        """
+        Asking to update a row with a nonexistent original_pk results in a
+        DatabaseError (same as combining force_update with a nonexistent pk).
+
+        """
+        self.assertRaises(DatabaseError, self.b.save, original_pk=7)
+
+    def test_force_insert_and_original_pk(self):
+        """
+        Asking to update a row with an original_pk and force_insert is
+        nonsensical and results in a ValueError.
+
+        """
+        self.b.id = 2
+        self.assertRaises(ValueError, self.b.save, original_pk=1, force_insert=True)
+
+    def test_force_update_and_original_pk(self):
+        """
+        Combining force_update and original_pk is just redundant, not
+        problematic.
+
+        """
+        self.b.id = 2
+        self.b.save(original_pk=1, force_update=True)
+        self.assertEqual(Beverage.objects.count(), 1)
+        self.assertEqual(Beverage.objects.filter(id=2).count(), 1)
+        self.assertEqual(Beverage.objects.filter(id=1).count(), 0)
+
+    def test_update_pk_cascade(self):
+        """
+        Updating the primary key of an object that is the referent of
+        ForeignKeys needs to cascade the update to referring row, or
+        referential integrity is broken.
+
+        XXX This test currently fails because we have neither ON UPDATE CASCADE
+        support nor any emulation of it.
+
+        """
+        train = Train.objects.create(name='California Zephyr')
+        car = TrainCar.objects.create(train=train, number=12)
+
+        train.name = 'Empire Builder'
+        train.save(original_pk='California Zephyr')
+        car = TrainCar.objects.get(pk=car.pk)
+        self.assertEqual(car.train.name, 'Empire Builder')