Produits Entr'ouvert » w.c.s.

@postgresql

def test_sql_table_wipe_and_drop():

    test_formdef = FormDef()

    test_formdef.name = 'tests wipe and drop'

    test_formdef.fields = []

    test_formdef.store()

    assert test_formdef.table_name is not None

    data_class = test_formdef.data_class(mode='sql')

    assert data_class.count() == 0

    conn, cur = sql.get_connection_and_cursor()

    assert table_exists(cur, test_formdef.table_name)

    conn.commit()

    cur.close()

    data_class.wipe(drop=True)

    conn, cur = sql.get_connection_and_cursor()

    assert not table_exists(cur, test_formdef.table_name)

    assert not table_exists(cur, test_formdef.table_name + '_evolutions')

    conn.commit()

    cur.close()

    test_formdef.store()

    conn, cur = sql.get_connection_and_cursor()

    assert table_exists(cur, test_formdef.table_name)

    @guard_postgres

    def wipe(cls, drop=False):

        conn, cur = get_connection_and_cursor()

        if drop:

            cur.execute('''DROP TABLE %s_evolutions CASCADE''' % cls._table_name)

            cur.execute('''DROP TABLE %s CASCADE''' % cls._table_name)

        else:

            cur.execute('''DELETE FROM %s_evolutions''' % cls._table_name)

            cur.execute('''DELETE FROM %s''' % cls._table_name)

        conn.commit()

        cur.close()

    @classmethod