雲端服務模擬測試實踐

在雲端服務開發中，有效測試至關重要。本文介紹如何運用模擬技術測試與 Google Cloud Storage（GCS）和 Amazon S3 等雲端服務互動的程式碼。對於 GCS，我們使用 mock.patch 模擬客戶端行為，驗證程式碼邏輯的正確性。而針對 S3，Moto 框架提供便捷的模擬環境，簡化測試流程。此外，文章也涵蓋了設定 AWS 憑證 Fixture、測試資料函式庫的運用，以及在醫療資料管理系統中進行 ETL 流程測試的策略，提供全面的雲端服務測試。

雲端服務模擬測試的深度解析

在現代軟體開發中，測試是確保程式碼品質的關鍵步驟。尤其是在處理雲端服務時，如何有效地進行單元測試是一個重要的課題。本文將探討如何使用模擬（Mocking）技術來測試雲端服務的相關程式碼，並以Google Cloud Storage（GCS）和Amazon S3為例，展示不同的模擬實作方式。

使用模擬技術測試GCS相關程式碼

在測試與GCS互動的程式碼時，直接連線到GCS進行測試不僅複雜，而且存在風險。使用模擬技術可以有效地解決這個問題。以下是一個使用delete_temp函式的例子，該函式負責刪除GCS中指定bucket和prefix下的所有物件。

原始程式碼解析

from google.cloud import storage

def delete_temp(bucket_name, prefix):
    storage_client = storage.Client()
    bucket = storage_client.get_bucket(bucket_name)
    blobs = bucket.list_blobs(prefix)
    for blob in blobs:
        blob.delete()

模擬測試實作

要測試delete_temp函式，我們需要模擬GCS客戶端的行為。以下是一個使用mock.patch進行模擬的測試範例：

@mock.patch('cloud_examples.storage', autospec=True)
def test_delete_temp(storage):
    blob = mock.Mock(spec=Blob)
    blob.delete.return_value = None
    mock_bucket = storage.Client.return_value.get_bucket.return_value
    mock_bucket.list_blobs.return_value = [blob, blob]
    
    cloud_examples.delete_temp("fake_bucket", "fake_prefix")
    assert blob.delete.call_count == 2
    
    client_mock = storage.Client.return_value
    client_mock.get_bucket.assert_called_with("fake_bucket")
    client_mock.get_bucket.return_value.list_blobs.assert_called_with("fake_prefix")

#### 內容解密：

1. 模擬GCS客戶端：使用mock.patch裝飾器模擬cloud_examples.storage，這樣我們就可以控制GCS客戶端的行為而不必實際連線GCS。
2. 模擬Blob物件：建立一個模擬的Blob物件，並設定其delete方法的傳回值。這樣我們就可以驗證delete方法是否被正確呼叫。
3. 驗證函式行為：透過檢查blob.delete.call_count，我們可以確認delete_temp函式是否正確地刪除了指定的物件。
4. 檢查客戶端呼叫：驗證GCS客戶端的方法是否被正確地呼叫，包括取得bucket和列出blobs。

使用Moto測試AWS S3相關程式碼

對於AWS S3，使用Moto函式庫可以簡化模擬測試的過程。Moto提供了一個pytest fixture，可以用來模擬S3環境。

原始程式碼解析

import boto3

def delete_temp_aws(bucket_name, prefix):
    s3 = boto3.client('s3', region_name='us-east-1')
    objects = s3.list_objects_v2(Bucket=bucket_name, Prefix=prefix)
    object_keys = [{'Key': item['Key']} for item in objects['Contents']]
    s3.delete_objects(Bucket=bucket_name, Delete={'Objects': object_keys})

模擬測試實作

def test_delete_temp_aws(s3):
    s3.create_bucket(Bucket="fake_bucket")
    s3.put_object(Bucket="fake_bucket", Key="fake_prefix/something", Body=b'Some info')
    
    delete_temp_aws("fake_bucket", "fake_prefix")
    obj_response = s3.list_objects_v2(Bucket="fake_bucket", Prefix="fake_prefix")
    assert obj_response['KeyCount'] == 0

#### 內容解密：

1. 使用Moto模擬S3環境：透過pytest fixture s3，Moto為我們建立了一個模擬的S3環境。
2. 建立測試資料：在模擬環境中建立一個bucket和一個物件，以供測試使用。
3. 執行測試：呼叫delete_temp_aws函式，並驗證物件是否被成功刪除。
4. 驗證結果：透過檢查list_objects_v2的傳回值，確認指定prefix下的物件是否已被刪除。

使用Moto進行雲端服務模擬測試

在進行雲端服務相關的單元測試時，Moto是一個非常有用的工具。它允許開發者模擬AWS服務的行為，從而避免在測試過程中對真實的AWS資源進行操作。

設定AWS憑證Fixture

@pytest.fixture(scope="function")
def aws_credentials():
    os.environ['AWS_ACCESS_KEY_ID'] = 'testing'
    # ...

這個fixture設定了環境變數中的AWS憑證。這些憑證是虛擬的，並不會真正用於連線AWS服務。

內容解密：

• aws_credentials fixture的作用是為測試提供一個虛擬的AWS憑證環境。
• 透過設定AWS_ACCESS_KEY_ID為'testing'，模擬了AWS憑證的存在。
• 這種做法確保了在測試過程中不會意外地連線到真實的AWS服務。

使用Moto模擬S3服務

@pytest.fixture(scope="function")
def s3(aws_credentials):
    with mock_s3():
        yield boto3.client('s3', region_name='us-east-1')

這個fixture利用Moto的mock_s3功能來模擬S3服務，並建立了一個S3客戶端。

內容解密：

• s3 fixture依賴於aws_credentials fixture，確保在模擬S3服務之前已經設定了虛擬的AWS憑證。
• mock_s3()上下文管理器用於模擬S3服務，使得在這個上下文中對S3的操作都是虛擬的。
• yield boto3.client('s3', region_name='us-east-1')提供了S3客戶端給測試使用。
• 使用yield而不是return，確保了S3客戶端的上下文在測試完成後能夠正確地被清理。

測試中的狀態管理

在進行測試時，狀態管理是非常重要的。Moto的fixture預設作用域是函式級別，這意味著每個測試函式都會重新執行fixture，從而避免了不同測試之間的狀態幹擾。

驗證Moto模擬的效果

透過註解掉建立bucket和object的程式碼，可以觀察到測試失敗並報出NoSuchBucket錯誤，這驗證了Moto模擬的有效性。

使用測試資料函式庫進行測試

對於涉及資料函式庫操作的程式碼，使用測試資料函式庫是一種常見的做法。這樣可以避免對生產資料函式庫造成影響，同時也減少了測試的成本。

資料函式庫選擇的一致性

理想情況下，測試資料函式庫應該與生產資料函式庫使用相同的型別，以確保測試結果的準確性。

醫療資料管理系統的ETL流程測試

在一個醫療資料管理系統中，ETL（提取、轉換、載入）流程涉及大量的資料函式庫操作。使用測試資料函式庫可以有效地對這些操作進行測試。

ETL流程示意圖

@startuml
skinparam backgroundColor #FEFEFE
skinparam componentStyle rectangle

title 雲端服務模擬測試實踐

package "AWS 雲端架構" {
    package "網路層" {
        component [VPC] as vpc
        component [Subnet] as subnet
        component [Security Group] as sg
        component [Route Table] as rt
    }

    package "運算層" {
        component [EC2] as ec2
        component [Lambda] as lambda
        component [ECS/EKS] as container
    }

    package "儲存層" {
        database [RDS] as rds
        database [DynamoDB] as dynamo
        storage [S3] as s3
    }

    package "服務層" {
        component [API Gateway] as apigw
        component [ALB/NLB] as lb
        queue [SQS] as sqs
    }
}

apigw --> lambda
apigw --> lb
lb --> ec2
lb --> container
lambda --> dynamo
lambda --> s3
ec2 --> rds
container --> rds
vpc --> subnet
subnet --> sg
sg --> rt

@enduml

圖表翻譯： 此圖示展示了醫療資料管理系統中的ETL流程。首先，從Patient Data表中提取Treatment Info，然後透過匹配處理生成Match Results。根據Match Results中的OK欄位值，將資料分為兩路：OK=True的資料進入Ingest表，而OK=False的資料則進入Review表等待審核。

內容解密：

• ETL流程涉及多個步驟，包括資料提取、轉換和載入。
• 測試資料函式庫的使用確保了每個步驟的正確性和可靠性。
• 圖表清晰地展示了資料流向和處理邏輯。

使用測試資料函式庫進行單元測試

在開發資料處理流程（ETL）時，使用測試資料函式庫是一種常見的做法。這種方法允許開發者在受控的環境中測試資料處理邏輯，而不會影響到生產環境中的資料。本文將探討如何使用測試資料函式庫進行單元測試，並介紹相關的最佳實踐。

為什麼使用測試資料函式庫？

使用測試資料函式庫的主要優點是可以在一個乾淨、可控的環境中測試資料處理邏輯。這種方法可以確保測試結果的可靠性和一致性，避免因為生產環境中的資料變化而導致測試結果的不確定性。

如何設定測試資料函式庫？

設定測試資料函式庫需要考慮幾個關鍵因素：

1. 選擇適合的測試資料函式庫工具：有一些現成的工具和套件可以幫助設定和管理測試資料函式庫，例如 pytest-postgresql。這些工具通常會在本地或容器中安裝一個資料函式庫例項。
2. 組態測試資料函式庫連線：需要組態測試框架以連線到測試資料函式庫。在本例中，使用了 SQLAlchemy 來管理資料函式庫連線。
3. 建立和銷毀測試資料函式庫：使用 pytest 的 fixture 功能，可以在測試會話開始時建立測試資料函式庫，並在會話結束時銷毀它。

程式碼範例：建立和銷毀測試資料函式庫

@pytest.fixture(scope="session")
def test_db():
    engine = setup_test_db()
    yield engine
    teardown_test_db(engine)

內容解密：

• setup_test_db() 函式負責建立測試資料函式庫。
• yield engine 陳述式確保在測試會話期間，資料函式庫連線保持開啟。
• teardown_test_db(engine) 函式負責在測試會話結束後銷毀測試資料函式庫。

管理測試資料函式庫的狀態

為了確保測試的可靠性，需要仔細管理測試資料函式庫的狀態。這包括建立必要的表格、插入測試資料，以及在測試結束後清理資料。

程式碼範例：建立表格和插入資料

def create_tables(conn):
    conn.execute("""
        CREATE TABLE treatment (
            id SERIAL PRIMARY KEY,
            name VARCHAR(255)
        );
        INSERT INTO treatment (name) VALUES ('Drug A');
    """)

@pytest.fixture(scope="session")
def test_conn(test_db):
    test_engine = create_engine(f"postgresql://{creds}@{host}/test_db")
    test_conn = test_engine.connect()
    create_tables(test_conn)
    yield test_conn
    test_conn.close()
    test_engine.dispose()

內容解密：

• create_tables 函式建立了 treatment 表格並插入了一條測試資料。
• test_conn fixture 建立了資料函式庫連線，並在測試會話結束後關閉連線和釋放引擎資源。

除錯測試失敗

當測試失敗時，可能需要檢查測試資料函式庫的狀態來除錯問題。可以使用自定義的命令列選項 --persist-db 來保留測試資料函式庫，以便除錯。

程式碼範例：保留測試資料函式庫

def pytest_addoption(parser):
    parser.addoption("--persist-db", action="store_true", help="Do not teardown the test db at the end of the session")

@pytest.fixture(scope="session")
def test_db(request):
    engine = setup_test_db()
    yield engine
    if request.config.getoption("--persist-db"):
        return
    teardown_test_db(engine)

內容解密：

• --persist-db 命令列選項允許使用者選擇是否保留測試資料函式庫。
• 在 test_db fixture 中，根據 --persist-db 選項的值決定是否銷毀測試資料函式庫。

使用測試替身最佳化資料管線測試

在測試資料管線時，如何有效地取代外部依賴是提升測試效率和降低成本的關鍵。本章節將探討多種技術來取代資料管線中的外部依賴，包括使用 Mock 物件、測試資料函式庫以及 pytest fixtures 來簡化測試流程。

取代資料函式庫依賴的測試策略

在進行單元測試時，使用 db_engine 這個 fixture 有多種方式可以實作。第六章介紹瞭如何使用抽象層來提供不同的物件儲存可能性，包括一個可以用於測試的 MockStorage 類別。同樣地，如果程式碼接受資料函式庫引擎作為引數，可以使用 TestDatabase 類別來連線到測試資料函式庫，或者將引擎作為引數傳遞給方法。

使用 Mock.patch 進行資料函式庫連線的測試

另一種方法是使用 mock.patch 來對資料函式庫連線進行 Mock，其中 engine 或 conn 物件被替換。這種方法可以避免在測試中使用真實的資料函式庫連線，從而降低測試成本和複雜度。

測試資料函式庫的使用

使用測試資料函式庫可以在不連線到外部資料函式庫的情況下進行測試，從而實作 CI 測試、降低管理分享測試環境的複雜度，並且避免雲端成本和資料損壞的風險。Docker Compose 技術可以用於在 CI 測試中包含測試資料函式庫容器。

建立有效的 Mock 物件

本章節介紹了多種建立和使用 Mock 物件的方法。responses 和 moto 函式庫分別為 API 和 AWS 服務提供了現成的 Mock 物件。Python 的標準 unittest 函式庫提供了豐富的工具來對 Mock 物件進行設定、驗證和控制。結合 pytest fixtures，可以簡化測試程式碼並管理測試狀態。

建立良好測試替身的原則

• 在依賴和程式碼之間的介面處放置測試替身。
• 重點模擬關鍵互動，並記錄未預期或罕見的情況。
• 注意跨多個測試分享的 Mock 和 Fake 物件中的狀態累積。
• 使用規格（如 autospec）確保 Mock 物件準確代表依賴。
• 在針對 Mock 進行測試時，驗證請求和回應承載、錯誤處理和重試。

練習與探索

更多 Moto Mock 的練習

在 cloud_examples.py 中，hod_has_night_heron_data 方法用於檢查是否有夜鷺資料。利用 Moto 建立 Mock 的技術，為這個方法撰寫單元測試，驗證當指定字首有夜鷺資料和沒有夜鷺資料時的情況。

Mock 放置的練習

給定一個物種和郵遞區號，get_hod_matches 方法會傳回一個 DataFrame，其中包含與這兩個引數相匹配的 HoD 資料函式庫中的 ID 和內容。在不重寫 SQL 的情況下，想出一個方法來測試 get_hod_matches，包括當沒有結果和有一或多個結果傳回時的情況。

第9章：測試資料

在前一章中，您瞭解瞭如何替換資料管道測試中的兩個依賴項之一：外部服務介面。這讓您朝向成本效益測試邁進了一步。本章將介紹如何替換第7章中提到的第二個依賴項：外部資料來源。您將瞭解如何使用合成資料替換實時資料來源進行測試。

本章中有許多建立合成資料的巧妙技術，但在啟動IDE之前，評估是否用合成資料替換資料依賴項是正確的選擇至關重要。本章首先提供了有關如何選擇實時資料或合成資料進行測試的指導，以及每種方法的優缺點。

使用實時資料的優缺點

優點

使用來自生產環境等實時資料，可以提供豐富、規模龐大且具有多樣性的資料特徵。如果使用生產資料的測試能夠透過，那麼至少可以保證程式碼更改不會破壞生產環境。這是相當不錯的結果。

當處理特徵變化不可預測或結構複雜、難以人工建立的資料時，使用實時資料更有可能獲得準確的測試結果。

缺點

使用實時資料進行測試時，絕對必須避免因測試活動而損壞生產資料。曾有一位同事分享了一個故事，某公司的開發人員在本地單元測試資料管道程式碼時增加了生產環境憑證。結果，由於執行單元測試，他們意外刪除了產品中最重要服務中最重要表中的所有生產資料，導致了完全的中斷和重大聲譽損害。

因此，建議將實時資料的使用限制在只讀範圍內。如果測試需要建立、修改或刪除資料，最好使用偽造資料，如測試資料函式庫或模擬資料。

使用實時資料進行測試時，需要額外的步驟來連線和驗證資料來源。根據公司政策和所涉及的資料型別，可能需要時間獲得存取批准，並可能需要額外的時間來匿名化資料以進行測試。您還需要組態本地開發環境以存取這些資源，並確保測試程式碼能夠驗證和連線。

實時資料測試案例

曾經參與的一個產品有兩個資料函式庫：一個儲存查詢組態，另一個儲存查詢所執行的源資料。每天新增的查詢組態可能會參照源資料中的任何專案；也許今天新增的查詢組態參照了昨天新增的源資料中的新表。

由於查詢組態和源資料都處於不斷變化的狀態，我們的團隊無法用合成版本替換實時資料，同時仍保持全面的程式碼覆寫率。相反，我們為本地開發建立了組態資料函式庫的副本，其佔用空間較小。對於源資料，我們連線到測試環境中的資料函式庫。透過這種設定，我們可以在本地環境中對實時資料執行單元測試。

這種方法提供了我們所需的測試覆寫率，但由於使用了這些實時服務，我們無法在CI中自動化單元測試。為了覆寫所有查詢組態的測試，我們在測試環境中每天執行一次端對端測試，然後再將程式碼提升到生產環境。我們發現潛在錯誤的迴路較長，但考慮到資料的複雜性，這是一個合理的折衷方案。

匿名化資料的考量

在某些情況下，可能需要對資料進行匿名化處理，以刪除或遮蔽敏感資訊，如個人身份資訊（PII）或受保護的健康資訊（PHI）。如果您遇到這種情況，請立即諮詢公司的法律和安全團隊。

在某些情況下，匿名化資料的過程可能會移除準確測試所需的關鍵資料特徵。如果需要使用匿名化資料進行測試，請將其與生產資料進行比較，以確保資料特徵保持一致，然後再根據它進行測試結果的投注。

測試對資源的影響

使用實時資料進行測試時，還需要考慮測試對提供或儲存資料的資源的影響。例如，在圖8-2中，HoD社交媒體資料函式庫正在處理和提供HoD網站的內容。網站的效能可能會受到額外負載的影響，因為資料函式庫正在進行測試。在這種情況下，提供專用的只讀副本或快照供測試使用是一個不錯的替代方案。

圖表翻譯：

此圖示說明瞭HoD社交媒體資料函式庫對網站效能的影響以及使用專用只讀副本的重要性。

結語

總之，在選擇使用實時資料還是合成資料進行測試時，需要綜合考慮多種因素，包括測試目標、資料特徵、以及對資源的影響等。透過仔細評估這些因素，可以選擇最合適的測試策略，以確保測試的有效性和效率。