單元測試金字塔與工作單元模式

軟體系統的測試工作需要有效組織，才能提升效率和可維護性。本文將探討如何運用服務層測試和工作單元模式 (UoW) 實作測試金字塔。首先，透過新增 API 端點簡化 E2E 測試，將測試重心轉移至服務層，接著引入 UoW 模式管理資料函式庫操作，確保資料一致性。UoW 模式作為服務層的一部分，與儲存函式庫協同工作，簡化 API 層與資料函式庫層、倉函式庫層的互動。文章進一步探討領域模型中的不變數和約束，以及領域物件如何維護其內部一致性，最後介紹一致性邊界和聚合根的概念，為後續使用事件和訊息匯流排改進系統奠定基礎。

單元測試金字塔與工作單元模式：提升測試效率與解耦

在開發複雜的軟體系統時，如何有效地組織測試是一個重要的課題。本文將探討如何透過服務層測試和工作單元模式（Unit of Work, UoW）來實作測試金字塔，並提高測試的效率和系統的可維護性。

從端對端測試到服務層測試

在前面的章節中，我們已經看到了如何透過引入服務層來簡化端對端測試（E2E測試）。透過將業務邏輯封裝在服務層中，我們可以將大部分的測試轉移到服務層進行，從而提高測試的效率。

API 端點的簡化

為了進一步簡化E2E測試，我們可以新增一個API端點來新增批次（batch），這樣就可以避免直接操作資料函式庫。透過服務函式的支援，實作這個端點非常簡單，只需要進行一些JSON資料的處理和呼叫服務函式即可。

@app.route("/add_batch", methods=['POST'])
def add_batch():
    session = get_session()
    repo = repository.SqlAlchemyRepository(session)
    eta = request.json['eta']
    if eta is not None:
        eta = datetime.fromisoformat(eta).date()
    services.add_batch(
        request.json['ref'], request.json['sku'], request.json['qty'], eta,
        repo, session
    )
    return 'OK', 201

內容解密：

• 使用Flask框架建立了一個處理POST請求的API端點/add_batch。
• 從請求的JSON資料中提取必要的資訊，如參考編號（ref）、產品SKU、數量（qty）和預計到貨日期（eta）。
• 如果eta不為空，則將其轉換為日期物件。
• 呼叫services.add_batch函式，將批次資訊新增到系統中。
• 傳回狀態碼201，表示建立成功。

E2E 測試的改進

透過新增的API端點，我們可以簡化E2E測試，不再需要直接操作資料函式庫。這樣，E2E測試就變得更加乾淨和獨立，只依賴於API。

def post_to_add_batch(ref, sku, qty, eta):
    url = config.get_api_url()
    r = requests.post(
        f'{url}/add_batch',
        json={'ref': ref, 'sku': sku, 'qty': qty, 'eta': eta}
    )
    assert r.status_code == 201

@pytest.mark.usefixtures('postgres_db')
@pytest.mark.usefixtures('restart_api')
def test_happy_path_returns_201_and_allocated_batch():
    sku, othersku = random_sku(), random_sku('other')
    earlybatch = random_batchref(1)
    laterbatch = random_batchref(2)
    otherbatch = random_batchref(3)
    post_to_add_batch(laterbatch, sku, 100, '2011-01-02')
    post_to_add_batch(earlybatch, sku, 100, '2011-01-01')
    post_to_add_batch(otherbatch, othersku, 100, None)
    data = {'orderid': random_orderid(), 'sku': sku, 'qty': 3}
    url = config.get_api_url()
    r = requests.post(f'{url}/allocate', json=data)
    assert r.status_code == 201
    assert r.json()['batchref'] == earlybatch

內容解密：

• 定義了一個輔助函式post_to_add_batch，用於透過API新增批次。
• 在測試函式test_happy_path_returns_201_and_allocated_batch中，使用post_to_add_batch新增了幾個批次，並進行了分配測試。
• 驗證了API傳回的狀態碼和分配結果。

工作單元模式（Unit of Work）

工作單元模式是一種設計模式，用於維護一系列操作的原子性。在我們的系統中，它將協助實作服務層與資料層之間的解耦。

為什麼需要工作單元模式？

在目前的實作中，API層直接與資料函式庫層、倉函式庫層（Repository）和服務層進行互動。這種緊密的耦合使得系統難以維護和擴充套件。透過引入工作單元模式，我們可以簡化這種互動，使API層只需要初始化工作單元並呼叫服務即可。

工作單元模式的實作

工作單元模式將作為服務層的一部分，與倉函式庫層協同工作。下面是服務層使用工作單元模式後的樣子：

def allocate(
        orderid: str, sku: str, qty: int,
        uow: unit_of_work.AbstractUnitOfWork
) -> str:
    line = OrderLine(orderid, sku, qty)
    with uow:
        batches = uow.batches.list()
        # ...

內容解密：

• allocate函式現在接受一個uow（工作單元）物件作為引數。
• 使用with uow:陳述式來確保工作單元的正確初始化和清理。
• 在with塊內部，透過uow.batches.list()取得批次列表。

使用工作單元（Unit of Work）模式管理資料函式庫操作

在軟體開發中，特別是在處理資料函式庫操作時，保持資料的一致性和完整性是非常重要的。工作單元（Unit of Work，簡稱UoW）模式是一種設計模式，它幫助我們管理資料函式庫事務，確保資料的一致性和完整性。本文將探討UoW模式的概念、實作及其在測試中的應用。

UoW模式的概念

UoW模式是一種用於管理資料函式庫操作的設計模式。它將所有的資料函式庫操作視為一個單元，要麼全部成功，要麼全部失敗，從而保持資料的一致性。UoW模式提供了對資料函式庫事務的管理，包括提交（commit）和回復（rollback）操作。

UoW模式的實作

在我們的例子中，UoW模式是透過AbstractUnitOfWork抽象基礎類別來定義的。它具有以下關鍵特性：

1. 批次倉儲（Batches Repository）

UoW提供了一個batches屬性，這是一個批次倉儲的介面，讓我們可以存取和操作批次資料。

2. 上下文管理器（Context Manager）

UoW實作了上下文管理器協定，即__enter__和__exit__方法。當我們使用with陳述式進入UoW上下文時，__enter__方法被呼叫；當我們離開該上下文時，__exit__方法被呼叫。這使得我們可以在進入和離開上下文時執行必要的設定和清理工作。

3. 提交和回復

UoW提供了commit和rollback方法，分別用於提交和回復資料函式庫事務。如果在UoW上下文中發生異常，__exit__方法會自動呼叫rollback來回復事務。

SQLAlchemy實作的UoW

我們的具體實作是根據SQLAlchemy的SqlAlchemyUnitOfWork類別。它利用SQLAlchemy的會話（session）來管理資料函式庫操作。

class SqlAlchemyUnitOfWork(AbstractUnitOfWork):
    def __init__(self, session_factory=DEFAULT_SESSION_FACTORY):
        self.session_factory = session_factory

    def __enter__(self):
        self.session = self.session_factory()
        self.batches = repository.SqlAlchemyRepository(self.session)
        return super().__enter__()

    def __exit__(self, *args):
        super().__exit__(*args)
        self.session.close()

    def commit(self):
        self.session.commit()

    def rollback(self):
        self.session.rollback()

在這個實作中，__enter__方法啟動一個新的資料函式庫會話，並建立一個SqlAlchemyRepository例項來操作批次資料。commit和rollback方法分別呼叫會話的提交和回復方法。

在測試中使用假的UoW

在測試中，我們可以使用一個假的UoW來隔離依賴，加速測試執行。假的UoW模擬了真實UoW的行為，但不實際操作資料函式庫。

class FakeUnitOfWork(unit_of_work.AbstractUnitOfWork):
    def __init__(self):
        self.batches = FakeRepository([])
        self.committed = False

    def commit(self):
        self.committed = True

    def rollback(self):
        pass

這個假的UoW可以用於服務層的測試，讓我們能夠專注於業務邏輯的測試，而不必擔心底層的資料函式庫操作。

單元工作模式（Unit of Work）與其在服務層的應用

在軟體開發中，特別是在涉及資料函式庫操作的應用程式中，如何有效地管理資料的存取和變更是一個重要的議題。單元工作模式（Unit of Work, UoW）提供了一種解決方案，使得我們能夠將多個相關的操作封裝在一起，作為一個原子單位來執行，從而保證資料的一致性和完整性。

為什麼不直接模擬 Session？

在討論 UoW 之前，我們先來探討為什麼不直接模擬 SQLAlchemy 的 Session 物件。雖然模擬 Session 可以達到快速測試的目的，但它暴露了太多的持久層相關的功能，容易導致資料存取程式碼散落在整個程式碼函式庫中。相反，UoW 提供了一個更簡單的抽象層，明確地分隔了職責，使得服務層只需關心業務邏輯，而不需要直接處理資料函式庫的複雜操作。

服務層中使用 UoW

我們的服務層現在依賴於抽象的 UoW 介面，這使得我們的程式碼更加簡潔和易於測試。例如，在 add_batch 和 allocate 函式中，我們使用 UoW 來新增批次和分配訂單，所有的資料函式庫操作都被封裝在 UoW 的上下文中。

def add_batch(ref: str, sku: str, qty: int, eta: Optional[date], uow: unit_of_work.AbstractUnitOfWork):
    with uow:
        uow.batches.add(model.Batch(ref, sku, qty, eta))
        uow.commit()

def allocate(orderid: str, sku: str, qty: int, uow: unit_of_work.AbstractUnitOfWork) -> str:
    line = OrderLine(orderid, sku, qty)
    with uow:
        batches = uow.batches.list()
        if not is_valid_sku(line.sku, batches):
            raise InvalidSku(f'Invalid sku {line.sku}')
        batchref = model.allocate(line, batches)
        uow.commit()
        return batchref

程式碼解析：

1. with uow: 陳述式用於建立 UoW 的上下文，這確保了在區塊內的所有操作要麼全部成功，要麼在出現異常時全部回復。
2. uow.batches.add(model.Batch(ref, sku, qty, eta)) 將新的批次加入到 UoW 管理的批次集合中。
3. uow.commit() 明確地提交變更到資料函式庫，這使得我們的程式碼在預設情況下是安全的，避免了意外的變更。
4. allocate 函式 內部邏輯保證了只有有效的 SKU 才會被分配，並且分配邏輯是根據當前可用的批次。

明確的提交/回復行為測試

為了驗證 UoW 的提交和回復行為是否正確，我們編寫了測試案例來檢查在不同場景下的行為。

def test_rolls_back_uncommitted_work_by_default(session_factory):
    uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory)
    with uow:
        insert_batch(uow.session, 'batch1', 'MEDIUM-PLINTH', 100, None)
    new_session = session_factory()
    rows = list(new_session.execute('SELECT * FROM "batches"'))
    assert rows == []

def test_rolls_back_on_error(session_factory):
    class MyException(Exception):
        pass
    uow = unit_of_work.SqlAlchemyUnitOfWork(session_factory)
    with pytest.raises(MyException):
        with uow:
            insert_batch(uow.session, 'batch1', 'LARGE-FORK', 100, None)
            raise MyException()
    new_session = session_factory()
    rows = list(new_session.execute('SELECT * FROM "batches"'))
    assert rows == []

測試解析：

1. 第一個測試驗證了預設情況下，UoW 會回復未提交的工作。
2. 第二個測試檢查了當發生異常時，UoW 是否正確地回復了變更。

使用 UoW 將多個操作組合成一個原子單位

UoW 的一個重要好處是它能夠將多個相關操作封裝成一個原子單位。這在需要確保資料一致性的場景下尤為重要。例如，在重新分配訂單或更改批次數量時，我們希望這些操作要麼全部成功，要麼全部失敗。

def reallocate(line: OrderLine, uow: AbstractUnitOfWork) -> str:
    with uow:
        batch = uow.batches.get(sku=line.sku)
        if batch is None:
            raise InvalidSku(f'Invalid sku {line.sku}')
        batch.deallocate(line)
        allocate(line)
        uow.commit()

重新分配邏輯解析：

1. batch.deallocate(line) 解除原有的分配。
2. allocate(line) 重新分配訂單。
3. uow.commit() 提交變更，確保操作的原子性。

單元工作模式（Unit of Work）與一致性邊界

在前一章中，我們討論了單元工作（UoW）模式的重要性及其在維護資料完整性方面的作用。現在，我們將探討領域模型中的不變數和約束，並瞭解領域物件如何維護其內部一致性。

為何需要單元工作模式？

單元工作模式提供了一種抽象機制，以確保資料操作的原子性。它與儲存函式庫（Repository）和服務層（Service Layer）模式緊密合作，共同構建了一個強大的資料存取抽象層。

單元工作模式的優缺點

為什麼要抽象化 SQLAlchemy 的 Session？

SQLAlchemy 的 Session 物件已經實作了單元工作模式，但我們仍然需要將其抽象化。原因在於：

1. 簡化介面：我們的 UnitOfWork 簡化了 Session 的介面，使其只保留了最核心的功能：啟動、提交或放棄事務。
2. 存取 Repository 物件：UnitOfWork 提供了一個便捷的方式，讓客戶端程式碼可以存取 Repository 物件。

一致性邊界與聚合根

現在，讓我們來討論一致性邊界的概念。如圖 7-1 所示，我們將引入一個新的模型物件 Product，用於包裝多個批次，並將原來的 allocate() 領域服務方法移至 Product 上。

為什麼需要聚合根？

聚合根（Aggregate Root）是一種領域模型設計模式，用於定義一致性邊界。在我們的例子中，Product 就是一個聚合根，它負責維護內部的一致性。

class Product:
    def __init__(self, product_id: str, batches: List[Batch]):
        self.product_id = product_id
        self.batches = batches

    def allocate(self, order: Order) -> str:
        # 在此實作分配邏輯
        pass

使用 Plantuml 圖表示聚合根關係

@startuml
skinparam backgroundColor #FEFEFE
skinparam componentStyle rectangle

title 單元測試金字塔與工作單元模式

package "測試金字塔架構" {
    package "測試層級" {
        component [E2E 測試] as e2e
        component [服務層測試] as service
        component [單元測試] as unit
    }

    package "UoW 模式" {
        component [AbstractUoW] as abstract
        component [SQLAlchemy UoW] as sqlalchemy
        component [Fake UoW] as fake
    }

    package "領域模型" {
        component [聚合根 Product] as aggregate
        component [Repository] as repo
        component [一致性邊界] as boundary
    }
}

e2e --> service : API 端點
service --> unit : 業務邏輯
unit --> abstract : 依賴注入
abstract --> sqlalchemy : 生產環境
abstract --> fake : 測試環境
sqlalchemy --> repo : batches
repo --> aggregate : 領域物件
aggregate --> boundary : 不變數維護

note right of abstract
  UoW 介面：
  - __enter__/__exit__
  - commit()
  - rollback()
end note

note right of aggregate
  一致性保證：
  - 原子操作
  - 事務管理
  - 資料完整性
end note

@enduml

此圖示展示了 Product、Batch 和 OrderLine 之間的關係。Product 是聚合根，負責維護 Batch 的一致性。

內容解密：

1. 單元工作模式：提供了一種抽象機制，用於確保資料操作的原子性。
2. 聚合根：定義了一致性邊界，負責維護內部的一致性。
3. 簡化介面：UnitOfWork 簡化了 Session 的介面，使其只保留了最核心的功能。

透過本章的學習，我們對領域模型設計和資料存取抽象層有了更深入的瞭解。下一章中，我們將繼續探討如何使用事件和訊息匯流排來進一步改進我們的系統。