我正在使用python的内置单元测试工具:unittest
class MyClass:
def a():
# some operations
self.spark = SparkSession.builder.config(xxxx).getOrCreate()
如何测试此方法?
由于a()
直接调用SparkSession方法。我该如何测试它?
使用unittest.mock.patch
修补SparkSession
。下面是一个自包含的单文件示例,它尽可能地遵循您的示例代码,并生成一个传递的pytest
结果:
SparkSession = None
class MyClass:
def a(self):
# some operations
self.spark = SparkSession.builder.config('xxxx').getOrCreate()
from unittest.mock import patch
@patch('test.SparkSession')
def test_myclass_a(mock_spark_session):
MyClass().a()
请注意,如果我不修补SparkSession
,那么调用MyClass().a()
将引发AttributeError
,因为SparkSession
在其他方面是None
。
CCD_ 11是在我创建的名为CCD_ 12的文件中定义的,所以我修补了CCD_;您应该在测试中的模块中进行修补,而不是在测试模块中。