python类RunSQL()的实例源码

def test_sql_create_function(self):
        project_state = self.set_up_test_model()

        operation = migrations.RunSQL(
            sql='CREATE FUNCTION das_func () RETURNS INTEGER AS $$ BEGIN RETURN 1; END $$ LANGUAGE plpgsql',
            reverse_sql='DROP FUNCTION das_func()'
        )
        new_state = project_state.clone()
        operation.state_forwards('tests', new_state)

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

        with connection.schema_editor() as editor:
            operation.database_backwards('tests', editor, project_state, new_state)

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')

def test_run_sql(self):
        project_state = self.set_up_test_model()
        operation = migrations.RunSQL(
            """CREATE TABLE i_love_ponies (id int, special_thing int);
            CREATE INDEX i_love_ponies_special_idx ON i_love_ponies (special_thing);
            INSERT INTO i_love_ponies (id, special_thing) VALUES (1, 42);
            INSERT INTO i_love_ponies (id, special_thing) VALUES (2, 51), (3, 60);
            DELETE FROM i_love_ponies WHERE special_thing = 42;
            UPDATE i_love_ponies SET special_thing = 42 WHERE id = 2;""",
            "DROP TABLE i_love_ponies")
        new_state = project_state.clone()
        operation.state_forwards('tests', new_state)
        self.assertTableNotExists('i_love_ponies')
        with connection.schema_editor() as editor:
            operation.database_forwards('tests', editor, project_state, new_state)

        cursor = connection.cursor()
        cursor.execute('SELECT * FROM a.i_love_ponies')

        self.assertTableExists('i_love_ponies')
        self.assertIndexExists('i_love_ponies', ['special_thing'])

        @all_schemata
        def objects_exist(cursor, **kwargs):
            cursor = connection.cursor()
            cursor.execute('SELECT * FROM i_love_ponies ORDER BY id')
            result = cursor.fetchmany(4)
            self.assertTrue(result, 'No objects found in {schema}'.format(**kwargs))
            expected = [(2, 42), (3, 60)]
            self.assertEqual(
                sorted(expected),
                sorted(result),
                'Mismatch objects found in schema {schema}: expected {0}, saw {1}'
                .format(expected, result, **kwargs)
            )

        with connection.cursor() as cursor:
            objects_exist(cursor)

        with connection.schema_editor() as editor:
            operation.database_backwards('tests', editor, new_state, project_state)
        self.assertTableNotExists('i_love_ponies')