Generator Queries for creating automated DDL test scripts in postgres

part 1

    -- current method of table testing
    do $$
        declare enabled_count integer;
    begin
        select count(*)
        into enabled_count
        from information_schema.tables
        where table_schema = '<TABLE SCHEMA>'
            and table_name = '<TABLE NAME>'

        assert enabled_count = 1, '<STEP REF>, The <SCHEMA.NAME> does not exist';
    end$$;


    -- function s
    do $$
        declare enabled_count integer;
    begin
        select count(*)
        into enabled_count
        from pg_proc as pp
            join pg_namespace as ns on pp.pronamespace = ns.oid
        where ns.nspname = '<FUNCTION SCHEMA>'
            and pp.oid = '<FUNCTION FINGERPRINT>'::regprocedure;

        assert enabled_count = 1, '1a. <FUNCTION SCHEMA.NAME> does not exist';
    end$$;

    -- assertion helpers
    CREATE OR REPLACE FUNCTION __assert(boolean) RETURNS VOID AS $$
        BEGIN
            IF NOT $1 THEN
                RAISE EXCEPTION 'ASSERTING FAILED';
            END IF;
        END;
    $$ LANGUAGE plpgsql;

    CREATE OR REPLACE FUNCTION __assert(boolean, text) RETURNS VOID AS $$
        BEGIN
            IF NOT $1 THEN
                RAISE EXCEPTION 'ASSERTING FAILED --> %', $2;
            END IF;
        END;
    $$ LANGUAGE plpgsql;
    -- could create third helper to log hints

    -- what current method looks like using helpers
    do $$
        declare
            tmp char;
        begin
        tmp := __assert(
            (select count(*) from pg_catalog.pg_tables pt where pt.schemaname = 'simple-test' and pt.tablename = 'person') = 1,
            'table simple-test.person does not exist'
        );
        end;
    $$ LANGUAGE plpgsql;

part 2

    -- a better way 
    -- table definition 
    CREATE  TABLE tdd_tutorial.person ( 
        id                   bigserial  NOT NULL,
        first_name           varchar(32),
        CONSTRAINT pk_person PRIMARY KEY ( id )
    );

    do $$
        declare
            tmp char;
        begin
        tmp := __assert(
            (
                select
                    count(*)
                from
                    information_schema."columns" c
                where
                    c.table_schema = 'simple-test'
                    and c.table_name = 'person'
                    and c.column_name = 'id'
                    and c.data_type = 'bigint'
                    and c.column_default = 'nextval(''person_person_id_seq''::regclass)'
                    and c.is_nullable = 'NO'
                    and c.ordinal_position = 1
            ) = 1,
            'table simple-test.person column person_id attributes are not correct'
        );
        end;
    $$ LANGUAGE plpgsql;

    do $$
        declare
            tmp char;
        begin
        tmp := __assert(
            (
    select
        count(*)
    from
        information_schema."columns" c
    where
        c.table_schema = 'simple-test'
        and c.table_name = 'person'
        and c.column_name = 'name'
        and c.data_type = 'character varying'
        and c.is_nullable = 'NO'
        and c.column_default is null
        and c.ordinal_position = 2
        and c.character_maximum_length = 32
            ) = 1,
            'table simple-test.person column last_name attributes are not correct'
        );
        end;
    $$ LANGUAGE plpgsql;

    do $$
        declare
            tmp char;
        begin
        tmp := __assert(
            (
                select
                    count(*)
                from
                    information_schema."sequences" s 
                where 
                    s.sequence_schema = 'simple-test'
                    and s.sequence_name = 'person_person_id_seq'
                    and s.data_type = 'bigint'
            ) = 1,
            'sequence simple-test.person_person_id_seq does not exist'
        );
        end;
    $$ LANGUAGE plpgsql;

    -- add a index to table
    CREATE INDEX person_name_idx ON "simple-test".person (name);

    do $$
        declare
            tmp char;
        begin
        tmp := __assert(
            (
                select
                    count(*)
                from
                    pg_class t, pg_class i, pg_index ix, pg_attribute a
                where
                    t.oid = ix.indrelid
                    and i.oid = ix.indexrelid
                    and a.attrelid = t.oid
                    and a.attnum = ANY(ix.indkey)
                    and t.relkind = 'r'
                    and ix.indexrelid = 'person_name_idx'::regclass
                    and a.attname = 'last_name'
                    and a.attnum = 2
            ) = 1,
            'index simple-test.person_last_name_idx attribute last_name is  not correct'
        );
        end;
    $$ LANGUAGE plpgsql;

    -- rollback script
    DROP INDEX if exists "simple-test".person_last_name_idx;
    DROP TABLE if exists "simple-test".person;
    DROP SEQUENCE if exists "simple-test".person_person_id_seq;

part 3

    -- automate table assertion tests
    -- find table with large col count
    select
        c.table_schema,
        c.table_name,
        count(*) as cols
    from
        information_schema."columns" c
    where c.table_schema = 'omop_dev'
    group by c.table_schema, c.table_name
    order by 3 desc;


    select
        '-- test for table ' || c.table_schema || '.' || c.table_name || ' column ' || c.column_name  || E'\n' || 
        'do $$
        declare
            tmp char;
        begin
        tmp := __assert(
            (
                select
                    count(*)
                from
                    information_schema."columns" c
                where
                    c.table_schema = ''' || c.table_schema || '''
                    and c.table_name = ''' || c.table_name  || '''
                    and c.column_name = ''' || c.column_name || '''
                    and c.data_type = ''' || c.data_type  || '''' || 
                    coalesce(E'\n' || '				c.column_default = ''' || replace(c.column_default,'''','''''') || '''' || E'\n',  E'\n') || 
    '				and c.is_nullable = ''' || c.is_nullable || '''
                    and c.ordinal_position = ''' || c.ordinal_position || '''
            ) = 1,
            ''table ' || c.table_schema || '.' || c.table_name || ' column ' || c.column_name || ' attributes are not correct''
        );
        end;
    $$ LANGUAGE plpgsql;
    ' as sttmt
    from
        information_schema."columns" c
    where
        c.table_schema = 'simple-test'
        and c.table_name = 'px_tr_tumor';


    -- next step, create a better function test
    -- next step, create helper function or set of helper functions to facilitate to create deployment artifacts
    ```

Updated on August 7, 2025