本文讲解如何通过依赖注入替代隐式模块级 Spark 会话引用，解决测试中 func1 及其内部调用函数无法访问 pytest fixture 初始化的 Spark 会话的问题，提升代码可测性与解耦度。
本文讲解如何通过依赖注入替代隐式模块级 spark 会话引用，解决测试中 `func1` 及其内部调用函数无法访问 pytest fixture 初始化的 spark 会话的问题，提升代码可测性与解耦度。
在 Pytest 中测试依赖 Spark 的函数时，一个常见痛点是：被测模块（如 file_to_be_tested.py）内部直接导入或初始化 Spark 会话，导致测试无法控制该依赖——即使你在 conftest.py 中通过 fixture（如 session()）创建了有效的 SparkSession，func1 及其调用的子函数仍可能尝试使用未初始化的全局/模块级 Spark 实例，引发 AttributeError 或连接失败。
根本原因在于隐式依赖：file_to_be_tested 模块硬编码了 Spark 获取逻辑（例如 from pyspark.sql import SparkSession + SparkSession.builder.getOrCreate()），使它无法感知测试环境提供的 fixture 实例。强行用 unittest.mock.patch 去 patch 模块内的 Spark 引用虽可行，但易出错、可读性差，且难以覆盖嵌套调用链。
✅ 推荐方案：显式依赖注入（Dependency Injection）
将 Spark 会话作为参数或对象状态传入，而非在模块内部自行创建。这不仅让测试变得简单直接，也显著提升了生产代码的可维护性与可复用性。
方式一：函数级参数注入（轻量、快速改造）
修改 file_to_be_tested.py 中的 func1，显式接收 spark 参数：
# file_to_be_tested.py
def func1(df, spark_session):
# ✅ 使用传入的 spark_session，而非自行创建
result_df = df.filter("age > 25")
# 假设内部调用的另一个函数也需 spark
return _process_with_spark(result_df, spark_session)
def _process_with_spark(df, spark_session):
# 同样接收并使用 spark_session
return df.select("name", "age").cache()对应测试文件保持简洁清晰：
立即学习“Python免费学习笔记（深入）”；
# test_file.py
import file_to_be_tested as t
def test_func(spark): # spark 是 conftest.py 中定义的 fixture
df = spark.read.json("test_data.json")
result = t.func1(df, spark) # ✅ 直接传入 fixture 实例
assert result.count() > 0方式二：类封装 + 构造注入（适合复杂场景）
当多个函数共用 Spark 上下文，或需管理生命周期（如自动 stop）时，推荐封装为类：
# file_to_be_tested.py
class SparkProcessor:
def __init__(self, spark_session):
self.spark = spark_session
def func1(self, df):
result_df = df.filter("age > 25")
return self._process_with_spark(result_df)
def _process_with_spark(self, df):
return df.select("name", "age").cache()
# 使用示例（非测试）
# processor = SparkProcessor(SparkSession.builder.getOrCreate())
# processor.func1(df)测试中只需实例化并调用：
def test_func(spark):
df = spark.read.json("test_data.json")
processor = t.SparkProcessor(spark) # ✅ 注入 fixture
result = processor.func1(df)
assert result.count() == 2⚠️ 注意事项与最佳实践
-
避免 patch 全局 Spark 引用：如
patch("pyspark.sql.SparkSession")或patch("file_to_be_tested.spark")易受导入顺序、缓存影响，调试困难； -
保持 fixture 职责单一：
conftest.py中的sessionfixture 应只负责创建/清理 SparkSession，不承担“向其他模块注入”的职责； -
生产代码一致性：确保线上运行时也传入有效 SparkSession（如通过
getOrCreate()创建），而非回退到模块内硬编码逻辑； -
类型提示增强可读性（可选）：
from pyspark.sql import SparkSession def func1(df: DataFrame, spark_session: SparkSession) -> DataFrame: ...
通过依赖注入，你不再需要“mock 函数内部调用的函数”，而是让整个调用链天然接受可控依赖——这才是可测试代码的设计本质。
