van.pg 2.0

脏数据库

测试数据库创建需要很长时间。通常在决定删除/重新创建测试数据库用例时需要小心。

此外，在PostgreSQL中似乎没有一种健壮的方法来确定数据库是否已提交。

因此，van.pg别无选择，只能将责任放在您身上，在数据库变脏时通知它。如果这没有做好，测试隔离将受到影响。这并不理想，但我们能做的最好的。

一个例外是如果您始终使用transaction包（http://pypi.python.org/pypi/transaction）来管理数据库提交。在这种情况下，您可以在事务提交时请求资源变脏。

与testresources的集成

使用这些用例的典型方式是通过testresources（http://pypi.python.org/pypi/testresources/）

>>> from testresources import ResourcedTestCase
>>> from van.pg import DatabaseManager
>>> import psycopg2

>>> def init_db(db):
...     conn = psycopg2.connect(host=db.host, database=db.database)
...     cur = conn.cursor()
...     cur.execute("CREATE TABLE foo (bar INTEGER);")
...     conn.commit()
...     conn.close()

>>> class MyTest(ResourcedTestCase):
...
...     resources = [('db', DatabaseManager(initialize_sql=init_db))]
...
...     def runTest(self):
...         conn = psycopg2.connect(host=self.db.host, database=self.db.database)
...         cur = conn.cursor()
...         cur.execute("INSERT INTO foo VALUES (1);")
...         conn.commit()
...         cur = conn.cursor()
...         cur.execute("SELECT * FROM foo")
...         self.assertEqual(cur.fetchall(), [(1, )])
...         # NOTE: must close connections or dropping databases fails
...         conn.close()
...         self.db.dirtied() # we changed the DB, so it needs re-loading

实际运行测试

>>> from unittest import TextTestRunner
>>> import sys
>>> runner = TextTestRunner(stream=sys.stdout)
>>> runner.run(MyTest()) # doctest: +ELLIPSIS
.
...
OK
...

使用模板数据库

如果您需要多次重新创建相同的数据库，让PostgreSQL从模板数据库复制数据库可能更快。您可以通过让一个DatabaseManager作为另一个的模板来实现这一点。

>>> template_db = DatabaseManager(initialize_sql=init_db)
>>> class MyTest2(MyTest):
...     resources = [('db', DatabaseManager(template=template_db))]

>>> runner.run(MyTest2()) # doctest: +ELLIPSIS
.
...
OK
...

transaction集成

如果关键字参数 dirty_on_commit 为 True，则数据库管理器将在通过 transaction 模块进行的每次成功提交后将数据库标记为已更改。这意味着任何使数据库变脏的测试无需手动通知。

>>> man = DatabaseManager(dirty_on_commit=True)

如果您使用此功能，则需要自行依赖 transaction (http://pypi.python.org/pypi/transaction) 包。

使用现有数据库

默认情况下，van.pg 在临时目录中创建一个新的 PostgreSQL 集群并启动 PostgreSQL 守护进程。这在大多数情况下都有效，但速度并不快。

如果您已经有一个运行的 PostgreSQL 集群，您可以通过设置环境变量 VAN_PG_HOST 来让 van.pg 使用它。例如，要运行 van.pg 的测试以针对其套接字位于 /tmp/pgcluster 的本地 PostgreSQL 服务器，请执行以下操作：

$ VAN_PG_HOST=/tmp/pgcluster python setup.py test

警告：目标数据库中以 test_db 开头的任何数据库都可能被删除。

关闭连接

当测试完成后，请务必正确关闭所有数据库连接。PostgreSQL 不允许在存在打开连接的情况下删除数据库。这将导致 van.pg 在尝试删除测试数据库时出错。

程序化创建集群

在更低的级别，您也可以程序化地操作自己的 PostgreSQL 集群。

初始化集群

>>> from van.pg import Cluster
>>> cluster = Cluster()
>>> cluster.initdb()

在临时目录中创建数据库

>>> import os
>>> dbdir = cluster.dbdir
>>> 'PG_VERSION' in os.listdir(dbdir)
True

启动它

>>> cluster.start()

创建/测试数据库

>>> dbname = cluster.createdb()

我们可以连接到数据库

>>> import psycopg2
>>> conn = psycopg2.connect(database=dbname, host=cluster.dbdir)
>>> cur = conn.cursor()

调整数据库以确保我们可以执行基本操作

>>> cur.execute("CREATE TABLE x (y int)")
>>> cur.execute("INSERT INTO x VALUES (1)")
>>> conn.commit()
>>> cur.execute("SELECT * from x")
>>> cur.fetchall()[0][0]
1

停止集群守护进程

>>> conn.close()
>>> cluster.stop()

再次启动它

>>> cluster.start()
>>> conn = psycopg2.connect(database=dbname, host=cluster.dbdir)
>>> cur = conn.cursor()
>>> cur.execute("SELECT * from x")
>>> cur.fetchall()[0][0]
1

并清理

>>> conn.close()
>>> cluster.cleanup()
>>> cluster.dbdir is None
True
>>> os.path.exists(dbdir)
False

开发

开发在 GitHub 上进行

http://github.com/jinty/van.pg