[2.2.x] Fixed #31866 -- Fixed locking proxy models in QuerySet.select…

…_for_update(of=()).

Backport of 6062616 from master

django/db/models/sql/compiler.py

@@ -948,7 +948,8 @@ def get_select_for_update_of_arguments(self):
         the query.
         """
         def _get_parent_klass_info(klass_info):
-            for parent_model, parent_link in klass_info['model']._meta.parents.items():
+            concrete_model = klass_info['model']._meta.concrete_model
+            for parent_model, parent_link in concrete_model._meta.parents.items():
                 parent_list = parent_model._meta.get_parent_list()
                 yield {
                     'model': parent_model,
@@ -973,8 +974,9 @@ def _get_first_selected_col_from_model(klass_info):
             select_fields is filled recursively, so it also contains fields
             from the parent models.
             """
+            concrete_model = klass_info['model']._meta.concrete_model
             for select_index in klass_info['select_fields']:
-                if self.select[select_index][0].target.model == klass_info['model']:
+                if self.select[select_index][0].target.model == concrete_model:
                     return self.select[select_index][0]
 
         def _get_field_choices():

docs/releases/2.2.16.txt

@@ -9,4 +9,7 @@ Django 2.2.16 fixes a data loss bug in 2.2.15.
 Bugfixes
 ========
 
-* ...
+* Fixed a data loss possibility in the
+  :meth:`~django.db.models.query.QuerySet.select_for_update()`. When using
+  related fields pointing to a proxy model in the ``of`` argument, the
+  corresponding model was not locked (:ticket:`31866`).

tests/select_for_update/models.py

@@ -23,6 +23,20 @@ class EUCity(models.Model):
     country = models.ForeignKey(EUCountry, models.CASCADE)
 
 
+class CountryProxy(Country):
+    class Meta:
+        proxy = True
+
+
+class CountryProxyProxy(CountryProxy):
+    class Meta:
+        proxy = True
+
+
+class CityCountryProxy(models.Model):
+    country = models.ForeignKey(CountryProxyProxy, models.CASCADE)
+
+
 class Person(models.Model):
     name = models.CharField(max_length=30)
     born = models.ForeignKey(City, models.CASCADE, related_name='+')

tests/select_for_update/tests.py

@@ -15,7 +15,9 @@
 )
 from django.test.utils import CaptureQueriesContext
 
-from .models import City, Country, EUCity, EUCountry, Person, PersonProfile
+from .models import (
+    City, CityCountryProxy, Country, EUCity, EUCountry, Person, PersonProfile,
+)
 
 
 class SelectForUpdateTests(TransactionTestCase):
@@ -195,6 +197,21 @@ def test_for_update_sql_multilevel_model_inheritance_ptr_generated_of(self):
         expected = [connection.ops.quote_name(value) for value in expected]
         self.assertTrue(self.has_for_update_sql(ctx.captured_queries, of=expected))
 
+    @skipUnlessDBFeature('has_select_for_update_of')
+    def test_for_update_sql_model_proxy_generated_of(self):
+        with transaction.atomic(), CaptureQueriesContext(connection) as ctx:
+            list(CityCountryProxy.objects.select_related(
+                'country',
+            ).select_for_update(
+                of=('country',),
+            ))
+        if connection.features.select_for_update_of_column:
+            expected = ['select_for_update_country"."entity_ptr_id']
+        else:
+            expected = ['select_for_update_country']
+        expected = [connection.ops.quote_name(value) for value in expected]
+        self.assertTrue(self.has_for_update_sql(ctx.captured_queries, of=expected))
+
     @skipUnlessDBFeature('has_select_for_update_of')
     def test_for_update_of_followed_by_values(self):
         with transaction.atomic():
@@ -353,6 +370,19 @@ def test_model_inheritance_of_argument_raises_error_ptr_in_choices(self):
             with transaction.atomic():
                 EUCountry.objects.select_for_update(of=('name',)).get()
 
+    @skipUnlessDBFeature('has_select_for_update', 'has_select_for_update_of')
+    def test_model_proxy_of_argument_raises_error_proxy_field_in_choices(self):
+        msg = (
+            'Invalid field name(s) given in select_for_update(of=(...)): '
+            'name. Only relational fields followed in the query are allowed. '
+            'Choices are: self, country, country__entity_ptr.'
+        )
+        with self.assertRaisesMessage(FieldError, msg):
+            with transaction.atomic():
+                CityCountryProxy.objects.select_related(
+                    'country',
+                ).select_for_update(of=('name',)).get()
+
     @skipUnlessDBFeature('has_select_for_update', 'has_select_for_update_of')
     def test_reverse_one_to_one_of_arguments(self):
         """