test_clone_schema_db_function.py 文件源码

def test_clone_schema_with_trigger(self):
        TRIGGER_FUNCTION = '''CREATE OR REPLACE FUNCTION trigger_this() RETURNS TRIGGER AS $$
                              BEGIN
                                RAISE EXCEPTION 'Trigger fired correctly';
                              END;
                              $$ LANGUAGE plpgsql'''
        TRIGGER_TRIGGER = '''CREATE TRIGGER "test_trigger_this"
                             BEFORE INSERT ON tests_awaremodel
                             FOR EACH STATEMENT
                             EXECUTE PROCEDURE trigger_this()'''
        project_state = ProjectState()
        new_state = project_state.clone()

        trigger_function_op = migrations.RunSQL(
            sql=TRIGGER_FUNCTION,
            reverse_sql='DROP FUNCTION trigger_this()'
        )
        trigger_op = migrations.RunSQL(
            sql=TRIGGER_TRIGGER,
            reverse_sql='DROP TRIGGER "test_trigger_this" BEFORE EACH UPDATE ON tests_awaremodel'
        )

        trigger_function_op.state_forwards('tests', new_state)
        trigger_op.state_forwards('tests', new_state)

        with connection.schema_editor() as editor:
            trigger_function_op.database_forwards('tests', editor, project_state, new_state)
            trigger_op.database_forwards('tests', editor, project_state, new_state)

        Schema.objects.mass_create('a', 'b', 'c')

        for schema in Schema.objects.all():
            schema.activate()
            with transaction.atomic():
                with self.assertRaises(Exception) as exc:
                    self.assertTrue('Trigger fired correctly' == exc.args[0])
                    AwareModel.objects.create(name='FAILCREATE')