@@ -688,19 +688,50 @@ def test_migration_test_flush_exception_requires_explicit_setting_and_test_name(
688688 source = migration .read_text ()
689689 self .assertIn ("BEFORE TRUNCATE" , source )
690690 self .assertIn ("CORE_ALLOW_APPEND_ONLY_TEST_FLUSH" , source )
691+ self .assertIn ('frozenset({"dtc_test"})' , source )
691692 self .assertIn ('database_name.startswith("test_")' , source )
692693 self .assertIn ("truncate_trigger" , source )
693694
694695 migration_module = import_module ("core.migrations.0001_initial" )
695- schema_editor = mock .Mock ()
696- schema_editor .connection .vendor = "postgresql"
697- schema_editor .connection .settings_dict = {"NAME" : "production" }
698- with (
699- override_settings (CORE_ALLOW_APPEND_ONLY_TEST_FLUSH = True ),
700- self .assertRaisesRegex (RuntimeError , "requires a Django test database" ),
696+
697+ def schema_editor_for (database_name : str ):
698+ schema_editor = mock .Mock ()
699+ schema_editor .connection .vendor = "postgresql"
700+ schema_editor .connection .settings_dict = {"NAME" : database_name }
701+ schema_editor .connection .ops .quote_name .side_effect = lambda value : f'"{ value } "'
702+ return schema_editor
703+
704+ for database_name in ("dtc_test" , "test_dtc_test" , "test_dtc_test_1" ):
705+ with self .subTest (database_name = database_name ):
706+ schema_editor = schema_editor_for (database_name )
707+ with override_settings (CORE_ALLOW_APPEND_ONLY_TEST_FLUSH = True ):
708+ migration_module .install_append_only_guards (None , schema_editor )
709+ statements = "\n " .join (
710+ str (call .args [0 ]) for call in schema_editor .execute .call_args_list
711+ )
712+ self .assertNotIn ("BEFORE TRUNCATE" , statements )
713+ self .assertIn ("BEFORE UPDATE OR DELETE" , statements )
714+
715+ for database_name in (
716+ "production" ,
717+ "website_test" ,
718+ "dtc_test_1" ,
719+ "dtc_test_backup" ,
701720 ):
721+ with self .subTest (database_name = database_name ):
722+ schema_editor = schema_editor_for (database_name )
723+ with (
724+ override_settings (CORE_ALLOW_APPEND_ONLY_TEST_FLUSH = True ),
725+ self .assertRaisesRegex (RuntimeError , "requires a Django test database" ),
726+ ):
727+ migration_module .install_append_only_guards (None , schema_editor )
728+ schema_editor .execute .assert_not_called ()
729+
730+ schema_editor = schema_editor_for ("dtc_test" )
731+ with override_settings (CORE_ALLOW_APPEND_ONLY_TEST_FLUSH = False ):
702732 migration_module .install_append_only_guards (None , schema_editor )
703- schema_editor .execute .assert_not_called ()
733+ statements = "\n " .join (str (call .args [0 ]) for call in schema_editor .execute .call_args_list )
734+ self .assertIn ("BEFORE TRUNCATE" , statements )
704735
705736 @skipUnlessDBFeature ("has_select_for_update" )
706737 def test_postgresql_production_truncate_guards_reject_and_reapply (self ) -> None :
0 commit comments