testing.common.database 2.0.3

关于

testing.common.database 是测试.*软件包的实用工具。

安装

使用pip

$ pip install testing.common.database

辅助工具

class Database(object)

Database 是数据库测试软件包的基类。要创建自己的数据库测试类，继承此类并覆盖以下方法。

def initialize(self)

初始化数据库对象的处理器。

def get_data_directory(self)

数据库数据目录的路径。

示例

def get_data_directory(self):
    return os.path.join(self.base_dir, 'data')

def initialize_database(self)

初始化数据库的处理器。

示例

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)

调用数据库服务器的命令行。

示例

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)

在调用数据库服务器之前调用的处理器。

def poststart(self)

在调用数据库服务器之后调用的处理器。

def is_server_available(self)

方法检查数据库服务器是否可用。 Database 类使用此方法来检查服务器是否启动。

示例

try:
    with closing(pg8000.connect(**self.dsn(database='template1'))):
        pass
except pg8000.Error:
    return False
else:
    return True

def is_alive(self)

方法检查数据库服务器是否存活。

@property def server_pid(self)

数据库服务器的进程ID。

class DatabaseFactory(object)

DatabaseFactory 是数据库类的工厂类。要创建自己的数据库工厂类，继承此类并设置 target_class 变量

class PostgresqlFactory(DatabaseFactory):
    target_class = Postgresql

工厂类应该像 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 ...
    #

它可以在每次实例化时绕过传递给 target_class 的参数

Postgresql = PostgresqlFactory(copy_data_from='/path/to/database')

with Postgresql() as pgsql:
    #
    # Test with ``copy_data_from`` parameter :-)
    #

此外，它还可以使用 cache_initialized_db 参数缓存 Database.initialize_database() 生成的数据库。这避免了每次测试都运行数据库初始化。

# Initialize database once
Postgresql = PostgresqlFactory(cache_initialized_db=True)

with Postgresql() as pgsql:
    # copy cached database for this test.

如果要将固定数据添加到数据库中，请使用 on_initialized 参数

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)

生成装饰器，当数据库命令找不到时跳过测试用例。要创建装饰器，继承此类并设置 name 变量，并重写 search_server() 方法。

示例

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()

获取空闲的 TCP 端口。

def get_path_of(name)

在搜索路径中搜索命令。它的工作方式类似于 which 命令。

需求

• Python 2.7, 3.4, 3.5, 3.6

许可证

Apache License 2.0

发布签名

发布使用以下密钥签名

• 498D6B9E

历史记录