資料管線單元測試實務應用

在資料工程中，確保資料管線的穩定和正確性至關重要。單元測試提供了一種有效的方法來驗證資料處理邏輯的各個組成部分。本文將介紹如何在資料管線中實施單元測試，涵蓋資料邏輯驗證、外部服務連線測試，以及如何透過 Mock 物件和相關函式庫簡化測試流程。此外，文章也將探討如何模擬 API 回應、測試重試機制，以及使用特定工具模擬雲端服務，以提升測試效率並降低對外部環境的依賴。

單元測試在資料管線中的重要性與實務應用

在現代的資料工程領域中，資料管線（Data Pipeline）的測試是確保資料處理流程正確性的關鍵步驟。單元測試（Unit Testing）作為測試策略的一部分，扮演著至關重要的角色。本篇文章將探討單元測試在資料管線中的應用，並介紹如何有效地進行測試。

資料邏輯測試

資料邏輯測試主要關注資料處理過程中的核心邏輯。這包括：

• 資料驗證：檢查輸入資料是否符合預期的格式和內容。
• 資料轉換：驗證資料在處理過程中的轉換邏輯是否正確。
• 特定條件處理：例如，識別夜鷺（Night Heron）資料的 Lambda 函式。

程式碼範例：資料驗證邏輯

def validate_data(data):
    # 資料驗證邏輯
    if 'required_field' not in data:
        raise ValueError("Missing required field")
    return data

#### 內容解密：
# 1. 函式 validate_data 負責檢查輸入資料是否包含必要欄位
# 2. 若缺少必要欄位 'required_field'，則丟擲 ValueError
# 3. 資料驗證透過則傳回原始資料

連線測試

連線測試關注的是資料管線與外部服務之間的連線，例如：

• 雲端儲存服務：測試與雲端儲存的連線是否穩定。
• 資料函式庫連線：檢查與資料函式庫的互動是否正常。
• 地理編碼服務：驗證地理編碼 API 的呼叫是否正確。

連線測試重點

• 處理重試邏輯（Retries）
• 處理不同錯誤程式碼（Error Codes）

可觀測性測試

可觀測性測試涉及將資訊傳送到日誌和監控工具，例如：

• 連線失敗指標：生成連線失敗的指標。
• 錯誤資料指標：記錄被標記為錯誤的資料。
• 日誌訊息：在步驟失敗時記錄日誌訊息。

資料修改流程測試

資料修改流程測試關注的是建立、刪除或修改資料的操作，例如：

• 提取物種資訊並建立臨時資料。
• 使用社交資料豐富調查資料並刪除臨時資料。

識別依賴關係

在進行單元測試時，識別依賴關係至關重要。需考慮每個單元測試所需的介面和資料，並記錄相關依賴。

識別依賴關係表格範例

使用 Mock 物件進行測試

在單元測試中，使用 Mock 物件來模擬真實依賴，可以提高測試的有效性和速度，並減少對外部資源的依賴。

為何使用 Mock 物件？

• 減少雲端成本
• 加速測試流程
• 提高測試覆寫率

測試替身（Test Doubles）：模擬物件的應用與考量

在軟體開發的測試過程中，測試替身（Test Doubles）是一種常見且重要的技術。測試替身主要包括模擬物件（Mocks）、存根（Stubs）和偽物件（Fakes），這些技術能夠協助開發者在隔離依賴的情況下進行單元測試或整合測試，從而提高測試的效率和可靠性。

模擬物件、存根與偽物件的區別

• 模擬物件（Mocks）：模擬物件是一種能夠模擬真實服務行為的技術。透過模擬物件，開發者可以程式設計來模擬特定方法的呼叫和回應，特別是在與外部服務（如 CSP 客戶端函式庫）互動時非常有用。模擬物件能夠有效地隔離外部依賴，使得測試更加穩定和可控。
• 存根（Stubs）：存根可以被視為模擬物件的簡化版本。與模擬物件不同的是，存根主要關注於提供預設的回應，而不關心服務的行為細節。存根適用於那些不需要驗證服務行為的測試場景。
• 偽物件（Fakes）：偽物件是一種真實的服務實作，但通常規模較小或功能較為簡化。例如，在測試環境中使用一個較小的本地資料函式庫來取代生產環境中的大型資料函式庫。

取代依賴的考量

使用測試替身時，需要考慮以下幾個重要的因素：

1. 放置位置：測試替身應該盡可能地靠近與被取代的依賴介面。這樣可以確保測試替身能夠有效地模擬真實的服務行為，並且避免因為放置位置不當而遺漏重要的測試路徑。
2. 依賴的穩定性：如果依賴的服務或介面處於變動狀態，例如 beta 產品功能，那麼建立測試替身可能並不划算。在這種情況下，直接使用真實的服務或整合測試可能是更好的選擇。
3. 複雜度與重要性：建立和維護測試替身需要額外的成本。因此，需要根據被測試程式碼的重要性和複雜度來評估是否值得建立測試替身。

模擬通用介面

在建立模擬物件時，可以參考系統中的通用介面。例如，在 HoD 調查資料管道中，「驗證資料」和「取得郵遞區號」步驟之間的介面可以透過提供偽資料來進行單元測試。

# 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}

# 使用模擬物件測試 lat_long_to_pop 方法
@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'}

內容解密：

1. lat_long_to_pop 方法根據給定的經緯度座標，取得對應區域的人口數量。
2. get_zip 和 get_population 方法分別負責取得郵遞區號和對應的人口數量。
3. 在測試 lat_long_to_pop 方法時，使用 @mock.patch 裝飾器來模擬 get_zip 和 get_population 方法的行為，避免實際呼叫外部 API。
4. 透過模擬物件，可以控制方法的回傳值，進而驗證 lat_long_to_pop 方法的正確性。

模擬通用介面：測試外部依賴

在測試過程中，經常需要處理外部依賴，例如介於「取得郵遞區號」步驟和地理編碼服務之間的介面。為了測試這些互動，可能需要使用模擬（mocks）來取代真實的外部介面。與外部介面互動通常涉及以下動作：

• 發出請求
• 處理回應
• 偵測和處理連線問題

有多種方法可以用模擬取代依賴。其中一種方法是依賴注入（dependency injection），這在第6章中已經介紹過。與其使用連線到依賴的模組，不如傳入一個在測試中提供模擬回應的模組。

使用模擬來測試API回應

在 test_geocoding.py 中的 test_get_zip_404 測試是一個使用 unittest.mock 來模擬API回應的例子：

@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))

在這個測試中，requests 函式庫被替換為一個模擬物件 mock_requests。當在 get_zip 方法中執行 requests.get(GEOCODING_API) 時，會呼叫模擬物件而不是真實的 requests 模組。

內容解密：

1. @mock.patch('geocoding.requests', autospec=True)：使用 unittest.mock 的 patch 裝飾器，將 geocoding.requests 替換為一個模擬物件。
2. mock_requests.get.return_value.status_code = 404：設定模擬物件的回應狀態碼為404。
3. mock_requests.get.return_value.json.return_value = {}：設定模擬物件的回應JSON內容為空字典。
4. with pytest.raises(GeocodingError):：預期 get_zip 方法會引發 GeocodingError 例外。

驗證成功的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"

內容解密：

1. 設定模擬物件的回應狀態碼為200，表示成功。
2. 設定模擬物件的回應JSON內容包含郵遞區號。
3. 驗證 get_zip 方法的回傳值是否正確。

驗證請求引數

@mock.patch('geocoding.requests', autospec=True)
def test_get_zip_ok(mock_requests):
    # ...
    mock_requests.get.assert_called_once_with(
        url=geocoding.GEOCODING_API,
        params={"lat_long": (38.4021, 122.8239)}
    )

內容解密：

1. 使用 assert_called_once_with 方法驗證 requests.get 方法是否被呼叫一次，並且引數是否正確。
2. 驗證請求的URL和引數是否符合預期。

使用responses函式庫

def test_get_zip_ok_resp(mock_responses):
    # 使用responses函式庫來模擬API行為
    pass

內容解密：

1. 使用 responses 函式庫來模擬API回應，簡化測試程式碼。

未來可以進一步探討如何使用其他測試工具和框架來簡化測試流程，並且提高測試的可靠性和效率。同時，也可以研究如何將這些測試技巧應用於其他型別的外部介面，例如資料函式庫或訊息佇列。

此圖示顯示了測試外部介面的流程：

@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

此圖示描述了測試外部介面的基本步驟，包括建立模擬物件、設定模擬回應、呼叫被測試方法、驗證回傳值和請求引數等。透過這個流程，可以確保外部介面的互動正確無誤。

模擬外部介面與雲端服務測試

在軟體開發過程中，測試是確保程式碼品質的重要環節。當程式碼依賴外部介面或雲端服務時，測試的複雜度會增加。本文將介紹如何使用 unittest.mock 和 responses 函式庫來模擬外部介面，以及如何結合 pytest fixtures 來測試雲端服務。

使用 Responses 模擬 API 回應

responses 函式庫允許開發者模擬 requests 函式庫的行為，從而在測試中控制 API 的回應。以下是一個範例：

@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

內容解密：

1. @responses.activate()：啟用 responses 函式庫的模擬功能。
2. responses.get()：模擬一個 GET 請求到指定的 API 地址，並傳回一個狀態碼為 200 的回應，回應內容為 JSON 格式的資料。
3. get_zip((38.4021, 122.8239))：呼叫被測試的函式 get_zip，並傳入經緯度座標。
4. assert 陳述式：驗證函式的傳回值是否正確，以及模擬的 API 請求是否被正確呼叫。

測試重試邏輯

當程式碼依賴外部介面時，重試邏輯是必要的，以處理暫時性的錯誤。以下是一個使用 tenacity 函式庫實作重試邏輯的範例：

@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()
    # ...

內容解密：

1. @tenacity.retry()：啟用重試邏輯。
2. retry=tenacity.retry_if_exception_type(GeocodingRetryException)：指定在丟擲 GeocodingRetryException 時進行重試。
3. stop=tenacity.stop_after_attempt(5)：最多重試 5 次。
4. wait=tenacity.wait_exponential()：重試之間的等待時間採用指數退避策略。
5. reraise=True：如果重試次數達到上限仍失敗，則重新丟擲異常。

使用 Mocks 測試重試邏輯

要測試重試邏輯，需要模擬不同的 API 回應。以下是一個範例：

@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"

內容解密：

1. @responses.activate(registry=OrderedRegistry)：啟用 responses 的模擬功能，並使用有序登入檔來保證 API 請求的順序。
2. 多個 responses.get()：模擬連續的三次 API 請求，分別傳回狀態碼 429、429 和 200。
3. get_zip_retry((38.4021, 122.8239))：呼叫被測試的重試函式。

雲端服務的模擬

對於雲端服務的測試，可以使用特定的模擬函式庫，如 AWS 的 moto。以下是一個範例：

# 使用 moto 模擬 AWS 服務
import moto

@moto.mock_s3
def test_s3_upload():
    # 測試上傳到 S3 的程式碼
    pass

內容解密：

1. @moto.mock_s3：啟用 moto 對 S3 服務的模擬。
2. test_s3_upload()：測試上傳到 S3 的函式。