问题描述
我正在使用 ibis 框架,我想模拟它进行一些单元测试。
我想模拟两个函数 un ibis:
import ibis
ibis.impala.connect
和
ibis.hdfs_connect
这些函数在自定义类中使用:
class Impala(metaclass=SingletonMeta):
"""An Impala wrapper to handle impala connexion et specific actions
This class is a thread safe singleton thus can be used in multi threads
programs
"""
def __init__(self, user: str = os.environ.get('HDFS_WRITER_LOGIN'),
password: str = os.environ.get('HDFS_WRITER_PWD')) -> None:
"""Initiate impala connexion
Parameters
----------
user : str, optional
User name, by default os.environ['HDFS_WRITER_LOGIN']
password : str, optional
User password, by default os.environ['HDFS_WRITER_PWD']
"""
if (type(user) is not str
or type(password) is not str):
raise TypeError("type of user or password is not str")
if (len(user) < 1
or len(password) < 1):
raise ValueError("Username or passwordd too short")
self.hdfs = ibis.hdfs_connect(
host=os.environ['HDFS_IP'], port=os.environ['HDFS_PORT'])
try:
self.impala = ibis.impala \
.connect(host=os.environ['IMPALA_IP'],
port=int(os.environ['IMPALA_PORT']),
hdfs_client=self.hdfs,
user=user, password=password, auth_mechanism='PLAIN')
signal.signal(signal.SIGTERM, self.__close)
atexit.register(self.__close)
except Exception as e:
print(e)
self.impala = None
基于这个文档,我以我的模拟测试文件结束(这个测试是 WIP,这就是它没有做太多的原因):
import pytest
from .impala import Impala
@pytest.mark.hadoopimpala
class TestImpala:
def test_impala_type(self, mocker):
mocker.patch(
'ibis.hdfs_connect',
lambda: True
)
mocker.patch(
'ibis.impala.connect',
lambda: False
)
assert Impala().get() is not None
但是,在该测试文件上运行 pytest 时,出现以下错误:
\anaconda\lib\site-packages\ibis\ init .py'> 没有属性“hdfs_connect”
问题
就我而言,我怎样才能为我的单元测试正确地模拟这两个函数?
编辑 1
看来我的本地机器没有我们集群使用的相同版本的 ibis。
我的本地是1.4.0
,而集群是0.12.0
. 朝这个方向研究。