Fixed #34149 -- Allowed adding deferrable conditional exclusion constraints on PostgreSQL.

This commit is contained in:
Márton Salomváry 2022-11-10 08:15:04 +01:00 committed by Mariusz Felisiak
parent 0931d5b087
commit d6cbf39a1b
2 changed files with 33 additions and 12 deletions

View File

@ -51,8 +51,6 @@ class ExclusionConstraint(BaseConstraint):
raise ValueError("The expressions must be a list of 2-tuples.") raise ValueError("The expressions must be a list of 2-tuples.")
if not isinstance(condition, (type(None), Q)): if not isinstance(condition, (type(None), Q)):
raise ValueError("ExclusionConstraint.condition must be a Q instance.") raise ValueError("ExclusionConstraint.condition must be a Q instance.")
if condition and deferrable:
raise ValueError("ExclusionConstraint with conditions cannot be deferred.")
if not isinstance(deferrable, (type(None), Deferrable)): if not isinstance(deferrable, (type(None), Deferrable)):
raise ValueError( raise ValueError(
"ExclusionConstraint.deferrable must be a Deferrable instance." "ExclusionConstraint.deferrable must be a Deferrable instance."

View File

@ -312,16 +312,6 @@ class ExclusionConstraintTests(PostgreSQLTestCase):
deferrable="invalid", deferrable="invalid",
) )
def test_deferrable_with_condition(self):
msg = "ExclusionConstraint with conditions cannot be deferred."
with self.assertRaisesMessage(ValueError, msg):
ExclusionConstraint(
name="exclude_invalid_condition",
expressions=[(F("datespan"), RangeOperators.OVERLAPS)],
condition=Q(cancelled=False),
deferrable=Deferrable.DEFERRED,
)
def test_invalid_include_type(self): def test_invalid_include_type(self):
msg = "ExclusionConstraint.include must be a list or tuple." msg = "ExclusionConstraint.include must be a list or tuple."
with self.assertRaisesMessage(ValueError, msg): with self.assertRaisesMessage(ValueError, msg):
@ -912,6 +902,39 @@ class ExclusionConstraintTests(PostgreSQLTestCase):
RangesModel.objects.create(ints=(10, 19)) RangesModel.objects.create(ints=(10, 19))
RangesModel.objects.create(ints=(51, 60)) RangesModel.objects.create(ints=(51, 60))
def test_range_adjacent_initially_deferred_with_condition(self):
constraint_name = "ints_adjacent_deferred_with_condition"
self.assertNotIn(
constraint_name, self.get_constraints(RangesModel._meta.db_table)
)
constraint = ExclusionConstraint(
name=constraint_name,
expressions=[("ints", RangeOperators.ADJACENT_TO)],
condition=Q(ints__lt=(100, 200)),
deferrable=Deferrable.DEFERRED,
)
with connection.schema_editor() as editor:
editor.add_constraint(RangesModel, constraint)
self.assertIn(constraint_name, self.get_constraints(RangesModel._meta.db_table))
RangesModel.objects.create(ints=(20, 50))
adjacent_range = RangesModel.objects.create(ints=(10, 20))
# Constraint behavior can be changed with SET CONSTRAINTS.
with self.assertRaises(IntegrityError):
with transaction.atomic(), connection.cursor() as cursor:
quoted_name = connection.ops.quote_name(constraint_name)
cursor.execute(f"SET CONSTRAINTS {quoted_name} IMMEDIATE")
# Remove adjacent range before the end of transaction.
adjacent_range.delete()
RangesModel.objects.create(ints=(10, 19))
RangesModel.objects.create(ints=(51, 60))
# Add adjacent range that doesn't match the condition.
RangesModel.objects.create(ints=(200, 500))
adjacent_range = RangesModel.objects.create(ints=(100, 200))
# Constraint behavior can be changed with SET CONSTRAINTS.
with transaction.atomic(), connection.cursor() as cursor:
quoted_name = connection.ops.quote_name(constraint_name)
cursor.execute(f"SET CONSTRAINTS {quoted_name} IMMEDIATE")
def test_range_adjacent_gist_include(self): def test_range_adjacent_gist_include(self):
constraint_name = "ints_adjacent_gist_include" constraint_name = "ints_adjacent_gist_include"
self.assertNotIn( self.assertNotIn(