In this chapter, you will learn how to write automated tests for existing stored procedures or develop procedures using the concepts of unit tests and Test-Driven Development (TDD). All test cases are stored in the database, so we do not need any external resources (such as files, version control, and command-line utilities) to save tests. Database tests may be grouped into test cases; each test case may have its own environment initialization code. All tests are still executed independently because their effects are automatically rolled back after the execution.

With the help of the project in this chapter, we will set up and run automated unit tests to test whether possible mistakes exist in our PostgreSQL 12 RDS from Amazon Web Services (AWS) of Chapter 2, Setting Up a PostgreSQL RDS for ATM Machines, of our banking ATM machine locations within a typical city.

The following topics will be covered in the chapter:

• Making unit tests with pgTAP
• Making unit tests in a simpler way with PGUnit
• PGUnit – same name but a different approach
• Testing with Python – Tesgres

Technical requirements

This chapter will take developers around 16-20 hours of working to develop four PostgreSQL test frameworks.

Making unit tests with pgTAP

Through the different topics that we will develop here, we will be applying different techniques to carry out tests. We are going to start with one of the most used techniques in production environments: pgTAP, which has several functions for use case tests and also provides the output in TAP. The Test Anything Protocol (TAP) is well known for being suitable for harvesting, analysis, and reporting with a TAP harness.

The official website of pgTAP is https://pgtap.org/.

Setting up pgTAP for PostgreSQL RDS

We'll set up pgTAP using the following RDS:

1. Use pgAdmin to connect to our ATM RDS on AWS and select the ATM database to set up pgTAP.
2. Then, navigate to the top menu bar, go to Tools | Query Tool, and then execute the following SQL statements by pressing the  icon on the toolbar, as in the following figure, Figure 11.1:

CREATE SCHEMA IF NOT EXISTS pgTAP;
CREATE EXTENSION IF NOT EXISTS pgTAP WITH SCHEMA pgTAP;

This is better illustrated here:

3. After that, you should be able to see the new pgTAP schema and the pgTAP stored procedures, as you can see in the preceding figure.

pgTAP test examples

Before anything else, you need a testing plan. This basically declares how many tests your script is going to run to protect against premature failure:

1. The preferred way to do this is to declare a plan by calling the plan () function for tests; if you only have one test, then you will call SELECT plan(1), but if you intend to implement n tests, you will declare SELECT plan(n): 

SET search_path TO pgTAP;

SELECT plan(1);

SELECT is_empty('select "ID" from public."ATM locations" where "ZipCode" NOT IN (select "zip" from public."Zip coordinates") LIMIT 1;', 'Check if there are any Bank ZipCodes not included in Zip coordinates table'); 

SELECT * FROM finish();

2. The preceding SQL code is a test to see whether there are any bank ZIP codes inside the ATM locations table that are not found in the ZIP coordinates table, hence we execute that code from pgAdmin:

Figure 11.2 – pgTAP test 1

3. The output shows that the test fails because the first ATM has ZipCode = 1000 but this value cannot be found in the ZIP coordinates table; the failure is # Looks like you failed 1 test of 1.
4. Now we test the second error, because the ATM database is used to store ATM locations of New York City, hence the ATM locations table must not be empty, so we have the following code for the next test:

SET search_path TO pgTAP;

SELECT plan(1);

SELECT isnt_empty('select distinct "ID" FROM public."ATM locations";', 'Check if the ATM locations inside the ATM database are not empty'); 

SELECT * FROM finish();

5. We now execute the second test with pgAdmin: 

6. Our RDS passes the second test OK, which means our ATM locations table has 654 records and it is not empty.
7. Now, we are going to combine the two preceding tests together into one plan. So, our code will be designed for two tests with SELECT plan(2), as follows:

SET search_path TO pgTAP;

SELECT plan(2);

SELECT is_empty('select "ID" from public."ATM locations" where "ZipCode" NOT IN (select "zip" from public."Zip coordinates") LIMIT 1;', 'Check if there are any Bank ZipCodes not included in Zip coordinates table'); 

SELECT isnt_empty('select distinct "ID" FROM public."ATM locations";', 'Check if the ATM locations inside the ATM database are not empty');

SELECT * FROM finish();

8. Hence, when we execute the two-test plan with pgAdmin, the result will show # Looks like you failed 1 test of 2:

Uninstalling pgTAP for PostgreSQL RDS

In order to uninstall pgTAP, please use pgAdmin to execute the following two SQL statements:

DROP EXTENSION IF EXISTS pgTAP;
DROP SCHEMA IF EXISTS pgTAP;

Up to this point, we have seen one of the most important and updated test tools for PostgreSQL. It is easy to install and is available on different platforms such as Linux or Windows.

One of its greatest strengths lies in that TAP output is easily customizable to use in tools such as Jenkins for continuous integration.

Making unit tests in a simple way with PG_Unit

In the previous section, we have seen a tool that, thanks to its regular updates, has many functions to apply in test cases. Now we will continue with a tool that is unfortunately not so often updated but that, thanks to standardization and its simplicity, does not lose its validity: PG_Unit. We will see that when applying it in pgAdmin, we must make some fixes to the installation script but apart from that, we will be able to demonstrate a test case and thus have another option when carrying out test cases.

The website for PGUnit is here: https://github.com/danblack/pgunit-sql/.

Setting up PGUnit for PostgreSQL RDS

We'll start setting up pgunit for PostgreSQL using the following steps:

1. The SQL script to set up PGUnit is found here: https://github.com/danblack/pgunit-sql/blob/master/sql/reinstall.sql. Please open the preceding link in your browser. You will see the script shown in Figure 11.5:

2. The reinstall.sql installation script calls to execute another install.sql file and in turn, the install.sql script calls to many other scrips, such as assert_array_equals.sp.sql, assert_equals.sp.sql, and assert_false.sp.sql, but we are not using psql as these scripts are intending:

3. Because we are using pgAdmin for the AWS PostgreSQL RDS and we are not using psql command-line scripts, we have to comment out o and echo for command-line psql command usages and then we have to copy thee sub-files' scripts to replace in the reinstall.sql script. Also, the following three statements need commenting out: 

• set QUIET on  
• set ON_ERROR_ROLLBACK on 
• set ON_ERROR_STOP on 

4. We copy and execute the script in pgAdmin as in the following screenshot:

5. The library is based on stored procedures that compare variables between themselves or compare variables with various constants:

• pgunit. assert_equals (_expected anyelement, actual anyelement, custom_message varchar): Compares two elements. If they are not equal, then the #assert_equalsn {custom_message} exception is thrown.

• pgunit. assert_not_equals (_expected anyelement, actual anyelement, custom_message varchar): Compares two elements. If they are equal, an exception is thrown: #assert_not_equalsn {custom_message}.
• pgunit. assert_array_equals (_expected anyelement [ ], actual [] anyelement, custom_message varchar): Compares two arrays. Arrays are considered equal if these arrays have the same elements and the sizes of the arrays are equal. If the arrays are not equal, an exception is thrown with the text #assert_array_equalsn {custom_message}.
• pgunit. assert_true (_value boolean, _custom_message varchar):  Compares _value to True. If they are not equal, a #assert_truen {custom_message} exception is thrown.
• pgunit. assert_false (_value boolean, _custom_message varchar): Compares _value to False. If they are not equal, an exception is thrown: #assert_falsen {custom_message}.
• pgunit. assert_null (_value boolean, _custom_message varchar): Compares _value to NULL. If they are not equal, a #assert_nulln {custom_message} exception is thrown.
• pgunit. assert_not_null (_value boolean, _custom_message varchar): 
  Compares _value to NULL. If they are equal, an exception is thrown: #assert_not_nulln {custom_message}.
• pgunit. fail (_custom_message varchar): Throws an exception with the text #assert_failn {custom_message}.
• pgunit. run_test (_sp varchar): Runs the specified stored procedure inside the test infrastructure. After starting the test procedure, data is rolled back.

PGUnit test examples

We'll check out some pgunit test examples as follows:

1. This is the PGUnit script for the same first test of checking whether there are any bank ATM ZIP codes that are not included in the ZIP coordinate table:

create or replace function pgunit.__test_bank_zipcode() returns void as $$
declare
    id INT;
begin
    select "ATM locations"."ID" from "ATM locations" where "ATM locations"."ZipCode" NOT IN (select "Zip coordinates".zip from "Zip coordinates") LIMIT 1 INTO id;
    perform pgunit.assert_null(id, 'Bank ZipCode is not included in Zip coordinates table'); 
end;
$$ language plpgsql;

2. After creating the first test in pgAdmin, we can execute the following query to run the first test:

select * from pgunit.__test_bank_zipcode();

3. Because the first ATM location has zipcode = 1000, which does not exist in the ZIP coordinates table, an exception is raised with our defined message: Bank ZipCode is not included in Zip coordinates table:

4. This is the PGUnit script for the same second test of checking that the ATM locations should not be empty:

create or replace function pgunit.__test_atm_locations() returns void as $$
declare
    atmtotal INT;
begin
    SELECT count(distinct "ID") INTO atmtotal FROM public."ATM locations";
    PERFORM pgunit.assert_true ((atmtotal IS NOT NULL) AND (atmtotal > 0), 'There are no ATM locations inside the ATM database'); 
end;
$$ language plpgsql;

5. After creating the second test in pgAdmin, we can execute the following query to run the second test:

select * from pgunit.__test_atm_locations();

6. Because we already stored 654 ATM locations, the second test should pass alright with no exceptions:

Uninstalling PGUnit for PostgreSQL RDS

In order to uninstall PGUnit, please use pgAdmin to execute the following SQL statement:

DROP SCHEMA IF EXISTS pgunit CASCADE;
//

As we have seen, PG_Unit is a simple tool for PostgreSQL tests; while it's not full of features for testing, it works in a modest way and we have verified it in the preceding sections.

PGUnit – same name but a different approach

When we decided to apply unit tests in PostgreSQL, we noticed there were two tools with the same name: PGUnit, but when we tried them, we realized that they share the fact of simplifying the tests but, unlike the previous one, the pgunit we are using has more recent updates. In the following steps, we will take care of applying it to our project and evaluating its behavior.

The website for simple pgunit is found here: https://github.com/adrianandrei-ca/pgunit.

Setting up simple pgunit for PostgreSQL RDS

1. The plpgsql code depends on the dblink extension being present in the database you run the tests on. We set up a simple pgunit in a dedicated schema such as pgunit and run these two lines of SQL:

CREATE SCHEMA pgunit;
CREATE EXTENSION DBLINK SCHEMA pgunit;

2. You should run the PGUnit.sql code using pgAdmin: https://github.com/adrianandrei-ca/pgunit/blob/master/PGUnit.sql.

3. We will copy the code from GitHub:

4. A convenient way to install the simple pgunit suite in our dedicated pgunit schema is to temporarily change the search path like this:

SET search_path TO pgunit;

We then revert search_path back to the public schema after the GitHub script.

5. Thereafter, please execute the whole of the script together in pgAdmin.

6. The result of a simple pgunit should be as follows: 

Here is a list of prefixes for all tests:

• "test_case_": It is a unit test procedure.
• "test_precondition_": It is a test precondition function.
• "test_postcondition_": It is a test postcondition function.
• "test_setup_": It is a test setup procedure.
• "test_teardown_": It is a test tear-down procedure.

Simple pgunit test examples

We'll now check out some pgunit test examples as follows:

1. This is the first test script to test whether there are any bank ATM ZIP codes not included in the ZIP coordinates table:

create or replace function pgunit.test_case_bank_zipcode() returns void as $$
declare
    id INT;
begin
    select "ATM locations"."ID" from "ATM locations" where "ATM locations"."ZipCode" NOT IN (select "Zip coordinates".zip from "Zip coordinates") LIMIT 1 INTO id;
    perform pgunit.test_assertNull('Bank ZipCode is not included in Zip coordinates table', id); end;
$$ language plpgsql;

2. After creating the first test by pgAdmin, please execute this query to proceed with the first test:

select * from pgunit.test_case_bank_zipcode();

3. The result is as follows: 

The first bank ATM location has zipcode = 1000 but as this value, 1000, does not exist in the ZIP coordinates table, the Bank ZipCode is not included in Zip coordinates table exception is raised.

4. The second test to make sure that the ATM locations table is not empty is a precondition function:

create or replace function pgunit.test_precondition_atm_locations() returns void as $$
declare
    atmtotal INT;
begin
    SELECT count(distinct "ID") INTO atmtotal FROM public."ATM locations";
    PERFORM pgunit.test_assertTrue('There are no ATM locations inside the ATM database', (atmtotal IS NOT NULL) AND (atmtotal > 0) ); 
end;
$$ language plpgsql;

5. After creating the first test by pgAdmin, please execute this query to proceed with the second test:

select * from pgunit.test_precondition_atm_locations();

6. The result of the second test is shown: 

There are no exceptions raised because our ATM locations table has 654 records and it is not empty.

For general tests of simple pgunit, such as select * from pgunit.test_run_all(), because AWS does not support the dblink_connect_u function on RDS for PostgreSQL, these general tests are not supported in AWS RDS.

Uninstalling simple pgunit for PostgreSQL RDS

We can uninstall simple pgunit using the following steps:

1. PGUnitDrop.sql has the code you can use to remove all PGUnit code from the database: https://github.com/adrianandrei-ca/pgunit/blob/master/PGUnitDrop.sql.

2. We will copy the content from the browser: 

3. Because we have installed all of the simple pgunit library inside our separate pgunit schema, we modify the preceding script to look like the following: 

--
-- Clears the PG Unit functions
--
--
drop function pgunit.test_run_suite(TEXT);
drop function pgunit.test_run_all();
drop function pgunit.test_run_condition(proc_name text);
drop function pgunit.test_build_procname(parts text[], p_from integer, p_to integer);
drop function pgunit.test_get_procname(test_case_name text, expected_name_count integer, result_prefix text);
drop function pgunit.test_terminate(db VARCHAR);
drop function pgunit.test_autonomous(p_statement VARCHAR);
drop function pgunit.test_dblink_connect(text, text);
drop function pgunit.test_dblink_disconnect(text);
drop function pgunit.test_dblink_exec(text, text);
drop function pgunit.test_detect_dblink_schema();
drop function pgunit.test_assertTrue(message VARCHAR, condition BOOLEAN);
drop function pgunit.test_assertTrue(condition BOOLEAN);
drop function pgunit.test_assertNotNull(VARCHAR, ANYELEMENT);
drop function pgunit.test_assertNull(VARCHAR, ANYELEMENT);
drop function pgunit.test_fail(VARCHAR);
drop type pgunit.test_results cascade;

DROP EXTENSION IF EXISTS DBLINK;
DROP SCHEMA IF EXISTS pgunit CASCADE;

We will execute the preceding script in pgAdmin to uninstall the simple pgunit.

As we have seen, using pgunit is pretty straightforward to install but at times uninstalling it is a bit more complicated. Despite this, we were able to run a case test successfully. 

In the next section, we will work with another test approach based on Python. 

Testing with Python – Testgres

Testgres was developed under the influence of the Postgres TAP test feature. As an extra feature, it can manage Postgres clusters: initialize, edit configuration files, start/stop the cluster, and execute queries. In the following steps, we will see how to install it, execute a case test, and properly uninstall it.

The website for Testgres is found here: https://github.com/postgrespro/testgres.

Setting up Testgres for PostgreSQL

Testgres is a Python test tool, hence we will set up Testgres on our Jenkins Ubuntu server 192.168.0.200 that we set up with Vagrant in Chapter 7, PostgreSQL with DevOps for Continuous Delivery, to connect to our RDS on AWS:

1. We open SSH into the Jenkins server. Please launch PowerShell as an administrator:

PS C:Windowssystem32>
PS C:Windowssystem32> cd C:ProjectsVagrantJenkins
PS C:ProjectsVagrantJenkins> vagrant up --provider virtualbox
PS C:ProjectsVagrantJenkins> vagrant ssh

vagrant@devopsubuntu1804:~$

2. Install pip3:

vagrant@devopsubuntu1804:~$ sudo apt install -y python3-pip

3. Please answer <Yes> when you get to the Package configuration screen:

4. You can check the version of pip3: 

vagrant@devopsubuntu1804:~$ pip3 -V
pip 9.0.1 from /usr/lib/python3/dist-packages (python 3.6)

5. Install Testgres using pip3: 

vagrant@devopsubuntu1804:~$ sudo pip3 install testgres 

This is a screenshot of the Testgres installation: 

7. We complete the Testgres installation by setting the PG_BIN and PYTHON_VERSION environment variables and then storing them: 

vagrant@devopsubuntu1804:~$ sudo su
root@devopsubuntu1804:~# export PG_BIN=/usr/lib/postgresql/12/bin 
root@devopsubuntu1804:~# export PYTHON_VERSION=3
root@devopsubuntu1804:~# source ~/.profile

Testgres test examples

We will check out some examples from Testgres:

1. We create the same two tests for our RDS, as follows: 

root@devopsubuntu1804:~# mkdir /usr/local/src/testgres
root@devopsubuntu1804:~# nano /usr/local/src/testgres/atmrds.py
----------------------------------------------------------------
#!/usr/bin/env python
# coding: utf-8

import unittest
import logging
import testgres

logging.basicConfig(filename='/tmp/testgres.log')

class RdsTest(unittest.TestCase):
    def test_bank_zipcode(self):
        #create a node with random name, port, etc
        with testgres.get_new_node() as node:
           # RDS endpoint
           node.host = 'atm.ck5074bwbilj.us-east-1.rds.amazonaws.com'
           node.port = 5432

           # run inidb
           node.init()

           with node.connect('atm', 'dba', 'bookdemo') as conn:
               conn.begin('serializable')

               # execute a query in the AWS RDS
               data = conn.execute('select "ATM locations"."ID" from "ATM locations" where "ATM locations"."ZipCode" NOT IN (select "Zip coordinates".zip from "Zip coordinates") LIMIT 1;')
               self.assertFalse(data, 'Bank ZipCode is not included in Zip coordinates table')

    def test_atm_locations(self):
        #create a node with random name, port, etc
        with testgres.get_new_node() as node:
           # RDS endpoint
           node.host = 'atm.ck5074bwbilj.us-east-1.rds.amazonaws.com'
           node.port = 5432

           # run inidb
           node.init()

           with node.connect('atm', 'dba', 'bookdemo') as conn:
               conn.begin('serializable')

               # execute a query in the AWS RDS
               data = conn.execute('SELECT count(distinct "ID") FROM public."ATM locations";')
               self.assertTrue(data, 'There are no ATM locations inside the ATM database')
               self.assertGreater(data[0][0], 0, 'There are no ATM locations inside the ATM database')

if __name__ == '__main__':
    unittest.main()
----------------------------------------------------------------     

The test_bank_zipcode function is the first test, to check whether there are any bank ATM ZIP codes that not included in the ZIP coordinates table, and the second test_atm_locations function checks that the ATM locations table is not empty.

2. Please set the 755 permission for the test:

root@devopsubuntu1804:~# chmod 755 /usr/local/src/testgres/atmrds.py

3. With the correct permission, we can execute the test for our RDS as a Postgres user:

root@devopsubuntu1804:~# cd /usr/local/src/testgres/
root@devopsubuntu1804:/usr/local/src/testgres# su postgres -c 'python3 atmrds.py' 
.F
======================================================================
FAIL: test_bank_zipcode (__main__.RdsTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "atmrds.py", line 26, in test_bank_zipcode
    self.assertFalse(data, 'Bank ZipCode is not included in Zip coordinates table')
AssertionError: [(1,)] is not false : Bank ZipCode is not included in Zip coordinates table

----------------------------------------------------------------------
Ran 2 tests in 1.941s

FAILED (failures=1)

We can see this is in the screenshot here: 

The two tests return one failure with the exception as Bank ZipCode is not included in Zip coordinates table.

5. Steps 1-4 show how to run tests directly with python3. We can improve our testing by using virtualenv. Please install virtualenv: 

root@devopsubuntu1804:/usr/local/src/testgres# cd /root
root@devopsubuntu1804:~# pip3 install virtualenv     

6. A screenshot of the virtualenv installation is as follows: 

7. Now we can create the Bash script for virtualenv as follows. This script calls to the python3 '/usr/local/src/testgres/atmrds.py' test file:

root@devopsubuntu1804:~# nano /usr/local/src/testgres/rds_tests.sh
----------------------------------------------------------------
#!/usr/bin/env bash

VIRTUALENV="virtualenv --python=/usr/bin/python$PYTHON_VERSION"
PIP="pip$PYTHON_VERSION"

# prepare environment
VENV_PATH=/tmp/testgres_venv
rm -rf $VENV_PATH
$VIRTUALENV $VENV_PATH
export VIRTUAL_ENV_DISABLE_PROMPT=1
source $VENV_PATH/bin/activate

# Install local version of testgres
$PIP install testgres

# run tests (PG_BIN)
/usr/local/src/testgres/atmrds.py
----------------------------------------------------------------

8. Please grant execution permission for the script: 

root@devopsubuntu1804:~# chmod 755 /usr/local/src/testgres/rds_tests.sh

9. We navigate away from the current /root folder and then execute the virtualenv Bash script as a Postgres user:

root@devopsubuntu1804:~# cd /usr/local/src/testgres/

root@devopsubuntu1804:/usr/local/src/testgres# su postgres -c './rds_tests.sh'
created virtual environment CPython3.6.9.final.0-64 in 184ms
  creator CPython3Posix(dest=/tmp/testgres_venv, clear=False, global=False)
  seeder FromAppData(download=False, pip=latest, setuptools=latest, wheel=latest, via=copy, app_data_dir=/var/lib/postgresql/.local/share/virtualenv/seed-app-data/v1.0.1)
  activators BashActivator,CShellActivator,FishActivator,PowerShellActivator,PythonActivator,XonshActivator
Processing /var/lib/postgresql/.cache/pip/wheels/a7/2c/a6/37870923d4e356392e53abab3a242cc67535075480e6c177b0/testgres-1.8.3-py3-none-any.whl
Collecting pg8000
  Using cached pg8000-1.15.3-py3-none-any.whl (24 kB)
Processing /var/lib/postgresql/.cache/pip/wheels/a1/d9/f2/b5620c01e9b3e858c6877b1045fda5b115cf7df6490f883382/psutil-5.7.0-cp36-cp36m-linux_x86_64.whl
Collecting port-for>=0.4
  Using cached port_for-0.4-py2.py3-none-any.whl (21 kB)
Collecting six>=1.9.0
  Using cached six-1.15.0-py2.py3-none-any.whl (10 kB)
Collecting scramp==1.2.0
  Using cached scramp-1.2.0-py3-none-any.whl (6.3 kB)
Installing collected packages: scramp, pg8000, psutil, port-for, six, testgres
Successfully installed pg8000-1.15.3 port-for-0.4 psutil-5.7.0 scramp-1.2.0 six-1.15.0 testgres-1.8.3
.F
======================================================================
FAIL: test_bank_zipcode (__main__.RdsTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/local/src/testgres/atmrds.py", line 26, in test_bank_zipcode
    self.assertFalse(data, 'Bank ZipCode is not included in Zip coordinates table')
AssertionError: [(1,)] is not false : Bank ZipCode is not included in Zip coordinates table

----------------------------------------------------------------------
Ran 2 tests in 2.018s

FAILED (failures=1)
root@devopsubuntu1804:/usr/local/src/testgres# 

10. The result confirms that we execute two tests and the first test has failed with the Bank ZipCode is not included in Zip coordinates table exception message:

Uninstalling Testgres for PostgreSQL RDS

Because we have installed Testgres with the Vagrant user, now we can exit from the root user to perform Testgres uninstallation as follows:

root@devopsubuntu1804:/usr/local/src/testgres# exit
vagrant@devopsubuntu1804:~$ sudo -H pip3 uninstall -y testgres
Uninstalling testgres-1.8.3:
  Successfully uninstalled testgres-1.8.3

As we have seen, Testgres is a serious test framework. It should be taken into consideration when testing in a real production environment. A disadvantage could be that it is hard to install in a Windows environment. 

Summary

In this chapter, with the help of a step-by-step project, we have learned PostgreSQL test skills such as schema validation and xUnit-style testing through pgTAP, automated tests for existing stored procedures or developing procedures with PGUnit and simple pgunit, and PostgreSQL node control with Testgres.

This book is now complete. We can proceed on to the book's Appendix to learn about PostgreSQL using other clouds that are different from AWS.