About
=====
``testing.common.database`` is utilities for testing.* package.
.. image:: https://travis-ci.org/tk0miya/testing.common.database.svg?branch=master
:target: https://travis-ci.org/tk0miya/testing.common.database
.. image:: https://codeclimate.com/github/tk0miya/testing.common.database/badges/gpa.svg
:target: https://codeclimate.com/github/tk0miya/testing.common.database
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 True
def 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 = Postgresql
The 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:
* `498D6B9E <https://pgp.mit.edu/pks/lookup?op=vindex&search=0x102C2C17498D6B9E>`_
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
Raw data
{
"_id": null,
"home_page": "https://github.com/tk0miya/testing.common.database",
"name": "testing.common.database",
"maintainer": "",
"docs_url": null,
"requires_python": "",
"maintainer_email": "",
"keywords": "",
"author": "Takeshi Komiya",
"author_email": "i.tkomiya@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/25/3c/5f7eef6ce8a16314a39f2b905ebd5cd2bfdcbaabafb7fd71dc10c3f32c4d/testing.common.database-2.0.3.tar.gz",
"platform": "",
"description": "About\n=====\n``testing.common.database`` is utilities for testing.* package.\n\n.. image:: https://travis-ci.org/tk0miya/testing.common.database.svg?branch=master\n :target: https://travis-ci.org/tk0miya/testing.common.database\n\n.. image:: https://codeclimate.com/github/tk0miya/testing.common.database/badges/gpa.svg\n :target: https://codeclimate.com/github/tk0miya/testing.common.database\n\n\nInstall\n=======\nUse pip::\n\n $ pip install testing.common.database\n\n\nHelpers\n=======\nclass Database(object):\n\n ``Database`` is a base class for database testing packages.\n To create your database testing class, inherit this class and override methods below.\n\n def initialize(self):\n\n Handler for initialize database object.\n\n def get_data_directory(self):\n\n Path to data directory of your databse.\n\n Example::\n\n def get_data_directory(self):\n return os.path.join(self.base_dir, 'data')\n\n def initialize_database(self):\n\n Handler to initialize your database.\n\n Example::\n\n def initialize_database(self):\n if not os.path.exists(os.path.join(self.base_dir, 'data', 'PG_VERSION')):\n args = ([self.initdb, '-D', os.path.join(self.base_dir, 'data'), '--lc-messages=C'] +\n self.settings['initdb_args'].split())\n\n try:\n p = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)\n output, err = p.communicate()\n if p.returncode != 0:\n raise RuntimeError(\"initdb failed: %r\" % err)\n except OSError as exc:\n raise RuntimeError(\"failed to spawn initdb: %s\" % exc)\n\n def get_server_commandline(self):\n\n Command line to invoke your database server.\n\n Example::\n\n def get_server_commandline(self):\n return (['postgres',\n '-p', str(self.settings['port']),\n '-D', os.path.join(self.base_dir, 'data'),\n '-k', os.path.join(self.base_dir, 'tmp')] +\n self.settings['postgres_args'].split())\n\n def prestart(self):\n\n Handler called before invoking your database server.\n\n def poststart(self):\n\n Hander called after invoking your database server.\n\n def is_server_available(self):\n\n Methods check your database server available.\n The ``Database`` class uses this method to check the server boots up.\n\n Example::\n\n try:\n with closing(pg8000.connect(**self.dsn(database='template1'))):\n pass\n except pg8000.Error:\n return False\n else:\n return True\n\n def is_alive(self):\n\n Methods check the database server is alive.\n\n @property\n def server_pid(self):\n\n Process ID of the database server.\n\n\nclass DatabaseFactory(object):\n\n ``DatabaseFactory`` is a factory class for the database class.\n To create your database factory class, inherit this class and set ``target_class`` variable::\n\n class PostgresqlFactory(DatabaseFactory):\n target_class = Postgresql\n\n The factory class should work like a ``target_class``::\n\n # The factory class generates like a ``target_class``, in this case, generates ``Postgresql`` class\n Postgresql = PostgresqlFactory()\n\n # The generated class works same as ``target_class``\n with Postgresql() as pgsql:\n #\n # do any tests using the database ...\n #\n\n It can bypass parameters to the ``target_class`` on every instantiation::\n\n Postgresql = PostgresqlFactory(copy_data_from='/path/to/database')\n\n with Postgresql() as pgsql:\n #\n # Test with ``copy_data_from`` parameter :-)\n #\n\n Also, it is able to cache the database generated at ``Database.initialize_database()``\n with ``cache_initialized_db`` parameter.\n It avoids running database initialization on every tests::\n\n # Initialize database once\n Postgresql = PostgresqlFactory(cache_initialized_db=True)\n\n with Postgresql() as pgsql:\n # copy cached database for this test.\n\n If you want to fixtures to the database, use ``on_initialized`` parameter::\n\n def handler(pgsql):\n # inserting fixtures\n\n # Initialize database once, and call ``on_initialized`` handler\n Postgresql = PostgresqlFactory(cache_initialized_db=True,\n on_initialized=handler)\n\nclass SkipIfNotInstalledDecorator(object):\n\n Generates decorator that skips the testcase if database command not found.\n To create decorator, inherit this class and set ``name`` variable and override ``search_server()`` method.\n\n Example::\n\n class PostgresqlSkipIfNotInstalledDecorator(SkipIfNotInstalledDecorator):\n name = 'PostgreSQL'\n\n def search_server(self):\n find_program('postgres', ['bin']) # raise exception if not found\n\n\n skipIfNotFound = skipIfNotInstalled = PostgresqlSkipIfNotInstalledDecorator()\n\n @skipIfNotFound\n def test():\n # testcase\n\ndef get_unused_port():\n\n Get free TCP port.\n\ndef get_path_of(name):\n\n Searchs command from search paths. It works like ``which`` command.\n\n\nRequirements\n============\n* Python 2.7, 3.4, 3.5, 3.6\n\nLicense\n=======\nApache License 2.0\n\nRelease signatures\n==================\nReleases are signed with following keys:\n\n* `498D6B9E <https://pgp.mit.edu/pks/lookup?op=vindex&search=0x102C2C17498D6B9E>`_\n\nHistory\n=======\n\n2.0.3 (2017-10-24)\n-------------------\n* Fix a bug:\n\n - Handle exceptions from get_path_of()\n\n2.0.2 (2017-10-08)\n-------------------\n* Fix a bug:\n\n - #18: Fix launch when using cache_initialized_db without init_handler \n\n2.0.1 (2017-07-15)\n-------------------\n* #9: Database always gets instantiated with correct settings\n* #10: Remove explicit path to which\n* #11: Make database server kill-timeout more configurable\n\n2.0.0 (2016-08-20)\n-------------------\n* Use subprocess.Popen() instead of fork & exec\n* Support windows platform (experimental)\n* #4: Add boot_timeout parameter\n* Fix bugs:\n\n - Fix syntax errors for Python3\n - Show error messages if rescue from GC failed (ref: #1)\n\n1.1.0 (2016-02-05)\n-------------------\n* Add Database#server_pid to get pid of the database server\n* Add Database#is_alive() to check server is alive\n* Define BOOT_TIMEOUT as constant\n* Fix AttributeError if any exceptions are raised in bootstrap\n\n1.0.0 (2016-02-01)\n-------------------\n* Initial release\n",
"bugtrack_url": null,
"license": "Apache License 2.0",
"summary": "utilities for testing.* packages",
"version": "2.0.3",
"project_urls": {
"Homepage": "https://github.com/tk0miya/testing.common.database"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "a71aca1c39544ed92fa8ea121ff3bf05bb4838520c498942054235ebc4a83b36",
"md5": "81f09a9fdb422aaf8337580bf99e8ab8",
"sha256": "e3ed492bf480a87f271f74c53b262caf5d85c8bc09989a8f534fa2283ec52492"
},
"downloads": -1,
"filename": "testing.common.database-2.0.3-py2.py3-none-any.whl",
"has_sig": false,
"md5_digest": "81f09a9fdb422aaf8337580bf99e8ab8",
"packagetype": "bdist_wheel",
"python_version": "3.6",
"requires_python": null,
"size": 10500,
"upload_time": "2017-10-23T15:02:40",
"upload_time_iso_8601": "2017-10-23T15:02:40.781630Z",
"url": "https://files.pythonhosted.org/packages/a7/1a/ca1c39544ed92fa8ea121ff3bf05bb4838520c498942054235ebc4a83b36/testing.common.database-2.0.3-py2.py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "253c5f7eef6ce8a16314a39f2b905ebd5cd2bfdcbaabafb7fd71dc10c3f32c4d",
"md5": "55e7b7fdf84adcaa4cf0c61115feb1d3",
"sha256": "965d80b2985315325dc358c3061b174a712f4d4d5bf6a80b58b11f9a1dd86d73"
},
"downloads": -1,
"filename": "testing.common.database-2.0.3.tar.gz",
"has_sig": false,
"md5_digest": "55e7b7fdf84adcaa4cf0c61115feb1d3",
"packagetype": "sdist",
"python_version": "source",
"requires_python": null,
"size": 11535,
"upload_time": "2017-10-23T15:02:37",
"upload_time_iso_8601": "2017-10-23T15:02:37.347692Z",
"url": "https://files.pythonhosted.org/packages/25/3c/5f7eef6ce8a16314a39f2b905ebd5cd2bfdcbaabafb7fd71dc10c3f32c4d/testing.common.database-2.0.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2017-10-23 15:02:37",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "tk0miya",
"github_project": "testing.common.database",
"travis_ci": true,
"coveralls": false,
"github_actions": false,
"tox": true,
"lcname": "testing.common.database"
}
JSON
