在一个测试环境中（所以它不是完全一对一的，但应该是类似的），我通过实现一个会话范围的fixture来实现这一点，该fixture构建一个 SparkSession . 比如：

@pytest.fixture(scope="session")
def spark():
    spark = (
        # In place for Spark 3.x to work.
        SparkSession.builder.config(
            "spark.driver.extraJavaOptions",
            "-Dio.netty.tryReflectionSetAccessible=true",
        )
        .config(
            "spark.executor.extraJavaOptions",
            "-Dio.netty.tryReflectionSetAccessible=true",
        )
        .appName("pytest-provider-test")
        .master("local[2]")
        .getOrCreate()
    )
    return spark

然后，对于我需要测试的所有函数（以及覆盖范围），使用 spark 固定装置。
在您的情况下，您可能需要一个特定于测试/覆盖范围的模块调用中的相关启动函数 myfile.py 用一个 SparkSession 对象，并将其传递给您的代码库。对于这些函数和任何子模块，您的覆盖范围仍然是准确的。
然后可以将其作为常规python模块运行。