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
```