Development Status
- 4 - Beta
Intended Audience
- Developers
License
- OSI Approved :: Apache Software License
Programming Language
- Python
- Python :: 2
- Python :: 2.7
- Python :: 3
- Python :: 3.4
- Python :: 3.5
- Python :: 3.6
Topic
- Database
- Software Development
- Software Development :: Testing
About
testing.common.database is utilities for testing.* package.
Install
Use pip:
$ pip install testing.common.database
Helpers
class Database(object):
Database is a base class for database testing packages. To create your database testing class, inherit this class and override methods below.
def initialize(self):
Handler for initialize database object.
def get_data_directory(self):
Path to data directory of your databse.
Example:
def get_data_directory(self): return os.path.join(self.base_dir, 'data')def initialize_database(self):
Handler to initialize your database.
Example:
def initialize_database(self): if not os.path.exists(os.path.join(self.base_dir, 'data', 'PG_VERSION')): args = ([self.initdb, '-D', os.path.join(self.base_dir, 'data'), '--lc-messages=C'] + self.settings['initdb_args'].split()) try: p = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE) output, err = p.communicate() if p.returncode != 0: raise RuntimeError("initdb failed: %r" % err) except OSError as exc: raise RuntimeError("failed to spawn initdb: %s" % exc)def get_server_commandline(self):
Command line to invoke your database server.
Example:
def get_server_commandline(self): return (['postgres', '-p', str(self.settings['port']), '-D', os.path.join(self.base_dir, 'data'), '-k', os.path.join(self.base_dir, 'tmp')] + self.settings['postgres_args'].split())def prestart(self):
Handler called before invoking your database server.
def poststart(self):
Hander called after invoking your database server.
def is_server_available(self):
Methods check your database server available. The Database class uses this method to check the server boots up.
Example:
try: with closing(pg8000.connect(**self.dsn(database='template1'))): pass except pg8000.Error: return False else: return Truedef is_alive(self):
Methods check the database server is alive.
@property def server_pid(self):
Process ID of the database server.
class DatabaseFactory(object):
DatabaseFactory is a factory class for the database class. To create your database factory class, inherit this class and set target_class variable:
class PostgresqlFactory(DatabaseFactory): target_class = PostgresqlThe factory class should work like a target_class:
# The factory class generates like a ``target_class``, in this case, generates ``Postgresql`` class Postgresql = PostgresqlFactory() # The generated class works same as ``target_class`` with Postgresql() as pgsql: # # do any tests using the database ... #It can bypass parameters to the target_class on every instantiation:
Postgresql = PostgresqlFactory(copy_data_from='/path/to/database') with Postgresql() as pgsql: # # Test with ``copy_data_from`` parameter :-) #Also, it is able to cache the database generated at Database.initialize_database() with cache_initialized_db parameter. It avoids running database initialization on every tests:
# Initialize database once Postgresql = PostgresqlFactory(cache_initialized_db=True) with Postgresql() as pgsql: # copy cached database for this test.If you want to fixtures to the database, use on_initialized parameter:
def handler(pgsql): # inserting fixtures # Initialize database once, and call ``on_initialized`` handler Postgresql = PostgresqlFactory(cache_initialized_db=True, on_initialized=handler)
class SkipIfNotInstalledDecorator(object):
Generates decorator that skips the testcase if database command not found. To create decorator, inherit this class and set name variable and override search_server() method.
Example:
class PostgresqlSkipIfNotInstalledDecorator(SkipIfNotInstalledDecorator): name = 'PostgreSQL' def search_server(self): find_program('postgres', ['bin']) # raise exception if not found skipIfNotFound = skipIfNotInstalled = PostgresqlSkipIfNotInstalledDecorator() @skipIfNotFound def test(): # testcase
def get_unused_port():
Get free TCP port.
def get_path_of(name):
Searchs command from search paths. It works like which command.
Requirements
Python 2.7, 3.4, 3.5, 3.6
License
Apache License 2.0
Release signatures
Releases are signed with following keys:
History
2.0.3 (2017-10-24)
Fix a bug:
Handle exceptions from get_path_of()
2.0.2 (2017-10-08)
Fix a bug:
#18: Fix launch when using cache_initialized_db without init_handler
2.0.1 (2017-07-15)
#9: Database always gets instantiated with correct settings
#10: Remove explicit path to which
#11: Make database server kill-timeout more configurable
2.0.0 (2016-08-20)
Use subprocess.Popen() instead of fork & exec
Support windows platform (experimental)
#4: Add boot_timeout parameter
Fix bugs:
Fix syntax errors for Python3
Show error messages if rescue from GC failed (ref: #1)
1.1.0 (2016-02-05)
Add Database#server_pid to get pid of the database server
Add Database#is_alive() to check server is alive
Define BOOT_TIMEOUT as constant
Fix AttributeError if any exceptions are raised in bootstrap
1.0.0 (2016-02-01)
Initial release