Python 没有为 class 分配模拟

Question

我有我的pythonclass

class FetchData:

    def __init__(self, spark):
        self.spark = spark

    def filter_data_from_s3(self, table_name, s3_location, sql_query, table_schema, partition_column):

        # DO SOMETHING

调用此 class 的另一个文件是 FilterData.py

from accessor.FetchData import FetchData
from pyspark.sql import SparkSession


import sys


def main(args):
    spark = SparkSession \
        .builder \
        .appName("ROAD") \
        .config(conf=sparkConf) \
        .getOrCreate()
    try:
        args_map = DataUtils.parse_args(args)
        logger.info("Parsed Argument Map for Filtering Data : {}".format(args_map))
        f = FetchData(spark)

        for table in table_list:
            table_name = table.get("table_name")
            s3_location = table.get("s3_location")
            sql_query = DataUtils.un_parametrize(table.get("sql_query"), args_map)
            table_schema = table.get("table_schema", args)
            f.filter_data_from_s3(table_name=table_name,
                                  s3_location=s3_location,
                                  sql_query=sql_query,
                                  table_schema=table_schema,
                                  partition_column=args_map.get('partition_column'))
    finally:
        spark.stop()


if __name__ == "__main__":
    main(sys.argv)

我写了一个测试用例来测试filterData.py主要功能我想模拟 filter_data_from_s3 电话。

这是我写的测试用例 -

class TestFilterData(unittest.TestCase):

    @patch('accessor.FetchData')
        def test_main_call_times_for_na(self, fetch_data_mock):
    print(fetch_data_mock)
    spark_session = get_spark_session()
    CUSTOMER_ACCESS_SCHEMA = StructType(
        [StructField('enc_customer_id', StringType(), False),
         StructField('marketplace_id', IntegerType(), False)]
    )
    df = spark_session.createDataFrame([
        ("customerid1", 1),
        ("customerid1", 2),
        ("customerid2", 2)
    ], CUSTOMER_ACCESS_SCHEMA)

    fetch_data_mock.filter_data_from_s3.return_value = df
    test_args = ["", "--run_date=2018-09-01", "--num_days=730",
                 "--region=NA", "--marketplace_id=1", "--region_id=1", "--num_asins=1000"]
    FilterData.main(test_args)


if __name__ == '__main__':
    unittest.main()

当我调用我的测试用例时，它实际上调用过滤数据但失败了。相反，我希望调用我的模拟方法。我是 python 的新手，现在面临这个问题大约一天了。谁能告诉我我在这里做错了什么。

Answer 1

如果您从 filterData.py 测试 main()，那么您应该在 filterData.py 中模拟 FetchData。也就是说，而不是 @patch('accessor.FetchData') 它应该是 @patch('filterData.FetchData')

始终在使用它们的地方模拟对象，而不是它们的来源。

你遇到的问题是，即使你正在修补符号 accessor.FetchData 以指向不同的对象（模拟），在你的测试中应用该指令时，模块在测试中，filterData 已经将 reference 导入到其命名空间中的原始对象。现在您已将 origin 模块的 FetchData 符号更改为其他内容并不重要，它不会影响 filterData 的命名空间。如果您希望 filterData 丢失原始引用并指向您的模拟，您必须将其自己的 FetchData 符号指向该模拟。

编辑

根据您的代码使用的方式，FetchData这就是我为您正在寻找的内省类型设置模拟的方式。

import mock

@mock.patch('filterData.FetchData')
def test_main_call_times_for_na(self, fetch_data_mock):
    spark_session = get_spark_session()
    CUSTOMER_ACCESS_SCHEMA = StructType(
        [StructField('enc_customer_id', StringType(), False),
         StructField('marketplace_id', IntegerType(), False)]
    )
    df = spark_session.createDataFrame([
        ("customerid1", 1),
        ("customerid1", 2),
        ("customerid2", 2)
    ], CUSTOMER_ACCESS_SCHEMA)
    fake_f = mock.Mock()
    fake_f.filter_data_from_s3.return_value = df
    # ensuring that I also control the instance returned by filterData.FetchData 
    fetch_data_mock.return_value = fake_f

    test_args = ["", "--run_date=2018-09-01", "--num_days=730",
                 "--region=NA", "--marketplace_id=1", "--region_id=1", "--num_asins=1000"]
    FilterData.main(test_args)

    assert fake_f.filter_data_from_s3.call_count==11

Python 没有为 class 分配模拟

Python Mock not getting assigned for class

python

python-unittest

pyspark