Produits Entr'ouvert » w.c.s.

def test_wcs_meta_dates(pub):

    conn, cur = sql.get_connection_and_cursor()

    cur.execute('INSERT INTO wcs_meta(key, value) VALUES (%s, %s)', ('foo', 'bar'))

    cur.execute('SELECT created_at, updated_at FROM wcs_meta WHERE key = %s', ('foo',))

    conn.commit()

    row = cur.fetchone()

    assert row[0] is not None

    assert row[1] is not None

    old_created_at = row[0]

    old_updated_at = row[0]

    cur.execute('UPDATE wcs_meta SET value = %s WHERE key = %s', ('bar', 'foo'))

    cur.execute('SELECT created_at, updated_at FROM wcs_meta WHERE key = %s', ('foo',))

    conn.commit()

    row = cur.fetchone()

    assert row[0] == old_created_at

    assert row[1] == old_updated_at

    cur.execute('UPDATE wcs_meta SET value = %s WHERE key = %s', ('bar-2', 'foo'))

    cur.execute('SELECT created_at, updated_at FROM wcs_meta WHERE key = %s', ('foo',))

    conn.commit()

    row = cur.fetchone()

    assert row[0] == old_created_at

    assert row[1] != old_updated_at

    table_exists = cur.fetchone()[0] > 0

    if not table_exists:

                                    value varchar,

                                    created_at timestamptz DEFAULT NOW(),

                                    updated_at timestamptz DEFAULT NOW())'''

    else:

        cur.execute(

            '''SELECT column_name FROM information_schema.columns

                        WHERE table_schema = 'public'

                          AND table_name = %s''',

            ('wcs_meta',),

        )

        existing_fields = {x[0] for x in cur.fetchall()}

        if 'created_at' not in existing_fields:

            cur.execute('''ALTER TABLE wcs_meta ADD COLUMN created_at timestamptz DEFAULT NOW()''')

        if 'updated_at' not in existing_fields:

            cur.execute('''ALTER TABLE wcs_meta ADD COLUMN updated_at timestamptz DEFAULT NOW()''')

    cur.execute(

        '''

CREATE OR REPLACE FUNCTION meta_table_dates_fn()

RETURNS trigger

LANGUAGE plpgsql

AS $$

BEGIN

  NEW.updated_at = NOW();

  RETURN NEW;

END;

$$;'''

    )

    cur.execute('DROP TRIGGER IF EXISTS meta_table_dates_trg ON wcs_meta;')

    cur.execute(

        '''CREATE TRIGGER meta_table_dates_trg BEFORE UPDATE ON wcs_meta

            FOR EACH ROW

            WHEN (OLD.value IS DISTINCT FROM NEW.value)

            EXECUTE PROCEDURE meta_table_dates_fn();'''

    )

    if not table_exists: