Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/firebird/qa/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,4 +40,4 @@

from .plugin import db_factory, Database, user_factory, User, isql_act, python_act, Action, \
temp_file, temp_files, role_factory, Role, envar_factory, Envar, Mapping, mapping_factory, \
ServerKeeper, ExecutionError, QA_GLOBALS, existing_db_factory
ServerKeeper, ExecutionError, QA_GLOBALS, existing_db_factory, ConfigManager
74 changes: 74 additions & 0 deletions src/firebird/qa/plugin.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@
from packaging.version import parse
import time
from datetime import datetime
from dataclasses import dataclass
from threading import Thread, Barrier, Event
from firebird.driver import connect, connect_server, create_database, driver_config, \
NetProtocol, Server, CHARSET_MAP, Connection, Cursor, \
Expand Down Expand Up @@ -2751,3 +2752,76 @@ def temp_files_fixture(tmp_path) -> List[Path]:

return temp_files_fixture

@dataclass
class ConfigManagerBackup:
action: str
config_file: Path
backup_file: Path = None

class ConfigManager:
"""Object to replace specified server configuration file.

Arguments:
tmp_path: Path to directory where backup will be stored.
old_config: Old config file which will be keeped in backup (e.g. databases.conf)

.. important::

Do not create instances of this class directly! Use **only** fixtures created by `store_config`.
"""

def __init__(self, tmp_path: Path):
self.__tmp_path = tmp_path
self.__bak_configs: Dict[str, ConfigManagerBackup] = {}

def _backup(self, config_name: str):
old_config = _vars_['home-dir'] / config_name
if config_name in self.__bak_configs:
return old_config
if old_config.exists():
backup = self.__tmp_path / (config_name.replace('/', '_') + '.bak')
if backup.exists():
backup.unlink()
shutil.copy(str(old_config), str(backup))
self.__bak_configs[config_name] = ConfigManagerBackup('replace', old_config, backup)
else:
self.__bak_configs[config_name] = ConfigManagerBackup('delete', old_config)
return old_config

def replace(self, config_name: str, new_config: Union[Path, str]):
"""
config_name: Relative path to config in server
new_config: Path to new config or content of config
"""
old_config = self._backup(config_name)
if isinstance(new_config, Path):
shutil.copy(str(new_config), str(old_config))
else:
with open(old_config, 'w') as old:
old.write(new_config)

def add(self, config_name: str, new_config: Union[Path, str]):
"""
config_name: Relative path to config in server
new_config: Path to new config or content of config
"""
old_config = self._backup(config_name)
new_content = new_config.read_text() if isinstance(new_config, Path) else new_config
with open(old_config, 'a') as old:
old.write(new_content)

def restore(self, final=False):
for backup in self.__bak_configs.values():
if backup.action == 'replace':
shutil.copy(str(backup.backup_file), str(backup.config_file))
if final:
backup.backup_file.unlink()
elif backup.action == 'delete':
backup.config_file.unlink()
self.__bak_configs = {}

@pytest.fixture
def store_config(db_path) -> ConfigManager:
manager = ConfigManager(db_path)
yield manager
manager.restore(final=True)
8 changes: 2 additions & 6 deletions tests/bugs/gh_8485_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,18 +35,14 @@
"""

@pytest.mark.version('>=3.0.13')
def test_1(act: Action, tmp_file: Path, capsys):
def test_1(act: Action, tmp_file: Path, store_config: ConfigManager, capsys):

store_config.replace('databases.conf', WRONG_DBCONF)
try:
shutil.copy2(act.home_dir/'databases.conf', tmp_file)
with open(act.home_dir/'databases.conf', 'w') as f:
f.write(WRONG_DBCONF)
act.isql(switches = ['-q'], input = f'connect {act.db.dsn};', combine_output = True)
except Error as e:
# Despite crash, no messages were issued here before fix.
print(e)
finally:
shutil.copy2(tmp_file, act.home_dir/'databases.conf')

for line in act.stdout.splitlines():
if (pos := line.lower().find('databases.conf')) > 0:
Expand Down