資料管道單元測試與模擬技術

– 10 分鐘閱讀

在建構穩健的資料管道時，單元測試是不可或缺的環節。透過單元測試，開發者可以及早發現程式碼缺陷，確保資料處理邏輯的正確性。本文將著重於資料管道中單元測試的實踐，特別是如何有效地處理外部依賴，例如 API 和雲端服務。透過模擬這些依賴，我們可以隔離測試環境，提高測試速度和穩定性，同時降低測試成本。以下將探討如何使用 Python 的 unittest.mock 和 responses 函式庫來模擬 API 請求，以及如何結合 pytest fixtures 測試雲端服務的互動，並提供程式碼範例和解析，幫助讀者理解和應用這些技術。

單元測試在資料管道中的重要性

在開發資料管道時，單元測試扮演著至關重要的角色。它讓開發者能夠在低層級上快速評估程式碼的有效性，確保資料處理邏輯、連線以及資料的建立、修改和刪除過程正確無誤。本篇文章將探討單元測試在資料管道中的應用，並介紹如何識別依賴項和使用模擬（Mocks）來取代這些依賴項。

識別依賴項

在進行單元測試之前，首先需要識別出程式碼中的依賴項。這些依賴項可能包括資料來源、外部API、雲端服務等。瞭解這些依賴項有助於我們決定如何替換它們，以提高測試的有效性和速度。

資料邏輯與資料修改過程的依賴項

下表列出了資料邏輯和資料修改過程中涉及的依賴項及其相關測試的注意事項：

單元介面資料備註
資料驗證調查桶（Survey bucket）調查資料透過和失敗的案例
取得郵遞區號地理編碼服務（Geocoding service）地理編碼API回應
提取郵遞區號並新增到資料中
提取物種有效的調查資料有或無物種的案例
社交媒體豐富化提取物種、HoD資料函式庫、臨時資料桶（Temp Data bucket）提取物種、HoD資料函式庫在HoD資料函式庫中有或無匹配的案例
Lambda：夜鷺識別豐富化的社交媒體內容桶（Enrich with social bucket）豐富化的社交媒體內容有或無夜鷺的案例
提取物種，建立臨時資料臨時資料桶（Temp Data bucket）提取物種驗證資料是否儲存在雲端
社交媒體豐富化，刪除臨時資料臨時資料桶（Temp Data bucket）社交媒體豐富化在重試案例中，驗證臨時資料是否被刪除
社交媒體豐富化，建立結果資料社交媒體豐富化桶（Enrich with social bucket）社交媒體豐富化驗證結果資料是否儲存在雲端

連線測試和可觀察性測試

連線測試主要關注與外部服務（如資料函式庫、API、雲端儲存）的連線問題，例如重試機制和錯誤程式碼處理。這些測試本身就是介面，不依賴於特定的資料。可觀察性測試，如記錄日誌和生成指標，可以作為相關測試的一部分進行。

使用模擬（Mocks）取代依賴項

為了提高測試的有效性和降低成本，我們可以使用模擬來取代真實的依賴項。這樣不僅可以減少對雲端資源的依賴，還能加快測試速度。接下來的章節將介紹如何為常見的資料管道依賴項建立模擬。

如何建立模擬

建立模擬的第一步是評估測試替身（test double）的放置位置和效能。然後，可以使用常見的Python模組和客戶端模擬函式庫來建立模擬。例如，可以使用Python的unittest.mock函式庫來模擬API呼叫或雲端服務互動。

程式碼範例：使用unittest.mock模擬API呼叫
import unittest
from unittest.mock import patch, MagicMock
import requests

def fetch_data(url):
    response = requests.get(url)
    if response.status_code == 200:
        return response.json()
    else:
        return None

class TestDataFetcher(unittest.TestCase):

    @patch('requests.get')
    def test_fetch_data_success(self, mock_get):
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.json.return_value = {'data': 'some data'}
        mock_get.return_value = mock_response
        
        data = fetch_data('https://example.com/api/data')
        self.assertEqual(data, {'data': 'some data'})

    @patch('requests.get')
    def test_fetch_data_failure(self, mock_get):
        mock_response = MagicMock()
        mock_response.status_code = 404
        mock_get.return_value = mock_response
        
        data = fetch_data('https://example.com/api/data')
        self.assertIsNone(data)

if __name__ == '__main__':
    unittest.main()

內容解密：

  1. 測試目標：本範例展示瞭如何使用unittest.mock函式庫來模擬requests.get呼叫，以測試fetch_data函式。
  2. @patch裝飾器：用於替換requests.get方法，使其傳回一個預設的回應物件。
  3. MagicMock物件：用於建立模擬回應物件，可以設定其屬性和方法傳回值。
  4. 測試案例：包含兩個測試案例，分別測試了API呼叫成功和失敗的情況。
  5. 斷言：使用assertEqual和assertIsNone來驗證函式的傳回值是否符合預期。

這個範例演示瞭如何使用模擬技術來隔離外部依賴，從而更有效地進行單元測試。

測試替身（Test Doubles）技術解析

在軟體開發與測試過程中，測試替身扮演著至關重要的角色。本章節將探討測試替身的概念、應用場景及其在實際開發中的重要性。

測試替身的基本型別

測試替身主要包含三種型別：模擬物件（Mocks）、存根（Stubs）及偽物件（Fakes）。每種型別都有其特定的使用場景和技術特點。

模擬物件（Mocks）

模擬物件主要用於模擬服務行為，特別是在與客戶端函式庫互動時，能夠有效地模擬方法呼叫和回應。這種技術在測試與外部服務互動的程式碼時尤其有用。

存根（Stubs）

存根可以視為模擬物件的簡化版本，主要關注於提供特定的回應，而非模擬完整的服務行為。這種方法在不關注服務內部實作細節時特別有效。

偽物件（Fakes）

偽物件是一種真實的服務實作，但通常會被簡化或區域性實作，例如使用本地小型資料函式庫取代生產環境中的大型資料函式庫，以提高測試效率和降低成本。

替換依賴元件的考量

使用測試替身的主要目的是為了隔離被測試程式碼與外部依賴之間的耦合度，從而提高測試的穩定性和效率。然而，不當的使用測試替身可能會隱藏程式碼中的問題。

放置位置

在建立測試替身時，應盡可能地將其放置在與被替換依賴元件的介面附近。這樣可以確保測試替身能夠準確地模擬真實的互動行為。

# geocoding.py 中的 lat_long_to_pop 方法
def lat_long_to_pop(lat_long):
    zipcode = get_zip(lat_long)
    pop = get_population(zipcode)
    return {zipcode: pop}

在測試上述程式碼時，直接對 get_zip 和 get_population 方法進行存根處理可能會忽略對真實 API 呼叫行為的測試，從而導致測試覆寫率不足。

# 使用模擬物件進行測試
@mock.patch('geocoding.get_zip', mock.Mock(return_value='95472'))
@mock.patch('geocoding.get_population', mock.Mock(return_value='1000'))
def test_lat_long_to_pop():
    assert lat_long_to_pop((38.4021, 122.8239)) == {'95472':'1000'}

依賴元件的穩定性

當依賴元件處於變動狀態時，例如使用 beta 版本的功能，直接建立測試替身可能並不合理。在這種情況下，採用整合測試可能更為合適。

複雜度與重要性的權衡

建立和維護測試替身本身也是一種額外的負擔。因此，需要根據被測試程式碼的重要性和複雜度進行權衡。

通用介面的模擬

在建立模擬物件時，需要考慮系統中的介面互動。例如，在 HoD 問卷資料處理流程中，「驗證資料」和「取得郵遞區號」兩個步驟之間的介面可以透過提供偽資料進行單元測試。

圖表說明

此圖示呈現了資料處理流程中的關鍵步驟及其相互關係。

@startuml
skinparam backgroundColor #FEFEFE
skinparam sequenceArrowThickness 2

title 資料管道單元測試與模擬技術

actor "客戶端" as client
participant "API Gateway" as gateway
participant "認證服務" as auth
participant "業務服務" as service
database "資料庫" as db
queue "訊息佇列" as mq

client -> gateway : HTTP 請求
gateway -> auth : 驗證 Token
auth --> gateway : 認證結果

alt 認證成功
    gateway -> service : 轉發請求
    service -> db : 查詢/更新資料
    db --> service : 回傳結果
    service -> mq : 發送事件
    service --> gateway : 回應資料
    gateway --> client : HTTP 200 OK
else 認證失敗
    gateway --> client : HTTP 401 Unauthorized
end

@enduml

圖表翻譯： 此圖示展示了資料處理流程的主要步驟。首先，資料會經過驗證，接著會取得郵遞區號，最後進行資料處理。每個步驟都是流程中的關鍵環節。

模擬通用介面以提升測試效率

在軟體開發過程中，外部依賴（如API介面）的測試是不可或缺的一環。為了確保程式碼的穩定性和可靠性，我們需要有效地模擬這些外部介賴的行為。本篇文章將探討如何使用模擬（Mocking）技術來測試與外部介面的互動。

為何需要模擬外部介面

當程式碼與外部服務（如Geocoding服務）互動時，直接測試這些互動可能會遇到諸多問題，例如：

  • 外部服務可能不穩定或不可用
  • 測試可能涉及額外的成本（如API呼叫次數限制）
  • 測試結果可能受到外部服務變更的影響

透過模擬外部介面，我們可以：

  • 隔離外部依賴，提高測試的穩定性
  • 加速測試過程，因為不需要等待外部服務的回應
  • 更精確地控制測試條件，模擬各種可能的回應（包括錯誤情況）

使用unittest.mock進行模擬

Python的unittest.mock函式庫提供了一套強大的工具來建立和管理模擬物件。以下是一個使用unittest.mock來模擬requests函式庫的例子，用於測試get_zip函式：

@mock.patch('geocoding.requests', autospec=True)
def test_get_zip_404(mock_requests):
    mock_requests.get.return_value.status_code = 404
    mock_requests.get.return_value.json.return_value = {}
    with pytest.raises(GeocodingError):
        get_zip((38.4021, 122.8239))

內容解密：

  1. @mock.patch('geocoding.requests', autospec=True)：這行程式碼使用mock.patch裝飾器來替換geocoding模組中的requests物件。autospec=True確保模擬物件具有與被替換物件相同的屬性和方法，從而避免因拼寫錯誤而導致的靜默失敗。
  2. mock_requests.get.return_value.status_code = 404：設定模擬的requests.get方法傳回的物件的status_code屬性為404，模擬API傳回404錯誤。
  3. mock_requests.get.return_value.json.return_value = {}：設定模擬的JSON回應為空字典。
  4. with pytest.raises(GeocodingError):：預期get_zip函式在接收到404回應時丟擲GeocodingError。

驗證成功的API呼叫

同樣地，我們可以測試成功的API呼叫：

@mock.patch('geocoding.requests', autospec=True)
def test_get_zip_ok(mock_requests):
    mock_requests.get.return_value.status_code = 200
    mock_requests.get.return_value.json.return_value = {"zipcode": "95472"}
    assert get_zip((38.4021, 122.8239)) == "95472"
    mock_requests.get.assert_called_once_with(
        url=geocoding.GEOCODING_API,
        params={"lat_long": (38.4021, 122.8239)}
    )

內容解密：

  1. 設定模擬的API回應為200 OK，並傳回包含郵政編碼的JSON物件。
  2. 驗證get_zip函式傳回正確的郵政編碼。
  3. 使用assert_called_once_with驗證requests.get被正確呼叫一次，引數包括正確的URL和引數。

使用responses函式庫進行API模擬

除了使用unittest.mock，我們還可以使用responses函式庫來簡化API模擬的過程。以下是一個例子：

import responses

@responses.activate
def test_get_zip_ok_resp():
    responses.add(
        responses.GET,
        geocoding.GEOCODING_API,
        json={"zipcode": "95472"},
        status=200
    )
    assert get_zip((38.4021, 122.8239)) == "95472"

內容解密：

  1. @responses.activate裝飾器啟動responses函式庫的功能。
  2. responses.add方法新增一個模擬的GET請求回應，傳回包含郵政編碼的JSON物件和200狀態碼。

使用 Mock 進行 API 測試與雲端服務模擬

在軟體開發中，測試是確保程式碼品質的重要環節。特別是在與外部服務（如 API 或雲端服務）互動時，良好的測試策略能夠幫助開發者捕捉潛在問題並提升程式的可靠性。本篇文章將介紹如何使用 unittest.mock 和 responses 函式庫來模擬 API 請求，以及如何結合 pytest fixtures 來測試雲端服務的互動。

模擬 API 請求

當程式碼依賴外部 API 時，直接測試這些 API 呼叫可能會遇到諸如網路不穩定、API 速率限制等問題。透過模擬（mocking），我們可以在隔離環境中測試程式碼邏輯，而無需實際呼叫外部服務。

使用 responses 模擬 API 請求

responses 是一個用於模擬 HTTP 請求的 Python 函式庫。它允許開發者定義當特定請求發生時應傳回的回應。

@responses.activate()
def test_get_zip_ok_resp():
    zip_resp = responses.get(
        geocoding.GEOCODING_API, status=200,
        json={"zipcode": "95472"})
    assert get_zip((38.4021, 122.8239)) == "95472"
    assert zip_resp.call_count == 1

程式碼解析

  • 使用 @responses.activate() 裝飾器啟用 responses 的模擬功能。
  • responses.get() 定義了一個對 GEOCODING_API 的 GET 請求模擬，回傳 HTTP 200 狀態碼和指定的 JSON 回應。
  • 測試 get_zip() 函式的行為，驗證其傳回正確的郵遞區號並檢查 API 被呼叫的次數。

測試 API 重試邏輯

在真實世界中，API 請求可能會因為網路問題或伺服器端錯誤而失敗。實作重試邏輯能夠提升程式的健壯性。

@tenacity.retry(retry=tenacity.retry_if_exception_type(GeocodingRetryException),
                stop=tenacity.stop_after_attempt(5),
                wait=tenacity.wait_exponential(multiplier=1, min=4, max=10),
                reraise=True)
def get_zip_retry(lat_long):
    response = requests.get(GEOCODING_API, {"lat_long": lat_long})
    if response.status_code == 429:
        raise GeocodingRetryException()
    # ...

程式碼解析

  • 使用 tenacity 函式庫實作重試機制，當遇到 GeocodingRetryException 時進行重試。
  • 最多重試 5 次，並採用指數退避策略等待下一次重試。

使用 unittest.mock 和 responses 測試重試邏輯

@responses.activate(registry=OrderedRegistry)
def test_get_zip_retry():
    responses.get(geocoding.GEOCODING_API, status=429, json={})
    responses.get(geocoding.GEOCODING_API, status=429, json={})
    responses.get(geocoding.GEOCODING_API, status=200, json={"zipcode": "95472"})
    zip = get_zip_retry((38.4021, 122.8239))
    assert zip == "95472"

程式碼解析

  • 使用 OrderedRegistry 確保 responses.get() 的呼叫順序與註冊順序一致。
  • 模擬兩次 429 回應後接著一次 200 回應，測試重試邏輯是否正確執行。

測試雲端服務互動

對於雲端服務（如 AWS S3 或 Google Cloud Storage），可以使用 unittest.mock 或特定的 mocking 函式庫（如 moto for AWS）來模擬服務互動。

自行建立 Mocks

對於沒有官方 mocking 函式庫的雲端服務，可以使用 unittest.mock 自行建立 mocks。

@mock.patch('geocoding.requests', autospec=True)
def test_get_zip_retry_mock(mock_requests):
    # ...

程式碼解析

  • 使用 @mock.patch 裝飾器替換 geocoding.requests 為一個 mock 物件。
  • 設定 mock 物件的行為以模擬不同的 API 回應。
玄貓 BlackCat

玄貓 BlackCat

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。