6
0
mirror of https://github.com/FirebirdSQL/firebird-qa.git synced 2025-01-22 13:33:07 +01:00

pytest plugin for Firebird QA

This commit is contained in:
Pavel Císař 2021-04-26 20:02:48 +02:00
parent 0f3e9cf0b5
commit 6314f3de9a
8 changed files with 744 additions and 0 deletions

7
.gitignore vendored
View File

@ -127,3 +127,10 @@ dmypy.json
# Pyre type checker
.pyre/
# WingIDE
*.wpr
*.wpu
# Sphinx build
docs/_build

43
README.rst Normal file
View File

@ -0,0 +1,43 @@
===========
Firebird QA
===========
This package contains:
- pytest plugin that provides support for testing the Firebird engine. It uses new Python
driver for Firebird (firebird-driver).
- tests for Firebird engine (directory 'tests')
- files needed by tests (directories 'databases', 'files', 'backups')
Requirements: Python 3.8+, Firebird 3+
Usage Guide
-----------
1. Clone the git repository
2. Install the plugin and required dependencies by running next command from repo. directory::
pip install -e .
3. Create / edit `firebird.conf` file. The default file defines `local` server with default
SYSDBA password. You may change it or add more servers.
3. Use pytest to run tests.
The plugin adds nex options to pytests::
Firebird server:
--server=SERVER Server configuration name
--bin-dir=PATH Path to directory with Firebird utilities
--protocol={xnet,inet,inet4,wnet}
Network protocol used for database attachments
--runslow Run slow tests
To run all tests (except slow ones) against local server use next command::
pytest --server local ./tests
Note: If plugin fails to determine the directory with Firebird utilities (isql, gbak etc.),
use `--bin-dir` option to specify it.

183
firebird.conf Normal file
View File

@ -0,0 +1,183 @@
[firebird.driver]
;
; Firebird driver configuration.
; Path to Firebird client library
; Type: str
;fb_client_library = <UNDEFINED>
; BLOB size threshold. Bigger BLOB will be returned as stream BLOBs.
; Type: int
;stream_blob_threshold = 65536
; Registered servers
; Type: list of configuration section names
servers = local
; Registered databases
; Type: list of configuration section names
;databases =
[firebird.db.defaults]
;
; Default database configuration.
; Name of server where database is located
; Type: str
;server = <UNDEFINED>
; Database connection string
; Type: str
;dsn = <UNDEFINED>
; Database file specification or alias
; Type: str
;database = <UNDEFINED>
; Database filename should be passed in UTF8
; Type: bool
;utf8filename = <UNDEFINED>
; Protocol to be used for database
; Type: enum [xnet, inet, inet4, wnet]
;protocol = <UNDEFINED>
; Defaul user name
; Type: str
;user = <UNDEFINED>
; Default user password
; Type: str
;password = <UNDEFINED>
; Use trusted authentication
; Type: bool
;trusted_auth = no
; User role
; Type: str
;role = <UNDEFINED>
; Character set for database connection
; Type: str
;charset = <UNDEFINED>
; SQL Dialect for database connection
; Type: int
;sql_dialect = 3
; Connection timeout
; Type: int
;timeout = <UNDEFINED>
; Do not use linger for database connection
; Type: bool
;no_linger = <UNDEFINED>
; Page cache size override for database connection
; Type: int
;cache_size = <UNDEFINED>
; Dummy packet interval
; Type: int
;dummy_packet_interval = <UNDEFINED>
; Configuration override
; Type: str
;config = <UNDEFINED>
; List of authentication plugins override
; Type: str
;auth_plugin_list = <UNDEFINED>
; Page size to be used for created database.
; Type: int
;page_size = <UNDEFINED>
; Write mode for created database (True = sync, False = async)
; Type: bool
;forced_writes = <UNDEFINED>
; Character set for created database
; Type: str
;db_charset = <UNDEFINED>
; SQL dialect for created database
; Type: int
;db_sql_dialect = <UNDEFINED>
; Page cache size override for created database
; Type: int
;db_cache_size = <UNDEFINED>
; Sweep interval for created database
; Type: int
;sweep_interval = <UNDEFINED>
; Data page space usage for created database (True = reserve space, False = Use all space)
; Type: bool
;reserve_space = <UNDEFINED>
[firebird.server.defaults]
;
; Default server configuration.
; Server host machine specification
; Type: str
;host = <UNDEFINED>
; Port used by Firebird server
; Type: str
;port = <UNDEFINED>
; Defaul user name
; Type: str
;user = <UNDEFINED>
; Default user password
; Type: str
;password = <UNDEFINED>
; Configuration override
; Type: str
;config = <UNDEFINED>
; List of authentication plugins override
; Type: str
;auth_plugin_list = <UNDEFINED>
; Use trusted authentication
; Type: bool
;trusted_auth = no
[local]
;
; Server configuration.
; Server host machine specification
; Type: str
host = localhost
; Port used by Firebird server
; Type: str
;port = <UNDEFINED>
; Defaul user name
; Type: str
user = SYSDBA
; Default user password
; Type: str
password = masterkey
; Configuration override
; Type: str
;config = <UNDEFINED>
; List of authentication plugins override
; Type: str
;auth_plugin_list = <UNDEFINED>
; Use trusted authentication
; Type: bool
;trusted_auth = no

39
firebird/qa/__init__.py Normal file
View File

@ -0,0 +1,39 @@
#coding:utf-8
#
# PROGRAM/MODULE: firebird-qa
# FILE: firebird/qa/__init__.py
# DESCRIPTION: Firebird QA module
# CREATED: 9.4.2021
#
# The contents of this file are subject to the MIT License
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#
# Copyright (c) 2021 Firebird Project (www.firebirdsql.org)
# All Rights Reserved.
#
# Contributor(s): Pavel Císař (original code)
# ______________________________________
"""firebird-qa - Firebird QA module
"""
from .plugin import db_factory, user_factory, isql_act, Database, Action

401
firebird/qa/plugin.py Normal file
View File

@ -0,0 +1,401 @@
#coding:utf-8
#
# PROGRAM/MODULE: firebird-qa
# FILE: firebird/qa/plugin.py
# DESCRIPTION: pytest plugin for Firebird QA
# CREATED: 9.4.2021
#
# The contents of this file are subject to the MIT License
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#
# Copyright (c) 2020 Firebird Project (www.firebirdsql.org)
# All Rights Reserved.
#
# Contributor(s): Pavel Císař (original code)
# ______________________________________
"""firebird-qa - pytest plugin for Firebird QA
"""
from __future__ import annotations
from typing import List, Dict
import os
import re
import shutil
import platform
import difflib
import pytest
from _pytest.fixtures import FixtureRequest
from subprocess import run, CompletedProcess, CalledProcessError
from pathlib import Path
from configparser import ConfigParser, ExtendedInterpolation
from packaging.specifiers import SpecifierSet
from packaging.version import Version, parse
from firebird.driver import connect, connect_server, create_database, driver_config, \
NetProtocol, PageSize, Server
_vars_ = {'server': None,
'bin-dir': None,
'firebird-config': None,
'runslow': False,
}
_platform = platform.system()
def pytest_addoption(parser, pluginmanager):
""
grp = parser.getgroup('firebird', "Firebird server", 'general')
grp.addoption('--server', help="Server configuration name", default='')
grp.addoption('--bin-dir', metavar='PATH', help="Path to directory with Firebird utilities")
grp.addoption('--protocol',
choices=[i.name.lower() for i in NetProtocol],
help="Network protocol used for database attachments")
grp.addoption('--runslow', action='store_true', default=False, help="Run slow tests")
def pytest_report_header(config):
return ["Firebird:",
f" driver configuration: {_vars_['firebird-config']}",
f" server: {_vars_['server']}",
f" protocol: {_vars_['protocol']}",
f" engine: v{_vars_['version']}, {_vars_['arch']}",
f" home: {_vars_['home-dir']}",
f" bin: {_vars_['bin-dir']}",
f" security db: {_vars_['security-db']}",
f" run slow test: {_vars_['runslow']}",
]
def set_tool(tool: str):
path: Path = _vars_['bin-dir'] / tool
if not path.is_file():
path = path.with_suffix('.exe')
if not path.is_file():
pytest.exit(f"Can't find '{tool}' in {_vars_['bin-dir']}")
_vars_[tool] = path
def pytest_configure(config):
# pytest.ini
config.addinivalue_line(
"markers", "version(versions): Firebird version specifications"
)
config.addinivalue_line(
"markers", "platform(platforms): Platform names"
)
config.addinivalue_line(
"markers", "slow: Mark test as slow to run"
)
if config.getoption('help'):
return
config_path: Path = Path.cwd() / 'firebird.conf'
if config_path.is_file():
driver_config.read(str(config_path))
_vars_['firebird-config'] = config_path
driver_config.register_database('pytest')
#
_vars_['runslow'] = config.getoption('runslow')
_vars_['root'] = config.rootpath
path = config.rootpath / 'databases'
_vars_['databases'] = path if path.is_dir() else config.rootpath
path = config.rootpath / 'backups'
_vars_['backups'] = path if path.is_dir() else config.rootpath
_vars_['server'] = config.getoption('server')
_vars_['bin-dir'] = config.getoption('bin_dir')
_vars_['protocol'] = config.getoption('protocol')
_vars_['password'] = config.getoption('password', 'masterkey')
srv_conf = driver_config.get_server(_vars_['server'])
_vars_['host'] = srv_conf.host.value if srv_conf is not None else ''
#
with connect_server(_vars_['server'], user='SYSDBA',
password=_vars_['password']) as srv:
_vars_['version'] = parse(srv.info.version)
_vars_['home-dir'] = Path(srv.info.home_directory)
_vars_['lock-dir'] = Path(srv.info.lock_directory)
_vars_['security-db'] = Path(srv.info.security_database)
_vars_['arch'] = srv.info.architecture
if _vars_['bin-dir'] is None:
path = _vars_['home-dir'] / 'bin'
if path.is_dir():
_vars_['bin-dir'] = path
else:
pytest.exit("Path to binary tools not determined")
else:
_vars_['bin-dir'] = Path(_vars_['bin-dir'])
# tools
for tool in ['isql', 'gbak', 'nbackup', 'gstat', 'gfix', 'gsec']:
set_tool(tool)
def pytest_collection_modifyitems(config, items):
skip_slow = pytest.mark.skip(reason="need --runslow option to run")
skip_platform = pytest.mark.skip(reason=f"test not designed for {_platform}")
skip_version = pytest.mark.skip(reason=f"test not designed for {_vars_['version']}")
for item in items:
if 'slow' in item.keywords and not _vars_['runslow']:
item.add_marker(skip_slow)
platforms = [mark.args for mark in item.iter_markers(name="platform")]
for items in platforms:
if _platform not in items:
item.add_marker(skip_platform)
versions = [mark.args for mark in item.iter_markers(name="version")]
if versions:
spec = SpecifierSet(','.join(list(versions[0])))
if _vars_['version'] not in spec:
item.add_marker(skip_version)
@pytest.fixture(autouse=True)
def firebird_server():
with connect_server(_vars_['server']) as srv:
yield srv
def substitute_macros(text: str, macros: Dict[str, str]):
f_text = text
for (pattern, replacement) in macros.items():
replacement = replacement.replace(os.path.sep,'/')
f_text = f_text.replace(f'$({pattern.upper()})', replacement)
return f_text
class Database:
""
def __init__(self, path: Path, filename: str='test.fdb',
user: str=None, password: str=None):
self.db_path: Path = path / filename
self.dsn: str = None
if _vars_['host']:
self.dsn = f"{_vars_['host']}:{str(self.db_path)}"
else:
self.dsn = str(self.db_path)
self.subs = {'temp_directory': str(path), 'database_location': str(path),
'DATABASE_PATH': str(path), 'DSN': self.dsn,
'files_location': str(_vars_['root'] / 'files'),
'backup_location': str(_vars_['root'] / 'backups'),
'suite_database_location': str(_vars_['root'] / 'databases'),
}
srv_conf = driver_config.get_server(_vars_['server'])
self.user: str = srv_conf.user.value if user is None else user
self.password: str = srv_conf.password.value if password is None else password
def _make_config(self, page_size: int=None, sql_dialect: int=None, charset: str=None) -> None:
db_conf = driver_config.get_database('pytest')
db_conf.clear()
db_conf.server.value = _vars_['server']
db_conf.database.value = str(self.db_path)
db_conf.user.value = self.user
db_conf.password.value = self.password
if sql_dialect is not None:
db_conf.db_sql_dialect.value = sql_dialect
if page_size is not None:
db_conf.page_size.value = page_size
if charset is not None:
db_conf.db_charset.value = charset
if _vars_['protocol'] is not None:
db_conf.protocol.value = NetProtocol._member_map_[_vars_['protocol'].upper()]
def create(self, page_size: int=None, sql_dialect: int=None, charset: str=None) -> None:
#__tracebackhide__ = True
self._make_config(page_size, sql_dialect, charset)
#print(f"Creating db: {self.db_path} [{page_size=}, {sql_dialect=}, {charset=}, user={self.user}, password={self.password}]")
db = create_database('pytest')
db.close()
def restore(self, backup: str) -> None:
#__tracebackhide__ = True
fbk_file: Path = _vars_['backups'] / backup
if not fbk_file.is_file():
raise ValueError(f"Backup file '{fbk_file}' not found")
print(f"Restoring db: {self.db_path} from {fbk_file}")
result = run([_vars_['gbak'], '-r', '-v', '-user', self.user,
'-password', self.password,
str(fbk_file), str(self.dsn)], capture_output=True)
if result.returncode:
print(f"-- stdout {'-' * 20}")
print(result.stdout)
print(f"-- stderr {'-' * 20}")
print(result.stderr)
raise CalledProcessError(result.returncode, result.args, result.stdout, result.stderr)
# Fix permissions
#if platform.system != 'Windows':
#os.chmod(self.db_path, 16895)
return result
def copy(self, filename: str) -> None:
#__tracebackhide__ = True
src_path = _vars_['databases'] / filename
#print(f"Copying db: {self.db_path} from {src_path}")
shutil.copyfile(src_path, self.db_path)
# Fix permissions
if platform.system != 'Windows':
os.chmod(self.db_path, 16895)
def init(self, script: str) -> CompletedProcess:
#__tracebackhide__ = True
#print("Running init script")
result = run([_vars_['isql'], '-ch', 'utf8', '-user', self.user,
'-password', self.password, str(self.dsn)],
input=substitute_macros(script, self.subs),
encoding='utf8', capture_output=True)
if result.returncode:
print(f"-- stdout {'-' * 20}")
print(result.stdout)
print(f"-- stderr {'-' * 20}")
print(result.stderr)
raise CalledProcessError(result.returncode, result.args, result.stdout, result.stderr)
return result
def execute(self, script: str, *, raise_on_fail: bool) -> CompletedProcess:
__tracebackhide__ = True
#print("Running test script")
result = run([_vars_['isql'], '-ch', 'utf8', '-user', self.user,
'-password', self.password, str(self.dsn)],
input=substitute_macros(script, self.subs),
encoding='utf8', capture_output=True)
if result.returncode and raise_on_fail:
print(f"-- stdout {'-' * 20}")
print(result.stdout)
print(f"-- stderr {'-' * 20}")
print(result.stderr)
raise CalledProcessError(result.returncode, result.args, result.stdout, result.stderr)
return result
def drop(self) -> None:
self._make_config()
db = connect('pytest')
#print(f"Removing db: {self.db_path}")
db.drop_database()
@pytest.fixture
def db_path(tmp_path) -> Path:
if platform.system != 'Windows':
os.chmod(tmp_path, 16895)
return tmp_path
class User:
def __init__(self, name: str, password: str, server: Server):
self.name: str = name
self.password: str = password
self.server: Server = server
def create(self) -> None:
if self.server.user.exists(self.name):
self.drop()
self.server.user.add(user_name=self.name, password=self.password)
#print(f"User {self.name} created")
def drop(self) -> None:
self.server.user.delete(self.name)
#print(f"User {self.name} dropped")
def user_factory(*, name: str, password: str) -> None:
@pytest.fixture
def user_fixture(request: FixtureRequest, firebird_server) -> User:
user = User(name, password, firebird_server)
user.create()
yield user
user.drop()
return user_fixture
def db_factory(*, filename: str='test.fdb', init: str=None, from_backup: str=None,
copy_of: str=None, page_size: int=None, sql_dialect: int=None,
charset: str=None, user: str=None, password: str=None):
@pytest.fixture
def database_fixture(request: FixtureRequest, db_path) -> Database:
db = Database(db_path, filename, user, password)
if from_backup is None and copy_of is None:
db.create(page_size, sql_dialect, charset)
elif from_backup is not None:
db.restore(from_backup)
elif copy_of is not None:
db.copy(copy_of)
if init is not None:
db.init(init)
yield db
db.drop()
return database_fixture
class Action:
def __init__(self, db: Database, script: str, substitutions: List[str]):
self.db: Database = db
self.script: str = script
self.return_code: int = 0
self.stdout: str = ''
self._clean_stdout: str = None
self.stderr: str = ''
self._clean_stderr: str = None
self.expected_stdout: str = ''
self._clean_expected_stdout: str = None
self.expected_stderr: str = ''
self._clean_expected_stderr: str = None
self.substitutions: List[str] = [x for x in substitutions]
def make_diff(self, left: str, right: str) -> str:
return '\n'.join(difflib.ndiff(left.splitlines(), right.splitlines()))
def space_strip(self, value: str) -> str:
"""Reduce spaces in value"""
value= re.sub("(?m)^\\s+", "", value)
return re.sub("(?m)\\s+$", "", value)
def string_strip(self, value: str, substitutions: List[str]=[], isql: bool=True,
remove_space: bool=True) -> str:
"""Remove unwanted isql noise strings and apply substitutions defined
in recipe to captured output value.
"""
if not value:
return value
if isql:
for regex in map(re.compile,['(?m)Database:.*\\n?', 'SQL>[ \\t]*\\n?',
'CON>[ \\t]*\\n?', '-->[ \\t]*\\n?']):
value = re.sub(regex, "", value)
for pattern, replacement in substitutions:
value= re.compile(pattern, re.M).sub(replacement, value)
if remove_space:
value = self.space_strip(value)
return value
def execute(self) -> None:
__tracebackhide__ = True
result: CompletedProcess = self.db.execute(self.script,
raise_on_fail=not bool(self.expected_stderr))
self.return_code: int = result.returncode
self.stdout: str = result.stdout
self.stderr: str = result.stderr
@property
def clean_stdout(self) -> str:
if self._clean_stdout is None:
self._clean_stdout = self.string_strip(self.stdout, self.substitutions)
return self._clean_stdout
@property
def clean_stderr(self) -> str:
if self._clean_stderr is None:
self._clean_stderr = self.string_strip(self.stderr, self.substitutions)
return self._clean_stderr
@property
def clean_expected_stdout(self) -> str:
if self._clean_expected_stdout is None:
self._clean_expected_stdout = self.string_strip(self.expected_stdout, self.substitutions)
return self._clean_expected_stdout
@property
def clean_expected_stderr(self) -> str:
if self._clean_expected_stderr is None:
self._clean_expected_stderr = self.string_strip(self.expected_stderr, self.substitutions)
return self._clean_expected_stderr
def isql_act(db_fixture_name: str, script: str, *, substitutions: List[str]=None):
@pytest.fixture
def isql_act_fixture(request: FixtureRequest) -> Action:
db: Database = request.getfixturevalue(db_fixture_name)
result: Action = Action(db, script, substitutions)
return result
return isql_act_fixture

3
pyproject.toml Normal file
View File

@ -0,0 +1,3 @@
[build-system]
requires = ["setuptools >= 53.0.0", "wheel"]
build-backend = "setuptools.build_meta"

58
setup.cfg Normal file
View File

@ -0,0 +1,58 @@
[build_sphinx]
source-dir=docs
all-files=True
[metadata]
name = firebird-qa
version = 0.1.0
description = pytest plugin for Firebird QA
long_description = file: README.rst
long_description_content_type = text/x-rst; charset=UTF-8
author = Pavel Císař
author_email = pcisar@users.sourceforge.net
license = MIT
license_file = LICENSE
url = https://github.com/FirebirdSQL/fbtest
keywords = Firebird RDBMS QA tools
project_urls =
Documentation = https://firebird-qa.rtfd.io
Bug Reports = https://github.com/FirebirdSQL/firebird-qa/issues
Funding = https://www.firebirdsql.org/en/donate/
Source = https://github.com/FirebirdSQL/firebird-qa
classifiers =
Development Status :: 5 - Production/Stable
Intended Audience :: Developers
License :: OSI Approved :: MIT License
Programming Language :: Python :: 3
Programming Language :: Python :: 3.8
Operating System :: POSIX :: Linux
Operating System :: Microsoft :: Windows
Operating System :: MacOS
Topic :: Software Development :: Testing
Topic :: Database
Framework :: Pytest
[options]
zip_safe = True
python_requires = >=3.8, <4
install_requires =
firebird-base>=0.6.0
firebird-driver>=0.8.0
pytest>=6.2.0
packages = find_namespace:
[options.packages.find]
include = firebird.*
[options.entry_points]
pytest11 =
firebird = firebird.qa.plugin
[bdist_wheel]
# This flag says to generate wheels that support both Python 2 and Python
# 3. If your code will not run unchanged on both Python 2 and 3, you will
# need to generate separate wheels for each Python version that you
# support.
universal=0

10
setup.py Normal file
View File

@ -0,0 +1,10 @@
#!/usr/bin/env python
#coding:utf-8
# This file is only a shim to allow editable installs. It's not necessary to build
# and install the package via pip (see pyproject.toml and setup.cfg).
import setuptools
if __name__ == "__main__":
setuptools.setup()