我正在尝试模拟在导入python脚本期间调用的sqlalchemycreate_engine
方法。但我不能正确地这样做。这是我的文件结构:
文件main.py
from database_wrapper import update_database
def do_something():
# do something and
update_database()
database_wrapper.py
from sqlalchemy import create_engine
engine = create_engine("connection_uri")
def update_database() -> bool:
con = engine.connect()
# do something with the connection, and so on...
在我的测试单元中,我做了一些类似的事情:
class TestMicroservice(TestCase):
@patch('database_wrapper.create_engine')
@patch('main.update_database')
def test_1(self, update_database_mock, create_engine_mock):
create_engine_mock.return_value = "Fake engine, babe"
update_database_mock.return_value = True
from main import do_something
do_something()
我也尝试过@patch('main.database_wrapper.create_engine')
、@patch('main.database_wrapper.sqlalchemy.create_engine')
、@patch('sqlalchemy.create_engine')
和许多其他变体,但没有成功。我读过一些类似的帖子,但找不到确切的案例。我宁愿避免更改database_wrapper.py
,因此我不能移动engine = create_engine("connection_uri")
。也因为我很乐意在程序启动时创建引擎,而无需四处传递。
我找到了一个解决方案。感谢Bean Bremen先生让我走上了正确的道路。事实上,engine = create_engine("connection_uri")
已经被延迟初始化了,正如这里所解释的。我唯一需要更改的是传递一个合理的字符串作为连接uri,类似
engine = create_engine("postgresql://user:pass@localhost:5432/db?sslmode=disable")
通过这种方式,引擎将被初始化,但因为我永远不会在update_database()
方法中实际使用它,所以它永远不会出现问题。