我有以下代码:
# config.py
from yaml import load, SafeLoader
from loguru import logger
""" Populate custom configuration for the
SERVICE
component
"""
APPNAME="acceloop-search-engine"
@logger.catch
def _get_field(config, fieldname):
"""Get specified field from yaml config file
Args:
config (dict): dictionary containing the parsed config yaml
fieldname (str): the name of the field to be searched
Returns:
str: the value corresponding to the given field name
"""
if fieldname not in config:
raise SystemExit(f"No field named {fieldname} is present in yaml configuration file, please add it.")
if not config[fieldname]:
raise SystemExit(f"Field {fieldname} is present in yaml configuration file, but must not be empty string, null or None. Please correctly evaluate it.")
return config[fieldname]
@logger.catch
def _get_loader():
"""Get customizable yaml loader
Returns:
loader: customized yaml loader
"""
loader = SafeLoader
return loader
@logger.catch
def _get_config(config_filename):
"""Get dictionary containing the parsed yaml config file
Returns:
dict: dictionary containing the parsed config yaml
"""
return load(open(config_filename, "rb"), Loader=_get_loader())
class Config(object):
def __new__(cls):
if not hasattr(cls, "instance"):
cls.instance = super(Config, cls).__new__(cls)
cls.instance._init_singleton()
return cls.instance
@logger.catch
def _init_singleton(self):
try:
__config = _get_config("config.yml")
except Exception as error:
raise SystemExit(f"Error while loading yaml config file: {error}")
__app_config = _get_field(__config, "app")
self.__table_db_path = _get_field(__app_config, "table-db-path")
@property
def table_db_path(self):
return self.__table_db_path
此文件声明一个单例配置对象,该对象读取各种字段(为了简洁起见,此处已删除),包括 table-db-path。
# database.py
import sqlite3
from urllib.request import pathname2url
from .config import Config
try:
dburi = "file:{}?mode=r".format(pathname2url(Config().table_db_path))
LOCAL_SESSION = sqlite3.connect(dburi, uri=True)
except sqlite3.OperationalError:
raise SystemExit(f"No db {Config().table_db_path} found. Please create it using the correct service.")
创建与 sqlite3 数据库的连接的简单类。
# test_database.py
from src.database import LOCAL_SESSION
def test_should_return_true_when_read_from_db_is_given():
with LOCAL_SESSION.cursor() as cursor:
cursor.execute("""
SELECT count(*) AS tot
FROM table
""")
assert cursor.fetchone()[0] == 1
简单的测试文件,应检查表中是否只有一条记录。
# conftest.py
import pytest
from unittest.mock import patch
from src.config import Config
@pytest.fixture(scope="session", autouse=True)
def mock_config_table_db_path():
with patch.object(Config, "table_db_path", "./tests/resources/ase_service.db"):
yield
print("CONFTEST Loaded")
最后,我尝试修补conftest.py中的Config().table_db_path类属性。
Conftest 已正确加载其他参数,但此参数奇怪地触发
sqlite3.OperationalError:无法打开数据库文件
当 pytest 尝试发现测试时出错,导致该测试无法运行。奇怪的是,如果我检查 Config().table_db_path 的值,它是未修补的值,而不是修补的值。 有谁知道为什么会发生这种情况?
经过一番挖掘并访问 pytest github后,问题出在database.py文件上。当数据库文件导入到 test_database.py 中时,代码会自动执行,从而逃避猴子修补。我解决了创建提供准备连接的上下文管理器类的问题。