使用pytest,如何正确模拟外部模块的功能?

2024-06-11 05:54:58 发布

您现在位置:Python中文网/ 问答频道 /正文

问题描述

我正在使用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'

问题:

在我的例子中,如何在单元测试中正确地模拟这两个函数

编辑1

我的本地机器似乎没有集群使用的相同版本的ibis

我的本地是1.4.0,而集群是0.12.0。朝这个方向调查


Tags: selfisosconnectenvironnotpasswordhdfs