Python Mockito имитирует возврат определенного вызова функции ⇐ Python

[Anonymous]

Моя тестируемая функция начинается так:

Код: Выделить всё

def ums_pipeline():
spark = get_gemini_spark()
ums_config = get_ums_config()

sm = StateManager(spark, ums_config)

all_ums_signals = spark.read.parquet(ums_config.UMS_INTERMEDIATE_FOLDER)

Я могу имитировать результаты get_gemini_spark() и get_ums_config(), используя:

Код: Выделить всё

when(gemini_spark).get_gemini_spark().thenReturn(self.spark)

и

Код: Выделить всё

when(src.ums_pipelines).get_ums_config().thenReturn(self.ums_config_mock)

Но как издеваться над spark.read.parquet(ums_config.UMS_INTERMEDIATE_FOLDER)
?
i пробовал:

Код: Выделить всё

(when2(self.spark.read.parquet)
.thenReturn(self.spark.createDataFrame([], schema=StructType([StructField("name", StringType(), True)]))))

я тоже пробовал

Код: Выделить всё

(when(self.spark.read).parquet(self.ums_config_mock.UMS_INTERMEDIATE_FOLDER)
.thenReturn(self.spark.createDataFrame([], schema=StructType([StructField("name", StringType(), True)]))))

издеваясь над src.ums_pipelines.spark.parquet, я, конечно, получаю сообщение об ошибке, что в ums_pipelines нет «искры».
Изменить:
это моя тестовая установка — часть настройки контекста искры:

Код: Выделить всё

@pytest.fixture(autouse=True)
def init_test(self, spark_unit_test_fixture, tmp_path, data_dir):
...
self.spark = spark_unit_test_fixture
when(gemini_spark).get_gemini_spark().thenReturn(self.spark)

Я не получаю сообщение об ошибке, но пустой фрейм данных, который я хочу использовать, не используется, а используется по умолчанию. Цель состоит в том, чтобы внедрить пустой фрейм данных для конкретного тестового примера.

Подробнее здесь: https://stackoverflow.com/questions/792 ... ction-call