Creates a temporary, local PostgreSQL database cluster and server specifically for unittesting, and cleans up after itself.
PGTest(username='postgres', port=None, log_file=None, no_cleanup=False,
copy_cluster=None, base_dir=None, pg_ctl=None, max_connections=5)
Args:
username - str, username for default database superuser
port - int, port to connect on; you must ensure that the port is unused
log_file - str, path to place the log file
no_cleanup - bool, don't clean up dirs after PGTest.close() is called
copy_cluster - str, copies cluster from this path
base_dir - str, path to the base directory to init the cluster
pg_ctl - str, path to the pg_ctl executable to use
max_connections - int, maximum number of connections to the cluster
Attributes:
PGTest.port - int, port number bound by PGTest
PGTest.cluster - str, cluster directory generated by PGTest
PGTest.username - str, username used by PGTest. Default is 'postgres'
PGTest.log_file - str, path to postgres log file
PGTest.pg_ctl - str, path to pg_ctl executable
PGTest.url - str, url for default postgres database on the cluster
PGTest.dsn - dict, dictionary containing dsn key-value pairs for the
default postgres database on the cluster
Methods:
close() - Closes this instance of PGTest, cleans up directories
Usage as an instance
>>> from pgtest import PGTest
>>> import psycopg2
>>> pg = PGTest()
Server started: postgresql://postgres@localhost:47251/postgres
>>> pg.port
47251
>>> pg.cluster
'/tmp/tmpiDtBjs/data'
>>> pg.username
'postgres'
>>> pg.log_file
'/tmp/tmpiDtBjs/pgtest_log.txt'
>>> pg.pg_ctl
u'/usr/lib/postgresql/9.4/bin/pg_ctl'
>>> pg.url
'postgresql://postgres@localhost:47251/postgres'
>>> pg.dsn
{'user': 'postgres', 'host': 'localhost',
'port': 47251, 'database': 'postgres'}
>>> # Connect with other db driver here, e.g. psql, psycopg2,
>>> # sqlalchemy etc
>>> psycopg2.connect(**pg.dsn)
>>> pg.close()
Server stopped
Or use as a context:
>>> from pgtest import PGTest
>>> import psycopg2
>>> with PGTest() as pg:
... # connect to db with psycopg/sqlalchemy etc
... psycopg2.connect(**pg.dsn)
... # Do other database actions here
>>> # When the context exits, the db cluster and service is deleted unless specified
Use with unittest
in the setUp()
and tearDown()
methods:
import unittest
from pgtest.pgtest import PGTest
class TestThirdPartyDrivers(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.pg = pgtest.PGTest()
cls.base_dir = cls.pg._base_dir
@classmethod
def tearDownClass(cls):
cls.pg.close()
def test_something(self):
self.assertTrue(isinstance(self.pg, PGTest)