1. import unittest

  2. 

  3. from django.core.exceptions import FieldError

  4. from django.db import IntegrityError, connection, transaction

  5. from django.db.models import CharField, Count, F, IntegerField, Max

  6. from django.db.models.functions import Abs, Concat, Lower

  7. from django.test import TestCase

  8. from django.test.utils import register_lookup

  9. 

  10. from .models import (

  11.     A,

  12.     B,

  13.     Bar,

  14.     D,

  15.     DataPoint,

  16.     Foo,

  17.     RelatedPoint,

  18.     UniqueNumber,

  19.     UniqueNumberChild,

  20. )

  21. 

  22. 

  23. class SimpleTest(TestCase):

  24.     @classmethod

  25.     def setUpTestData(cls):

  26.         cls.a1 = A.objects.create()

  27.         cls.a2 = A.objects.create()

  28.         for x in range(20):

  29.             B.objects.create(a=cls.a1)

  30.             D.objects.create(a=cls.a1)

  31. 

  32.     def test_nonempty_update(self):

  33.         """

  34.         Update changes the right number of rows for a nonempty queryset

  35.         """

  36.         num_updated = self.a1.b_set.update(y=100)

  37.         self.assertEqual(num_updated, 20)

  38.         cnt = B.objects.filter(y=100).count()

  39.         self.assertEqual(cnt, 20)

  40. 

  41.     def test_empty_update(self):

  42.         """

  43.         Update changes the right number of rows for an empty queryset

  44.         """

  45.         num_updated = self.a2.b_set.update(y=100)

  46.         self.assertEqual(num_updated, 0)

  47.         cnt = B.objects.filter(y=100).count()

  48.         self.assertEqual(cnt, 0)

  49. 

  50.     def test_nonempty_update_with_inheritance(self):

  51.         """

  52.         Update changes the right number of rows for an empty queryset

  53.         when the update affects only a base table

  54.         """

  55.         num_updated = self.a1.d_set.update(y=100)

  56.         self.assertEqual(num_updated, 20)

  57.         cnt = D.objects.filter(y=100).count()

  58.         self.assertEqual(cnt, 20)

  59. 

  60.     def test_empty_update_with_inheritance(self):

  61.         """

  62.         Update changes the right number of rows for an empty queryset

  63.         when the update affects only a base table

  64.         """

  65.         num_updated = self.a2.d_set.update(y=100)

  66.         self.assertEqual(num_updated, 0)

  67.         cnt = D.objects.filter(y=100).count()

  68.         self.assertEqual(cnt, 0)

  69. 

  70.     def test_foreign_key_update_with_id(self):

  71.         """

  72.         Update works using <field>_id for foreign keys

  73.         """

  74.         num_updated = self.a1.d_set.update(a_id=self.a2)

  75.         self.assertEqual(num_updated, 20)

  76.         self.assertEqual(self.a2.d_set.count(), 20)

  77. 

  78. 

  79. class AdvancedTests(TestCase):

  80.     @classmethod

  81.     def setUpTestData(cls):

  82.         cls.d0 = DataPoint.objects.create(name="d0", value="apple")

  83.         cls.d2 = DataPoint.objects.create(name="d2", value="banana")

  84.         cls.d3 = DataPoint.objects.create(name="d3", value="banana")

  85.         cls.r1 = RelatedPoint.objects.create(name="r1", data=cls.d3)

  86. 

  87.     def test_update(self):

  88.         """

  89.         Objects are updated by first filtering the candidates into a queryset

  90.         and then calling the update() method. It executes immediately and

  91.         returns nothing.

  92.         """

  93.         resp = DataPoint.objects.filter(value="apple").update(name="d1")

  94.         self.assertEqual(resp, 1)

  95.         resp = DataPoint.objects.filter(value="apple")

  96.         self.assertEqual(list(resp), [self.d0])

  97. 

  98.     def test_update_multiple_objects(self):

  99.         """

  100.         We can update multiple objects at once.

  101.         """

  102.         resp = DataPoint.objects.filter(value="banana").update(value="pineapple")

  103.         self.assertEqual(resp, 2)

  104.         self.assertEqual(DataPoint.objects.get(name="d2").value, "pineapple")

  105. 

  106.     def test_update_fk(self):

  107.         """

  108.         Foreign key fields can also be updated, although you can only update

  109.         the object referred to, not anything inside the related object.

  110.         """

  111.         resp = RelatedPoint.objects.filter(name="r1").update(data=self.d0)

  112.         self.assertEqual(resp, 1)

  113.         resp = RelatedPoint.objects.filter(data__name="d0")

  114.         self.assertEqual(list(resp), [self.r1])

  115. 

  116.     def test_update_multiple_fields(self):

  117.         """

  118.         Multiple fields can be updated at once

  119.         """

  120.         resp = DataPoint.objects.filter(value="apple").update(

  121.             value="fruit", another_value="peach"

  122.         )

  123.         self.assertEqual(resp, 1)

  124.         d = DataPoint.objects.get(name="d0")

  125.         self.assertEqual(d.value, "fruit")

  126.         self.assertEqual(d.another_value, "peach")

  127. 

  128.     def test_update_all(self):

  129.         """

  130.         In the rare case you want to update every instance of a model, update()

  131.         is also a manager method.

  132.         """

  133.         self.assertEqual(DataPoint.objects.update(value="thing"), 3)

  134.         resp = DataPoint.objects.values("value").distinct()

  135.         self.assertEqual(list(resp), [{"value": "thing"}])

  136. 

  137.     def test_update_slice_fail(self):

  138.         """

  139.         We do not support update on already sliced query sets.

  140.         """

  141.         method = DataPoint.objects.all()[:2].update

  142.         msg = "Cannot update a query once a slice has been taken."

  143.         with self.assertRaisesMessage(TypeError, msg):

  144.             method(another_value="another thing")

  145. 

  146.     def test_update_respects_to_field(self):

  147.         """

  148.         Update of an FK field which specifies a to_field works.

  149.         """

  150.         a_foo = Foo.objects.create(target="aaa")

  151.         b_foo = Foo.objects.create(target="bbb")

  152.         bar = Bar.objects.create(foo=a_foo)

  153.         self.assertEqual(bar.foo_id, a_foo.target)

  154.         bar_qs = Bar.objects.filter(pk=bar.pk)

  155.         self.assertEqual(bar_qs[0].foo_id, a_foo.target)

  156.         bar_qs.update(foo=b_foo)

  157.         self.assertEqual(bar_qs[0].foo_id, b_foo.target)

  158. 

  159.     def test_update_m2m_field(self):

  160.         msg = (

  161.             "Cannot update model field "

  162.             "<django.db.models.fields.related.ManyToManyField: m2m_foo> "

  163.             "(only non-relations and foreign keys permitted)."

  164.         )

  165.         with self.assertRaisesMessage(FieldError, msg):

  166.             Bar.objects.update(m2m_foo="whatever")

  167. 

  168.     def test_update_transformed_field(self):

  169.         A.objects.create(x=5)

  170.         A.objects.create(x=-6)

  171.         with register_lookup(IntegerField, Abs):

  172.             A.objects.update(x=F("x__abs"))

  173.             self.assertCountEqual(A.objects.values_list("x", flat=True), [5, 6])

  174. 

  175.     def test_update_annotated_queryset(self):

  176.         """

  177.         Update of a queryset that's been annotated.

  178.         """

  179.         # Trivial annotated update

  180.         qs = DataPoint.objects.annotate(alias=F("value"))

  181.         self.assertEqual(qs.update(another_value="foo"), 3)

  182.         # Update where annotation is used for filtering

  183.         qs = DataPoint.objects.annotate(alias=F("value")).filter(alias="apple")

  184.         self.assertEqual(qs.update(another_value="foo"), 1)

  185.         # Update where annotation is used in update parameters

  186.         qs = DataPoint.objects.annotate(alias=F("value"))

  187.         self.assertEqual(qs.update(another_value=F("alias")), 3)

  188.         # Update where aggregation annotation is used in update parameters

  189.         qs = DataPoint.objects.annotate(max=Max("value"))

  190.         msg = (

  191.             "Aggregate functions are not allowed in this query "

  192.             "(another_value=Max(Col(update_datapoint, update.DataPoint.value)))."

  193.         )

  194.         with self.assertRaisesMessage(FieldError, msg):

  195.             qs.update(another_value=F("max"))

  196. 

  197.     def test_update_annotated_multi_table_queryset(self):

  198.         """

  199.         Update of a queryset that's been annotated and involves multiple tables.

  200.         """

  201.         # Trivial annotated update

  202.         qs = DataPoint.objects.annotate(related_count=Count("relatedpoint"))

  203.         self.assertEqual(qs.update(value="Foo"), 3)

  204.         # Update where annotation is used for filtering

  205.         qs = DataPoint.objects.annotate(related_count=Count("relatedpoint"))

  206.         self.assertEqual(qs.filter(related_count=1).update(value="Foo"), 1)

  207.         # Update where aggregation annotation is used in update parameters

  208.         qs = RelatedPoint.objects.annotate(max=Max("data__value"))

  209.         msg = "Joined field references are not permitted in this query"

  210.         with self.assertRaisesMessage(FieldError, msg):

  211.             qs.update(name=F("max"))

  212. 

  213.     def test_update_with_joined_field_annotation(self):

  214.         msg = "Joined field references are not permitted in this query"

  215.         with register_lookup(CharField, Lower):

  216.             for annotation in (

  217.                 F("data__name"),

  218.                 F("data__name__lower"),

  219.                 Lower("data__name"),

  220.                 Concat("data__name", "data__value"),

  221.             ):

  222.                 with self.subTest(annotation=annotation):

  223.                     with self.assertRaisesMessage(FieldError, msg):

  224.                         RelatedPoint.objects.annotate(

  225.                             new_name=annotation,

  226.                         ).update(name=F("new_name"))

  227. 

  228.     def test_update_ordered_by_m2m_aggregation_annotation(self):

  229.         msg = (

  230.             "Cannot update when ordering by an aggregate: "

  231.             "Count(Col(update_bar_m2m_foo, update.Bar_m2m_foo.foo))"

  232.         )

  233.         with self.assertRaisesMessage(FieldError, msg):

  234.             Bar.objects.annotate(m2m_count=Count("m2m_foo")).order_by(

  235.                 "m2m_count"

  236.             ).update(x=2)

  237. 

  238.     def test_update_ordered_by_inline_m2m_annotation(self):

  239.         foo = Foo.objects.create(target="test")

  240.         Bar.objects.create(foo=foo)

  241. 

  242.         Bar.objects.order_by(Abs("m2m_foo")).update(x=2)

  243.         self.assertEqual(Bar.objects.get().x, 2)

  244. 

  245.     def test_update_ordered_by_m2m_annotation(self):

  246.         foo = Foo.objects.create(target="test")

  247.         Bar.objects.create(foo=foo)

  248. 

  249.         Bar.objects.annotate(abs_id=Abs("m2m_foo")).order_by("abs_id").update(x=3)

  250.         self.assertEqual(Bar.objects.get().x, 3)

  251. 

  252. 

  253. @unittest.skipUnless(

  254.     connection.vendor == "mysql",

  255.     "UPDATE...ORDER BY syntax is supported on MySQL/MariaDB",

  256. )

  257. class MySQLUpdateOrderByTest(TestCase):

  258.     """Update field with a unique constraint using an ordered queryset."""

  259. 

  260.     @classmethod

  261.     def setUpTestData(cls):

  262.         UniqueNumber.objects.create(number=1)

  263.         UniqueNumber.objects.create(number=2)

  264. 

  265.     def test_order_by_update_on_unique_constraint(self):

  266.         tests = [

  267.             ("-number", "id"),

  268.             (F("number").desc(), "id"),

  269.             (F("number") * -1, "id"),

  270.         ]

  271.         for ordering in tests:

  272.             with self.subTest(ordering=ordering), transaction.atomic():

  273.                 updated = UniqueNumber.objects.order_by(*ordering).update(

  274.                     number=F("number") + 1,

  275.                 )

  276.                 self.assertEqual(updated, 2)

  277. 

  278.     def test_order_by_update_on_unique_constraint_annotation(self):

  279.         updated = (

  280.             UniqueNumber.objects.annotate(number_inverse=F("number").desc())

  281.             .order_by("number_inverse")

  282.             .update(number=F("number") + 1)

  283.         )

  284.         self.assertEqual(updated, 2)

  285. 

  286.     def test_order_by_update_on_parent_unique_constraint(self):

  287.         # Ordering by inherited fields is omitted because joined fields cannot

  288.         # be used in the ORDER BY clause.

  289.         UniqueNumberChild.objects.create(number=3)

  290.         UniqueNumberChild.objects.create(number=4)

  291.         with self.assertRaises(IntegrityError):

  292.             UniqueNumberChild.objects.order_by("number").update(

  293.                 number=F("number") + 1,

  294.             )

  295. 

  296.     def test_order_by_update_on_related_field(self):

  297.         # Ordering by related fields is omitted because joined fields cannot be

  298.         # used in the ORDER BY clause.

  299.         data = DataPoint.objects.create(name="d0", value="apple")

  300.         related = RelatedPoint.objects.create(name="r0", data=data)

  301.         with self.assertNumQueries(1) as ctx:

  302.             updated = RelatedPoint.objects.order_by("data__name").update(name="new")

  303.         sql = ctx.captured_queries[0]["sql"]

  304.         self.assertNotIn("ORDER BY", sql)

  305.         self.assertEqual(updated, 1)

  306.         related.refresh_from_db()

  307.         self.assertEqual(related.name, "new")
About Sign Up Leave Feedback Contact