我正在使用ibis框架,我想模拟它进行一些单元测试
我想模拟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
基于this documentation,我用这个作为模拟的测试文件结束了测试(这个测试是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'> does not have the attribute 'hdfs_connect'
在我的例子中,如何在单元测试中正确地模拟这两个函数
我的本地机器似乎没有集群使用的相同版本的ibis
我的本地是1.4.0
,而集群是0.12.0
。朝这个方向调查
目前没有回答
相关问题 更多 >
编程相关推荐