pyembedpg

Provide a platform neutral way to run Postgres in integration tests. More info at: http://www.simulmedia.com/resources/blog/pyembedpg-simple-way-use-postgres-python-integration-tests/

Why

Running tests in parallel on different instances of Postgres
Testing on different versions can be useful (e.g., prod has another version of Postgres installed)
Easier to install a particular version than manually
Dropping databases and users all the time is painful (and cleanup might not be done properly everytime)

Features

It downloads the specified version of Postgres automatically from https://ftp.postgresql.org/pub/source/, build it and cache it in .pyembedpg so subsequent uses won't download it. You can start the Postgres server on any port in a context (using with) that will be shutdown when you exit the context.

How to use it

You can install pyembedpg with pip:

pip install pyembedpg

You can start a new postgres server as follows:

    pg = PyEmbedPg('9.4.0')
    # Start the database and download it and build it if it hasn't been done so already
    with pg.start(15432) as db:
        # do your thing

or:

    pg = PyEmbedPg('9.4.0')
    # Start the database and download it and build it if it hasn't been done so already
    db = pg.start(15432)
    try:
        # do your thing
    finally:
        db.shutdown()

If you don't specify the version (e.g., PyEmbedPg()), it will use the latest version found in .pyembedpg or on the FTP server if .pyembedpg is empty.

If you want to use the current Postgres version installed on the server you can do PyEmbedPg('local').

You can use it as follows:

pg = PyEmbedPg('9.4.0')
with pg.start(15432) as postgres:
    postgres.create_user('scott', 'tiger')
    postgres.create_database('testdb', 'scott')

    # Postgres is initialized, now run some queries
    with psycopg2.connect(database='postgres', user='scott', password='tiger', port=postgres.running_port) as conn:
        with conn.cursor() as cursor:
            cursor.execute('CREATE TABLE employee(name VARCHAR(32), age INT)')
            cursor.execute("INSERT INTO employee VALUES ('John', 32)")
            cursor.execute("INSERT INTO employee VALUES ('Mary', 22)")
            cursor.execute('SELECT * FROM employee ORDER BY age')
            assert cursor.fetchall() == [('Mary', 22), ('John', 32)]

or in unit tests:

import unittest
import psycopg2
from pyembedpg import PyEmbedPg


class TestPyEmbedPgWithContext(unittest.TestCase):
    def setUp(self):
        self.postgres = PyEmbedPg('9.4.0').start(15432)

    def test_simple_run(self):
        self.postgres.create_user('scott', 'tiger')
        self.postgres.create_database('testdb', 'scott')

        # Postgres is initialized, now run some queries
        with psycopg2.connect(database='postgres', user='scott', password='tiger', port=self.postgres.running_port) as conn:
            with conn.cursor() as cursor:
                cursor.execute('CREATE TABLE employee(name VARCHAR(32), age INT)')
                cursor.execute("INSERT INTO employee VALUES ('John', 32)")
                cursor.execute("INSERT INTO employee VALUES ('Mary', 22)")
                cursor.execute('SELECT * FROM employee ORDER BY age')
                assert cursor.fetchall() == [('Mary', 22), ('John', 32)]

    def tearDown(self):
        self.postgres.shutdown()

You can also specify a list of ports to use (e.g., PyEmbedPg('9.4.0').start(range(15432, 15440))). The first available port will be used. This can be useful when you run multiple tests in parallel.

Troubleshooting

Error while building postgres

Check that gcc and readline (e.g., libreadline-dev on Debian, readline on MacOS) are installed.

Alternatives

Alternatively, there is another project testing.postgres that allows to use the current Postgres application installed on the server and initialize a new instance.