1. import unittest

  2. from unittest import mock

  3. 

  4. from migrations.test_base import OperationTestBase

  5. 

  6. from django.db import IntegrityError, NotSupportedError, connection, transaction

  7. from django.db.migrations.state import ProjectState

  8. from django.db.models import CheckConstraint, Index, Q, UniqueConstraint

  9. from django.db.utils import ProgrammingError

  10. from django.test import modify_settings, override_settings, skipUnlessDBFeature

  11. from django.test.utils import CaptureQueriesContext

  12. 

  13. from . import PostgreSQLTestCase

  14. 

  15. try:

  16.     from django.contrib.postgres.indexes import BrinIndex, BTreeIndex

  17.     from django.contrib.postgres.operations import (

  18.         AddConstraintNotValid,

  19.         AddIndexConcurrently,

  20.         BloomExtension,

  21.         CreateCollation,

  22.         CreateExtension,

  23.         RemoveCollation,

  24.         RemoveIndexConcurrently,

  25.         ValidateConstraint,

  26.     )

  27. except ImportError:

  28.     pass

  29. 

  30. 

  31. @unittest.skipUnless(connection.vendor == "postgresql", "PostgreSQL specific tests.")

  32. @modify_settings(INSTALLED_APPS={"append": "migrations"})

  33. class AddIndexConcurrentlyTests(OperationTestBase):

  34.     app_label = "test_add_concurrently"

  35. 

  36.     def test_requires_atomic_false(self):

  37.         project_state = self.set_up_test_model(self.app_label)

  38.         new_state = project_state.clone()

  39.         operation = AddIndexConcurrently(

  40.             "Pony",

  41.             Index(fields=["pink"], name="pony_pink_idx"),

  42.         )

  43.         msg = (

  44.             "The AddIndexConcurrently operation cannot be executed inside "

  45.             "a transaction (set atomic = False on the migration)."

  46.         )

  47.         with self.assertRaisesMessage(NotSupportedError, msg):

  48.             with connection.schema_editor(atomic=True) as editor:

  49.                 operation.database_forwards(

  50.                     self.app_label, editor, project_state, new_state

  51.                 )

  52. 

  53.     def test_add(self):

  54.         project_state = self.set_up_test_model(self.app_label, index=False)

  55.         table_name = "%s_pony" % self.app_label

  56.         index = Index(fields=["pink"], name="pony_pink_idx")

  57.         new_state = project_state.clone()

  58.         operation = AddIndexConcurrently("Pony", index)

  59.         self.assertEqual(

  60.             operation.describe(),

  61.             "Concurrently create index pony_pink_idx on field(s) pink of model Pony",

  62.         )

  63.         operation.state_forwards(self.app_label, new_state)

  64.         self.assertEqual(

  65.             len(new_state.models[self.app_label, "pony"].options["indexes"]), 1

  66.         )

  67.         self.assertIndexNotExists(table_name, ["pink"])

  68.         # Add index.

  69.         with connection.schema_editor(atomic=False) as editor:

  70.             operation.database_forwards(

  71.                 self.app_label, editor, project_state, new_state

  72.             )

  73.         self.assertIndexExists(table_name, ["pink"])

  74.         # Reversal.

  75.         with connection.schema_editor(atomic=False) as editor:

  76.             operation.database_backwards(

  77.                 self.app_label, editor, new_state, project_state

  78.             )

  79.         self.assertIndexNotExists(table_name, ["pink"])

  80.         # Deconstruction.

  81.         name, args, kwargs = operation.deconstruct()

  82.         self.assertEqual(name, "AddIndexConcurrently")

  83.         self.assertEqual(args, [])

  84.         self.assertEqual(kwargs, {"model_name": "Pony", "index": index})

  85. 

  86.     def test_add_other_index_type(self):

  87.         project_state = self.set_up_test_model(self.app_label, index=False)

  88.         table_name = "%s_pony" % self.app_label

  89.         new_state = project_state.clone()

  90.         operation = AddIndexConcurrently(

  91.             "Pony",

  92.             BrinIndex(fields=["pink"], name="pony_pink_brin_idx"),

  93.         )

  94.         self.assertIndexNotExists(table_name, ["pink"])

  95.         # Add index.

  96.         with connection.schema_editor(atomic=False) as editor:

  97.             operation.database_forwards(

  98.                 self.app_label, editor, project_state, new_state

  99.             )

  100.         self.assertIndexExists(table_name, ["pink"], index_type="brin")

  101.         # Reversal.

  102.         with connection.schema_editor(atomic=False) as editor:

  103.             operation.database_backwards(

  104.                 self.app_label, editor, new_state, project_state

  105.             )

  106.         self.assertIndexNotExists(table_name, ["pink"])

  107. 

  108.     def test_add_with_options(self):

  109.         project_state = self.set_up_test_model(self.app_label, index=False)

  110.         table_name = "%s_pony" % self.app_label

  111.         new_state = project_state.clone()

  112.         index = BTreeIndex(fields=["pink"], name="pony_pink_btree_idx", fillfactor=70)

  113.         operation = AddIndexConcurrently("Pony", index)

  114.         self.assertIndexNotExists(table_name, ["pink"])

  115.         # Add index.

  116.         with connection.schema_editor(atomic=False) as editor:

  117.             operation.database_forwards(

  118.                 self.app_label, editor, project_state, new_state

  119.             )

  120.         self.assertIndexExists(table_name, ["pink"], index_type="btree")

  121.         # Reversal.

  122.         with connection.schema_editor(atomic=False) as editor:

  123.             operation.database_backwards(

  124.                 self.app_label, editor, new_state, project_state

  125.             )

  126.         self.assertIndexNotExists(table_name, ["pink"])

  127. 

  128. 

  129. @unittest.skipUnless(connection.vendor == "postgresql", "PostgreSQL specific tests.")

  130. @modify_settings(INSTALLED_APPS={"append": "migrations"})

  131. class RemoveIndexConcurrentlyTests(OperationTestBase):

  132.     app_label = "test_rm_concurrently"

  133. 

  134.     def test_requires_atomic_false(self):

  135.         project_state = self.set_up_test_model(self.app_label, index=True)

  136.         new_state = project_state.clone()

  137.         operation = RemoveIndexConcurrently("Pony", "pony_pink_idx")

  138.         msg = (

  139.             "The RemoveIndexConcurrently operation cannot be executed inside "

  140.             "a transaction (set atomic = False on the migration)."

  141.         )

  142.         with self.assertRaisesMessage(NotSupportedError, msg):

  143.             with connection.schema_editor(atomic=True) as editor:

  144.                 operation.database_forwards(

  145.                     self.app_label, editor, project_state, new_state

  146.                 )

  147. 

  148.     def test_remove(self):

  149.         project_state = self.set_up_test_model(self.app_label, index=True)

  150.         table_name = "%s_pony" % self.app_label

  151.         self.assertTableExists(table_name)

  152.         new_state = project_state.clone()

  153.         operation = RemoveIndexConcurrently("Pony", "pony_pink_idx")

  154.         self.assertEqual(

  155.             operation.describe(),

  156.             "Concurrently remove index pony_pink_idx from Pony",

  157.         )

  158.         operation.state_forwards(self.app_label, new_state)

  159.         self.assertEqual(

  160.             len(new_state.models[self.app_label, "pony"].options["indexes"]), 0

  161.         )

  162.         self.assertIndexExists(table_name, ["pink"])

  163.         # Remove index.

  164.         with connection.schema_editor(atomic=False) as editor:

  165.             operation.database_forwards(

  166.                 self.app_label, editor, project_state, new_state

  167.             )

  168.         self.assertIndexNotExists(table_name, ["pink"])

  169.         # Reversal.

  170.         with connection.schema_editor(atomic=False) as editor:

  171.             operation.database_backwards(

  172.                 self.app_label, editor, new_state, project_state

  173.             )

  174.         self.assertIndexExists(table_name, ["pink"])

  175.         # Deconstruction.

  176.         name, args, kwargs = operation.deconstruct()

  177.         self.assertEqual(name, "RemoveIndexConcurrently")

  178.         self.assertEqual(args, [])

  179.         self.assertEqual(kwargs, {"model_name": "Pony", "name": "pony_pink_idx"})

  180. 

  181. 

  182. class NoMigrationRouter:

  183.     def allow_migrate(self, db, app_label, **hints):

  184.         return False

  185. 

  186. 

  187. @unittest.skipUnless(connection.vendor == "postgresql", "PostgreSQL specific tests.")

  188. class CreateExtensionTests(PostgreSQLTestCase):

  189.     app_label = "test_allow_create_extention"

  190. 

  191.     @override_settings(DATABASE_ROUTERS=[NoMigrationRouter()])

  192.     def test_no_allow_migrate(self):

  193.         operation = CreateExtension("tablefunc")

  194.         project_state = ProjectState()

  195.         new_state = project_state.clone()

  196.         # Don't create an extension.

  197.         with CaptureQueriesContext(connection) as captured_queries:

  198.             with connection.schema_editor(atomic=False) as editor:

  199.                 operation.database_forwards(

  200.                     self.app_label, editor, project_state, new_state

  201.                 )

  202.         self.assertEqual(len(captured_queries), 0)

  203.         # Reversal.

  204.         with CaptureQueriesContext(connection) as captured_queries:

  205.             with connection.schema_editor(atomic=False) as editor:

  206.                 operation.database_backwards(

  207.                     self.app_label, editor, new_state, project_state

  208.                 )

  209.         self.assertEqual(len(captured_queries), 0)

  210. 

  211.     def test_allow_migrate(self):

  212.         operation = CreateExtension("tablefunc")

  213.         self.assertEqual(

  214.             operation.migration_name_fragment, "create_extension_tablefunc"

  215.         )

  216.         project_state = ProjectState()

  217.         new_state = project_state.clone()

  218.         # Create an extension.

  219.         with CaptureQueriesContext(connection) as captured_queries:

  220.             with connection.schema_editor(atomic=False) as editor:

  221.                 operation.database_forwards(

  222.                     self.app_label, editor, project_state, new_state

  223.                 )

  224.         self.assertEqual(len(captured_queries), 4)

  225.         self.assertIn("CREATE EXTENSION IF NOT EXISTS", captured_queries[1]["sql"])

  226.         # Reversal.

  227.         with CaptureQueriesContext(connection) as captured_queries:

  228.             with connection.schema_editor(atomic=False) as editor:

  229.                 operation.database_backwards(

  230.                     self.app_label, editor, new_state, project_state

  231.                 )

  232.         self.assertEqual(len(captured_queries), 2)

  233.         self.assertIn("DROP EXTENSION IF EXISTS", captured_queries[1]["sql"])

  234. 

  235.     def test_create_existing_extension(self):

  236.         operation = BloomExtension()

  237.         self.assertEqual(operation.migration_name_fragment, "create_extension_bloom")

  238.         project_state = ProjectState()

  239.         new_state = project_state.clone()

  240.         # Don't create an existing extension.

  241.         with CaptureQueriesContext(connection) as captured_queries:

  242.             with connection.schema_editor(atomic=False) as editor:

  243.                 operation.database_forwards(

  244.                     self.app_label, editor, project_state, new_state

  245.                 )

  246.         self.assertEqual(len(captured_queries), 3)

  247.         self.assertIn("SELECT", captured_queries[0]["sql"])

  248. 

  249.     def test_drop_nonexistent_extension(self):

  250.         operation = CreateExtension("tablefunc")

  251.         project_state = ProjectState()

  252.         new_state = project_state.clone()

  253.         # Don't drop a nonexistent extension.

  254.         with CaptureQueriesContext(connection) as captured_queries:

  255.             with connection.schema_editor(atomic=False) as editor:

  256.                 operation.database_backwards(

  257.                     self.app_label, editor, project_state, new_state

  258.                 )

  259.         self.assertEqual(len(captured_queries), 1)

  260.         self.assertIn("SELECT", captured_queries[0]["sql"])

  261. 

  262. 

  263. @unittest.skipUnless(connection.vendor == "postgresql", "PostgreSQL specific tests.")

  264. class CreateCollationTests(PostgreSQLTestCase):

  265.     app_label = "test_allow_create_collation"

  266. 

  267.     @override_settings(DATABASE_ROUTERS=[NoMigrationRouter()])

  268.     def test_no_allow_migrate(self):

  269.         operation = CreateCollation("C_test", locale="C")

  270.         project_state = ProjectState()

  271.         new_state = project_state.clone()

  272.         # Don't create a collation.

  273.         with CaptureQueriesContext(connection) as captured_queries:

  274.             with connection.schema_editor(atomic=False) as editor:

  275.                 operation.database_forwards(

  276.                     self.app_label, editor, project_state, new_state

  277.                 )

  278.         self.assertEqual(len(captured_queries), 0)

  279.         # Reversal.

  280.         with CaptureQueriesContext(connection) as captured_queries:

  281.             with connection.schema_editor(atomic=False) as editor:

  282.                 operation.database_backwards(

  283.                     self.app_label, editor, new_state, project_state

  284.                 )

  285.         self.assertEqual(len(captured_queries), 0)

  286. 

  287.     def test_create(self):

  288.         operation = CreateCollation("C_test", locale="C")

  289.         self.assertEqual(operation.migration_name_fragment, "create_collation_c_test")

  290.         self.assertEqual(operation.describe(), "Create collation C_test")

  291.         project_state = ProjectState()

  292.         new_state = project_state.clone()

  293.         # Create a collation.

  294.         with CaptureQueriesContext(connection) as captured_queries:

  295.             with connection.schema_editor(atomic=False) as editor:

  296.                 operation.database_forwards(

  297.                     self.app_label, editor, project_state, new_state

  298.                 )

  299.         self.assertEqual(len(captured_queries), 1)

  300.         self.assertIn("CREATE COLLATION", captured_queries[0]["sql"])

  301.         # Creating the same collation raises an exception.

  302.         with self.assertRaisesMessage(ProgrammingError, "already exists"):

  303.             with connection.schema_editor(atomic=True) as editor:

  304.                 operation.database_forwards(

  305.                     self.app_label, editor, project_state, new_state

  306.                 )

  307.         # Reversal.

  308.         with CaptureQueriesContext(connection) as captured_queries:

  309.             with connection.schema_editor(atomic=False) as editor:

  310.                 operation.database_backwards(

  311.                     self.app_label, editor, new_state, project_state

  312.                 )

  313.         self.assertEqual(len(captured_queries), 1)

  314.         self.assertIn("DROP COLLATION", captured_queries[0]["sql"])

  315.         # Deconstruction.

  316.         name, args, kwargs = operation.deconstruct()

  317.         self.assertEqual(name, "CreateCollation")

  318.         self.assertEqual(args, [])

  319.         self.assertEqual(kwargs, {"name": "C_test", "locale": "C"})

  320. 

  321.     @skipUnlessDBFeature("supports_non_deterministic_collations")

  322.     def test_create_non_deterministic_collation(self):

  323.         operation = CreateCollation(

  324.             "case_insensitive_test",

  325.             "und-u-ks-level2",

  326.             provider="icu",

  327.             deterministic=False,

  328.         )

  329.         project_state = ProjectState()

  330.         new_state = project_state.clone()

  331.         # Create a collation.

  332.         with CaptureQueriesContext(connection) as captured_queries:

  333.             with connection.schema_editor(atomic=False) as editor:

  334.                 operation.database_forwards(

  335.                     self.app_label, editor, project_state, new_state

  336.                 )

  337.         self.assertEqual(len(captured_queries), 1)

  338.         self.assertIn("CREATE COLLATION", captured_queries[0]["sql"])

  339.         # Reversal.

  340.         with CaptureQueriesContext(connection) as captured_queries:

  341.             with connection.schema_editor(atomic=False) as editor:

  342.                 operation.database_backwards(

  343.                     self.app_label, editor, new_state, project_state

  344.                 )

  345.         self.assertEqual(len(captured_queries), 1)

  346.         self.assertIn("DROP COLLATION", captured_queries[0]["sql"])

  347.         # Deconstruction.

  348.         name, args, kwargs = operation.deconstruct()

  349.         self.assertEqual(name, "CreateCollation")

  350.         self.assertEqual(args, [])

  351.         self.assertEqual(

  352.             kwargs,

  353.             {

  354.                 "name": "case_insensitive_test",

  355.                 "locale": "und-u-ks-level2",

  356.                 "provider": "icu",

  357.                 "deterministic": False,

  358.             },

  359.         )

  360. 

  361.     def test_create_collation_alternate_provider(self):

  362.         operation = CreateCollation(

  363.             "german_phonebook_test",

  364.             provider="icu",

  365.             locale="de-u-co-phonebk",

  366.         )

  367.         project_state = ProjectState()

  368.         new_state = project_state.clone()

  369.         # Create an collation.

  370.         with CaptureQueriesContext(connection) as captured_queries:

  371.             with connection.schema_editor(atomic=False) as editor:

  372.                 operation.database_forwards(

  373.                     self.app_label, editor, project_state, new_state

  374.                 )

  375.         self.assertEqual(len(captured_queries), 1)

  376.         self.assertIn("CREATE COLLATION", captured_queries[0]["sql"])

  377.         # Reversal.

  378.         with CaptureQueriesContext(connection) as captured_queries:

  379.             with connection.schema_editor(atomic=False) as editor:

  380.                 operation.database_backwards(

  381.                     self.app_label, editor, new_state, project_state

  382.                 )

  383.         self.assertEqual(len(captured_queries), 1)

  384.         self.assertIn("DROP COLLATION", captured_queries[0]["sql"])

  385. 

  386.     def test_nondeterministic_collation_not_supported(self):

  387.         operation = CreateCollation(

  388.             "case_insensitive_test",

  389.             provider="icu",

  390.             locale="und-u-ks-level2",

  391.             deterministic=False,

  392.         )

  393.         project_state = ProjectState()

  394.         new_state = project_state.clone()

  395.         msg = "Non-deterministic collations require PostgreSQL 12+."

  396.         with connection.schema_editor(atomic=False) as editor:

  397.             with mock.patch(

  398.                 "django.db.backends.postgresql.features.DatabaseFeatures."

  399.                 "supports_non_deterministic_collations",

  400.                 False,

  401.             ):

  402.                 with self.assertRaisesMessage(NotSupportedError, msg):

  403.                     operation.database_forwards(

  404.                         self.app_label, editor, project_state, new_state

  405.                     )

  406. 

  407. 

  408. @unittest.skipUnless(connection.vendor == "postgresql", "PostgreSQL specific tests.")

  409. class RemoveCollationTests(PostgreSQLTestCase):

  410.     app_label = "test_allow_remove_collation"

  411. 

  412.     @override_settings(DATABASE_ROUTERS=[NoMigrationRouter()])

  413.     def test_no_allow_migrate(self):

  414.         operation = RemoveCollation("C_test", locale="C")

  415.         project_state = ProjectState()

  416.         new_state = project_state.clone()

  417.         # Don't create a collation.

  418.         with CaptureQueriesContext(connection) as captured_queries:

  419.             with connection.schema_editor(atomic=False) as editor:

  420.                 operation.database_forwards(

  421.                     self.app_label, editor, project_state, new_state

  422.                 )

  423.         self.assertEqual(len(captured_queries), 0)

  424.         # Reversal.

  425.         with CaptureQueriesContext(connection) as captured_queries:

  426.             with connection.schema_editor(atomic=False) as editor:

  427.                 operation.database_backwards(

  428.                     self.app_label, editor, new_state, project_state

  429.                 )

  430.         self.assertEqual(len(captured_queries), 0)

  431. 

  432.     def test_remove(self):

  433.         operation = CreateCollation("C_test", locale="C")

  434.         project_state = ProjectState()

  435.         new_state = project_state.clone()

  436.         with connection.schema_editor(atomic=False) as editor:

  437.             operation.database_forwards(

  438.                 self.app_label, editor, project_state, new_state

  439.             )

  440. 

  441.         operation = RemoveCollation("C_test", locale="C")

  442.         self.assertEqual(operation.migration_name_fragment, "remove_collation_c_test")

  443.         self.assertEqual(operation.describe(), "Remove collation C_test")

  444.         project_state = ProjectState()

  445.         new_state = project_state.clone()

  446.         # Remove a collation.

  447.         with CaptureQueriesContext(connection) as captured_queries:

  448.             with connection.schema_editor(atomic=False) as editor:

  449.                 operation.database_forwards(

  450.                     self.app_label, editor, project_state, new_state

  451.                 )

  452.         self.assertEqual(len(captured_queries), 1)

  453.         self.assertIn("DROP COLLATION", captured_queries[0]["sql"])

  454.         # Removing a nonexistent collation raises an exception.

  455.         with self.assertRaisesMessage(ProgrammingError, "does not exist"):

  456.             with connection.schema_editor(atomic=True) as editor:

  457.                 operation.database_forwards(

  458.                     self.app_label, editor, project_state, new_state

  459.                 )

  460.         # Reversal.

  461.         with CaptureQueriesContext(connection) as captured_queries:

  462.             with connection.schema_editor(atomic=False) as editor:

  463.                 operation.database_backwards(

  464.                     self.app_label, editor, new_state, project_state

  465.                 )

  466.         self.assertEqual(len(captured_queries), 1)

  467.         self.assertIn("CREATE COLLATION", captured_queries[0]["sql"])

  468.         # Deconstruction.

  469.         name, args, kwargs = operation.deconstruct()

  470.         self.assertEqual(name, "RemoveCollation")

  471.         self.assertEqual(args, [])

  472.         self.assertEqual(kwargs, {"name": "C_test", "locale": "C"})

  473. 

  474. 

  475. @unittest.skipUnless(connection.vendor == "postgresql", "PostgreSQL specific tests.")

  476. @modify_settings(INSTALLED_APPS={"append": "migrations"})

  477. class AddConstraintNotValidTests(OperationTestBase):

  478.     app_label = "test_add_constraint_not_valid"

  479. 

  480.     def test_non_check_constraint_not_supported(self):

  481.         constraint = UniqueConstraint(fields=["pink"], name="pony_pink_uniq")

  482.         msg = "AddConstraintNotValid.constraint must be a check constraint."

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

  484.             AddConstraintNotValid(model_name="pony", constraint=constraint)

  485. 

  486.     def test_add(self):

  487.         table_name = f"{self.app_label}_pony"

  488.         constraint_name = "pony_pink_gte_check"

  489.         constraint = CheckConstraint(check=Q(pink__gte=4), name=constraint_name)

  490.         operation = AddConstraintNotValid("Pony", constraint=constraint)

  491.         project_state, new_state = self.make_test_state(self.app_label, operation)

  492.         self.assertEqual(

  493.             operation.describe(),

  494.             f"Create not valid constraint {constraint_name} on model Pony",

  495.         )

  496.         self.assertEqual(

  497.             operation.migration_name_fragment,

  498.             f"pony_{constraint_name}_not_valid",

  499.         )

  500.         self.assertEqual(

  501.             len(new_state.models[self.app_label, "pony"].options["constraints"]),

  502.             1,

  503.         )

  504.         self.assertConstraintNotExists(table_name, constraint_name)

  505.         Pony = new_state.apps.get_model(self.app_label, "Pony")

  506.         self.assertEqual(len(Pony._meta.constraints), 1)

  507.         Pony.objects.create(pink=2, weight=1.0)

  508.         # Add constraint.

  509.         with connection.schema_editor(atomic=True) as editor:

  510.             operation.database_forwards(

  511.                 self.app_label, editor, project_state, new_state

  512.             )

  513.         msg = f'check constraint "{constraint_name}"'

  514.         with self.assertRaisesMessage(IntegrityError, msg), transaction.atomic():

  515.             Pony.objects.create(pink=3, weight=1.0)

  516.         self.assertConstraintExists(table_name, constraint_name)

  517.         # Reversal.

  518.         with connection.schema_editor(atomic=True) as editor:

  519.             operation.database_backwards(

  520.                 self.app_label, editor, project_state, new_state

  521.             )

  522.         self.assertConstraintNotExists(table_name, constraint_name)

  523.         Pony.objects.create(pink=3, weight=1.0)

  524.         # Deconstruction.

  525.         name, args, kwargs = operation.deconstruct()

  526.         self.assertEqual(name, "AddConstraintNotValid")

  527.         self.assertEqual(args, [])

  528.         self.assertEqual(kwargs, {"model_name": "Pony", "constraint": constraint})

  529. 

  530. 

  531. @unittest.skipUnless(connection.vendor == "postgresql", "PostgreSQL specific tests.")

  532. @modify_settings(INSTALLED_APPS={"append": "migrations"})

  533. class ValidateConstraintTests(OperationTestBase):

  534.     app_label = "test_validate_constraint"

  535. 

  536.     def test_validate(self):

  537.         constraint_name = "pony_pink_gte_check"

  538.         constraint = CheckConstraint(check=Q(pink__gte=4), name=constraint_name)

  539.         operation = AddConstraintNotValid("Pony", constraint=constraint)

  540.         project_state, new_state = self.make_test_state(self.app_label, operation)

  541.         Pony = new_state.apps.get_model(self.app_label, "Pony")

  542.         obj = Pony.objects.create(pink=2, weight=1.0)

  543.         # Add constraint.

  544.         with connection.schema_editor(atomic=True) as editor:

  545.             operation.database_forwards(

  546.                 self.app_label, editor, project_state, new_state

  547.             )

  548.         project_state = new_state

  549.         new_state = new_state.clone()

  550.         operation = ValidateConstraint("Pony", name=constraint_name)

  551.         operation.state_forwards(self.app_label, new_state)

  552.         self.assertEqual(

  553.             operation.describe(),

  554.             f"Validate constraint {constraint_name} on model Pony",

  555.         )

  556.         self.assertEqual(

  557.             operation.migration_name_fragment,

  558.             f"pony_validate_{constraint_name}",

  559.         )

  560.         # Validate constraint.

  561.         with connection.schema_editor(atomic=True) as editor:

  562.             msg = f'check constraint "{constraint_name}"'

  563.             with self.assertRaisesMessage(IntegrityError, msg):

  564.                 operation.database_forwards(

  565.                     self.app_label, editor, project_state, new_state

  566.                 )

  567.         obj.pink = 5

  568.         obj.save()

  569.         with connection.schema_editor(atomic=True) as editor:

  570.             operation.database_forwards(

  571.                 self.app_label, editor, project_state, new_state

  572.             )

  573.         # Reversal is a noop.

  574.         with connection.schema_editor() as editor:

  575.             with self.assertNumQueries(0):

  576.                 operation.database_backwards(

  577.                     self.app_label, editor, new_state, project_state

  578.                 )

  579.         # Deconstruction.

  580.         name, args, kwargs = operation.deconstruct()

  581.         self.assertEqual(name, "ValidateConstraint")

  582.         self.assertEqual(args, [])

  583.         self.assertEqual(kwargs, {"model_name": "Pony", "name": constraint_name})
About Sign Up Leave Feedback Contact