模擬，它們是什麼？

本文是我將在此處和我的個人部落格上發布的一系列關於資料處理應用程式測試的文本中的第一篇。

當我從軟體工程師轉變為資料工程師時，我開始與資料領域沒有軟體工程背景的人進行對話。在這些對話中，反覆出現一個問題：如何寫測驗？

事實上，對於那些不習慣編寫測試的人來說，編寫測試似乎是一項複雜的任務，因為它需要改變編寫程式碼的方式。事實上，這並不神秘，而是一個練習和重複的問題。我在本文中的主要目標是指導剛起步的您，展示如何為處理資料的應用程式建立測試，確保程式碼的品質和可靠性。

本文是我將在接下來的幾週內推出的系列文章的一部分，我將在其中分享如何在針對資料工程的程式碼中編寫自動化測試。在今天的文章中，我想探討一些關於模擬的內容。在一些程式碼場景中，資料管道將進行連接、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中文網其他相關文章！