本文是我将在此处和我的个人博客上发布的有关数据处理应用程序测试的一系列文本中的第一篇。
当我从软件工程师转变为数据工程师时，我开始与数据领域没有软件工程背景的人进行对话。在这些对话中，反复出现一个问题：如何编写测试？
事实上，对于那些不习惯编写测试的人来说，编写测试似乎是一项复杂的任务，因为它需要改变编写代码的方式。事实上，这并不神秘，而是一个练习和重复的问题。我在本文中的主要目标是指导刚刚起步的您，展示如何为处理数据的应用程序创建测试，确保代码的质量和可靠性。
本文是我将在接下来的几周内推出的系列文章的一部分，我将在其中分享如何在针对数据工程的代码中编写自动化测试。在今天的文章中，我想探讨一些有关模拟的内容。在一些代码场景中，数据管道将进行连接、API 调用、与云服务集成等，这可能会对我们如何测试该应用程序造成一些混乱。今天我们将探索一些有趣的库来编写测试，重点关注模拟的使用。
毕竟，什么是 Mock？
模拟是测试中使用的模拟对象，用于模仿非测试重点的外部依赖项或组件的行为。它们允许您隔离正在测试的代码单元，确保测试更加可控和可预测。使用模拟是单元测试和集成测试中的常见做法。
我们应该在以下情况下使用模拟：
- 依赖关系与测试无关；
- 依赖项不可用；
- 我们想要测试特殊行为、错误模拟或特定响应。
在数据管道中，Mocking 允许您创建外部组件的表示——例如数据库、消息服务或 API——而不依赖于它们的真实基础设施。这在集成了多种技术的数据处理环境中特别有用，例如用于分布式处理的 PySpark、用于消息传递的 Kafka，以及 AWS 和 GCP 等云服务。
在我们拥有数据管道的这些场景中，模拟有助于执行隔离和快速的测试，从而最大限度地降低成本和执行时间。它允许准确验证管道的每个部分，不会因真实连接或外部基础设施导致间歇性故障，并确信每个集成都按预期工作。
在每种编程语言中，我们都可以找到已经提供了要实现的 Mock 功能的内部模块。在 Python 中，原生的 unittest.mock 库是创建模拟的主要工具，使您可以轻松且控制地模拟对象和函数。在Go中，Mocking过程通常由外部包支持，例如mockery，因为该语言没有原生的Mock库； mockery 对于从接口生成模拟特别有用，这是 Go 的本机功能，Mockito 是一个流行且强大的用于创建模拟的库，与 JUnit 集成以促进健壮的单元测试。这些库为测试隔离组件提供了重要的基础，特别是在数据管道和分布式系统中，外部数据源和 API 的模拟至关重要。
实施模拟
让我们从一个如何使用 Mock 的基本示例开始。假设我们有一个进行 API 调用的函数，我们需要为此函数编写单元测试：
def get_data_from_api(url):
import requests
response = requests.get(url)
if response.status_code == 200:
return response.json()
else:
return None
登录后复制
登录后复制
登录后复制
为了正确处理测试场景，我们首先需要了解应该覆盖哪些情况。当我们的函数进行 REST 调用时，测试必须至少考虑两种主要场景：一种是请求成功，另一种是响应不符合预期。我们可以使用真实的 URL 运行代码来观察行为，但这种方法有缺点，因为我们无法控制不同类型的响应，此外，测试还容易受到 URL 响应变化或其最终不可用的影响。 。为了避免这些不一致，我们将使用 Mocks。
from unittest import mock
@mock.patch('requests.get')
def test_get_data_from_api_success(mock_get):
# Configura o mock para retornar uma resposta simulada
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {"key": "value"}
# Chama a função com o mock ativo
result = get_data_from_api("http://fakeurl.com")
# Verifica se o mock foi chamado corretamente e o resultado é o esperado
mock_get.assert_called_once_with("http://fakeurl.com")
self.assertEqual(result, {"key": "value"})
登录后复制
登录后复制
登录后复制
使用Python unittest库中的@mock.patch装饰，我们可以用mock替换requests.get调用，mock是一个“假对象”，它在测试上下文中模拟get函数的行为，消除了外部依赖
通过定义模拟的 return_value 值，我们可以准确指定在我们正在测试的函数中调用时我们期望对象返回的内容。重要的是 return_value 结构与我们要替换的真实对象相同。例如，来自 requests 模块的响应对象具有 status_code 等属性和 json() 等方法。因此，为了模拟 requests.get 函数的响应，我们可以直接在模拟中将期望值分配给这些属性和方法。
def get_data_from_api(url):
import requests
response = requests.get(url)
if response.status_code == 200:
return response.json()
else:
return None
登录后复制
登录后复制
登录后复制
在这个具体案例中，重点是模拟请求响应，即在不依赖外部 URL 且不影响我们的测试环境的情况下，以不同的预期结果来测试函数的行为。
from unittest import mock
@mock.patch('requests.get')
def test_get_data_from_api_success(mock_get):
# Configura o mock para retornar uma resposta simulada
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {"key": "value"}
# Chama a função com o mock ativo
result = get_data_from_api("http://fakeurl.com")
# Verifica se o mock foi chamado corretamente e o resultado é o esperado
mock_get.assert_called_once_with("http://fakeurl.com")
self.assertEqual(result, {"key": "value"})
登录后复制
登录后复制
登录后复制
通过在测试中模拟 API 错误响应，我们可以超越基础知识，根据不同类型的 HTTP 状态代码（例如 404、401、500 和 503）检查应用程序行为。这提供了更广泛的覆盖范围，并确保应用程序正确处理对于每种类型的故障，我了解调用中的这些变化如何影响我们的应用程序/数据处理。在 POST 方法调用中，我们可以添加额外的验证层，不仅检查调用的 status_code 和基本功能，还检查发送和接收响应的 schema，确保返回的数据遵循预期的格式。这种更详细的测试方法可以确保应用程序准备好处理各种错误场景，并且接收到的数据始终与设计一致，从而有助于防止未来出现问题。
使用 PySpark 实现模拟
现在我们已经看到了在纯 Python 代码中使用 Mocks 的简单案例，让我们将案例扩展到使用 Pyspark 的代码片段。
要测试 PySpark 功能，尤其是 Filter、groupBy 和 join 等 DataFrame 操作，使用模拟是一种有效的方法，无需运行真正的 Spark，从而减少了测试时间并简化了开发环境。 Python 的unittest.mock 库允许您模拟这些方法的行为，从而可以在不依赖Spark 基础设施的情况下验证代码流程和逻辑。
让我们看看，给定以下函数，我们有一个转换，可以对 Spark 中的数据帧执行筛选、groupBy 和连接操作。
def get_data_from_api(url):
import requests
response = requests.get(url)
if response.status_code == 200:
return response.json()
else:
return None
登录后复制
登录后复制
登录后复制
要运行 PySpark 测试，我们需要在本地完成 Spark 配置。此配置在 setUpClass 方法中完成，该方法创建将在该类的所有测试中使用的 Spark 实例。这使我们能够独立运行 PySpark，从而可以在不依赖完整集群的情况下执行真正的转换操作。测试完成后，tearDownClass方法负责终止Spark会话，确保所有资源都得到正确释放，测试环境干净。
from unittest import mock
@mock.patch('requests.get')
def test_get_data_from_api_success(mock_get):
# Configura o mock para retornar uma resposta simulada
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {"key": "value"}
# Chama a função com o mock ativo
result = get_data_from_api("http://fakeurl.com")
# Verifica se o mock foi chamado corretamente e o resultado é o esperado
mock_get.assert_called_once_with("http://fakeurl.com")
self.assertEqual(result, {"key": "value"})
登录后复制
登录后复制
登录后复制
在 test_transform_data 测试中，我们首先为 df 和 df_other 创建示例 DataFrame，其中包含将在转换中使用的数据。然后，我们在不应用模拟的情况下执行transform_data函数，允许过滤器、groupBy和连接操作实际发生并产生一个新的DataFrame。执行后，我们使用collect()方法从生成的DataFrame中提取数据，这使我们能够将此数据与预期值进行比较，从而验证以真实准确的方式进行的转换。
但是我们也可能有想要测试这些 pyspark 函数之一的结果的场景。有必要模拟代码的另一部分，该部分可能代表执行时的瓶颈，但不会对我们的流程带来风险。因此，我们可以使用模拟函数/模块的技术，正如我们在前面使用请求的示例中看到的那样。
response.status_code = mock_get.return_value.status_code
response.json() = mock_get.return_value.json.return_value
登录后复制
对特定操作的Mock测试是在test_transform_data_with_mocked_join方法中进行的，我们专门针对filter方法应用了mock。该模拟用模拟的 DataFrame 替换了 join 操作的结果，允许以前的操作（例如 groupBy 和 join）以真实的方式执行。然后测试将生成的 DataFrame 与预期值进行比较，确保正确使用连接模拟，而不会干扰执行的其他转换。
这种混合方法带来了几个优点。通过确保维护 join 和 groupBy 等实际 PySpark 操作，我们可以验证转换逻辑，而不会失去用模拟替换过滤器等特定操作的灵活性。这样可以实现更稳健、更快速的测试，无需完整的 Spark 集群，从而使持续的代码开发和验证变得更加容易。
需要强调的是，应谨慎使用此策略，并且仅在不会造成结果偏差的情况下使用。测试的目的是确保处理正确进行；我们不应该在没有实际测试函数的情况下简单地赋值。虽然模拟部分是有效的，我们可以保证不会影响单元测试过程，但必须记住，必须执行该函数才能验证其真实行为。
因此，当我们向此函数添加其他类型的处理时，混合方法就更有意义。该策略允许真实操作和模拟操作的有效结合，确保测试更加稳健和可靠
结论
模拟是创建有效测试的宝贵盟友，尤其是在使用 PySpark 和其他云服务时。我们在 Python 中使用单元测试探索的实现不仅帮助我们模拟操作，而且还保持了数据和流程的完整性。借助模拟提供的灵活性，我们可以测试管道，而不必担心对生产环境造成严重破坏。那么，准备好迎接下一个挑战了吗？在下一篇文章中，我们将深入探讨与 AWS 和 GCP 服务集成的世界，展示如何模拟这些调用并确保您的管道完美运行。下次见！
以上是模拟，它们是什么？的详细内容。更多信息请关注PHP中文网其他相关文章！
