如何为Spark读写ADLS挂载路径的函数编写单元测试（Mock ADLS）

核心思路 #

你的read_file函数核心依赖的是spark.read.load方法，而非直接操作ADLS客户端。所以不需要Mock ADLS的BlobServiceClient，直接Mock Spark的DataFrame读取逻辑即可——这更简单直接，也能覆盖函数的核心业务逻辑。

具体实现步骤 #

1. 依赖准备 #

使用pytest作为测试框架，配合unittest.mock做Mock，同时确保环境安装了pyspark。

2. 测试代码示例 #

import pytest
from unittest.mock import patch, MagicMock
from pyspark.sql import SparkSession
from your_module import read_file  # 替换为你的函数所在模块

# 全局创建测试用SparkSession
@pytest.fixture(scope="session")
def spark():
    return SparkSession.builder \
        .master("local[1]") \
        .appName("read_file_unit_test") \
        .getOrCreate()

def test_read_file(spark):
    # 1. 构造模拟返回的DataFrame
    mock_data = [("row1", 100), ("row2", 200)]
    mock_df = spark.createDataFrame(mock_data, ["col_a", "col_b"])
    
    # 2. Mock spark.read.load方法，让它返回我们构造的模拟DataFrame
    with patch.object(spark.read, "load", return_value=mock_df) as mock_load:
        # 3. 调用待测试函数
        result_df = read_file(
            path="abfss://container@storageaccount.dfs.core.windows.net/test/path",
            format="parquet",
            logger=MagicMock()  # Mock logger，避免实际日志输出
        )
        
        # 4. 验证测试结果
        # 检查返回的DataFrame与模拟数据一致
        assert result_df.collect() == mock_df.collect()
        # 检查spark.read.load是否被正确调用，参数符合预期
        mock_load.assert_called_once_with(
            "abfss://container@storageaccount.dfs.core.windows.net/test/path",
            format="parquet"
        )

为什么不需要Mock BlobServiceClient？ #

你的函数并没有直接使用ADLS的Python SDK（比如BlobServiceClient），而是通过Spark封装的API读取ADLS路径。Spark已经处理了和ADLS的交互逻辑，所以我们只需要Mock Spark的读取行为，就能完全隔离外部依赖，完成单元测试。

额外提示 #

• 如果函数包含其他逻辑（比如数据校验、日志打点），可以扩展Mock逻辑，比如验证logger的调用次数和参数。
• 要测试不同文件格式（csv、json等），只需复制测试用例，修改format参数和对应的模拟DataFrame结构即可。

内容的提问来源于stack exchange，提问作者skaur