斷路器模式與 Mock 物件單元測試

斷路器模式能有效提升分散式系統的穩定性及容錯能力，避免因外部服務故障導致的連鎖反應。它透過監控外部服務呼叫，並在錯誤次數超過閾值時停止呼叫，直到服務還原正常。此機制類別似現實生活中的斷路器，能防止系統過載並提高用性。文章以 Python 和 pybreaker 函式庫示範斷路器的實作，並解析程式碼的關鍵部分，包含斷路器定義、脆弱函式設計以及主函式的迴圈呼叫。除了斷路器，文章也提到了其他分散式系統模式，如 CQRS、兩階段提交、Saga 模式等，提供更全面的架構設計參考。此外，文章強調單元測試的重要性，特別是在處理外部依賴時，Mock 物件和依賴注入模式能有效隔離測試目標，確保測試的可靠性和可重複性。文章以實際案例說明如何運用 Mock 物件模擬檔案寫入和 API 呼叫，並示範如何結合依賴注入模式，提升程式碼的模組化和可測試性。

分散式系統模式：斷路器模式詳解

在開發分散式系統時，容錯（Fault Tolerance, FT）是確保系統穩定執行的關鍵。斷路器模式（Circuit Breaker pattern）是一種重要的設計模式，用於處理與外部元件通訊時可能出現的長期故障。本篇文章將探討斷路器模式的原理、應用場景及實作方法。

斷路器模式的原理

斷路器模式的核心思想是透過監控對外部服務或元件的呼叫，當錯誤次數達到一定閾值時，斷路器會「跳閘」，後續的呼叫將直接傳回錯誤，而無需實際執行該操作。這樣可以避免浪費資源在重試可能失敗的請求上，從而提高系統的回應速度和穩定性。

現實世界的類別比

斷路器模式的名稱來源於現實中的電路斷路器。當電路中檢測到過載或短路時，斷路器會自動斷開電路，防止進一步的損壞。在軟體系統中，這種機制同樣適用於處理外部服務的故障。

斷路器模式的應用場景

電子商務結帳流程：如果支付閘道服務不可用，斷路器可以停止進一步的支付嘗試，防止系統過載。
速率受限的API呼叫：當API呼叫達到其速率限制時，斷路器可以阻止額外的請求，避免受到懲罰。

實作斷路器模式

以下是一個使用Python和pybreaker函式庫實作斷路器模式的例子：

import pybreaker
from datetime import datetime
import random
from time import sleep

# 定義斷路器，當連續失敗2次時開啟斷路器，5秒後重置
breaker = pybreaker.CircuitBreaker(fail_max=2, reset_timeout=5)

@breaker
def fragile_function():
    if not random.choice([True, False]):
        print(" / OK", end="")
    else:
        print(" / FAIL", end="")
        raise Exception("這是一個範例例外")

def main():
    while True:
        print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), end="")
        try:
            fragile_function()
        except Exception as e:
            print(" / {} {}".format(type(e), e), end="")
        finally:
            print("")
        sleep(1)

if __name__ == "__main__":
    main()

程式碼解析：

斷路器的定義：breaker = pybreaker.CircuitBreaker(fail_max=2, reset_timeout=5)定義了一個斷路器，當函式fragile_function連續失敗2次時，斷路器將開啟，後續呼叫將直接傳回錯誤。5秒後，斷路器將重置，允許再次呼叫該函式。
脆弱函式的定義：fragile_function是一個模擬可能失敗的函式，使用@breaker裝飾器將其納入斷路器的監控之下。
主函式：main函式迴圈呼叫fragile_function，並輸出呼叫結果和時間。

輸出結果：

執行上述程式碼，可以觀察到當fragile_function連續失敗達到閾值時，斷路器開啟，後續呼叫立即傳回錯誤。經過5秒後，斷路器重置，允許再次呼叫。

其他分散式系統模式

除了斷路器模式外，還有多種分散式系統模式可用於解決不同的挑戰，例如：

CQRS（命令查詢責任分離）：分離讀寫操作的責任，以最佳化資料存取和可擴充套件性。
兩階段提交：一種分散式事務協定，確保跨多個資源的原子性和一致性。
Saga模式：一系列本地事務形成一個分散式事務，提供補償機制以維護一致性。
Sidecar模式：佈署輔助服務以增強主服務的功能，如監控、日誌記錄或安全功能。
服務註冊：集中管理服務的註冊和發現，促進服務之間的通訊和可擴充套件性。
艙壁模式：透過隔離資源或元件來防止故障擴散，提高容錯能力和彈性。

測試模式：開發可靠的分散式系統

在前面的章節中，我們討論了架構模式以及針對特定使用案例（如平行處理或效能）的模式。在本章中，我們將探討特別適用於測試的設計模式。這些模式有助於隔離元件，使測試更加可靠，並促程式式碼的可重用性。

本章將涵蓋以下主要主題：

模擬物件模式（Mock Object pattern）
依賴注入模式（Dependency Injection pattern）

技術需求

請參閱第1章中介紹的技術需求。

模擬物件模式

模擬物件模式是一種強大的工具，用於在測試過程中隔離元件，模擬其行為。模擬物件有助於建立可控的測試環境，並驗證元件之間的互動。

模擬物件模式的三個特點

隔離：模擬物件隔離了被測試的程式碼單元，確保測試在可控的環境中執行，依賴項可預測且無外部副作用。
行為驗證：使用模擬物件，可以驗證測試過程中是否發生了特定的行為，例如方法呼叫或屬性存取。
簡化：模擬物件簡化了測試的設定，透過替換複雜的真實物件，這些物件可能需要大量的設定。

與存根（Stub）的比較

存根也替換真實實作，但僅用於向被測試程式碼提供間接輸入。模擬物件則可以驗證互動，使其在許多測試場景中更加靈活。

實際例子

現實世界中有以下類別似的概念或工具：

飛行模擬器，用於複製駕駛飛機的體驗，讓飛行員在可控和安全的環境中學習如何處理各種飛行場景。
心肺復甦（CPR）假人，用於教導學生如何有效地進行CPR，模擬人體，提供真實且可控的學習環境。
撞擊測試假人，汽車製造商用於模擬人類對車輛碰撞的反應，提供有關汽車安全功能的寶貴資料，而無需冒著實際人類生命的風險。

模擬物件模式的使用案例

在單元測試中，模擬物件用於替換被測試程式碼的複雜、不可靠或不可用的依賴項。這使得開發人員能夠專注於單元本身，而不是它與外部系統的互動。例如，在測試一個從API取得資料的服務時，模擬物件可以模擬API，傳回預定義的回應，確保服務能夠處理各種資料場景或錯誤，而無需與實際API互動。

實作模擬物件模式

假設我們有一個函式，將訊息記錄到檔案中。我們可以模擬檔案寫入機制，以確保我們的記錄函式將預期的內容寫入日誌檔案，而無需寫入實際檔案。下面是使用Python的unittest模組實作此功能的範例：

import unittest
from unittest.mock import mock_open, patch

class Logger:
    def __init__(self, filepath):
        self.filepath = filepath

    def log(self, message):
        with open(self.filepath, "a") as file:
            file.write(f"{message}\n")

class TestLogger(unittest.TestCase):
    def test_log(self):
        msg = "Hello, logging world!"
        m_open = mock_open()
        with patch("builtins.open", m_open):
            logger = Logger("dummy.log")
            logger.log(msg)
            #### 內容解密：
            # 在這個範例中，我們首先定義了一個Logger類別，它具有將訊息記錄到檔案的功能。
            # 然後，我們使用unittest.mock.patch()函式來模擬Python內建的open()函式。
            # 這樣，我們就可以在不實際寫入檔案的情況下測試Logger類別的log()方法。
            # 我們驗證了log()方法是否正確地呼叫了open()函式，並寫入了預期的訊息。

圖表翻譯：

此圖示展示了模擬物件模式的基本原理，包括如何使用模擬物件隔離被測試元件，以及如何驗證元件之間的互動。

  graph LR
    A[被測試元件] -->|互動|> B[模擬物件]
    B -->|驗證|> C[測試結果]

圖表翻譯： 此圖示說明瞭被測試元件如何與模擬物件互動，以及如何驗證測試結果。

使用 Mock 物件進行單元測試的最佳實踐

在軟體開發中，單元測試是確保程式碼品質的重要環節。當程式碼依賴外部資源（如檔案系統、資料函式庫或網路服務）時，如何有效地隔離這些依賴，進行可靠的測試，成為一大挑戰。unittest.mock 模組提供了一套強大的工具，幫助開發者透過模擬（mocking）技術來解決這一問題。

關於 `mock_open`

當呼叫 mock_open() 函式時，它會傳回一個組態好的 Mock 物件，使其行為類別似於內建的 open() 函式。此 Mock 物件被設定為模擬檔案操作，例如讀取和寫入。

from unittest.mock import mock_open

# 使用 mock_open 模擬 open() 函式
m_open = mock_open()

內容解密：

mock_open() 用於建立一個模擬的檔案開啟操作，使測試不依賴真實檔案系統。
傳回的 Mock 物件可用於檢查檔案操作是否正確執行。

關於 `unittest.mock.patch`

unittest.mock.patch 用於在測試過程中替換物件為 Mock 物件。其引數包括指定要替換的目標物件，以及可選的替換物件、屬性規範、副作用定義和傳回值等。

from unittest.mock import patch

# 使用 patch 裝飾器替換 open() 函式
@patch("builtins.open", new_callable=mock_open)
def test_log_message(m_open):
    # 測試程式碼
    pass

內容解密：

@patch("builtins.open", new_callable=mock_open) 將內建的 open() 函式替換為由 mock_open() 傳回的 Mock 物件。
這使得測試能夠控制和檢查檔案操作，而無需實際操作檔案系統。

使用 Mock 物件進行單元測試

以下是一個使用 Mock 物件測試日誌記錄功能的範例：

import unittest
from unittest.mock import patch, mock_open

def log_message(msg):
    with open("dummy.log", "a") as f:
        f.write(f"{msg}\n")

class TestLogMessage(unittest.TestCase):
    @patch("builtins.open", new_callable=mock_open)
    def test_log_message(self, m_open):
        msg = "Test message"
        log_message(msg)
        
        # 檢查 open() 是否以正確引數被呼叫
        m_open.assert_called_once_with("dummy.log", "a")
        
        # 檢查 write() 是否寫入正確訊息
        m_open().write.assert_called_once_with(f"{msg}\n")

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

內容解密：

log_message 函式將訊息寫入指定的日誌檔案。
使用 @patch("builtins.open", new_callable=mock_open) 裝飾器，將 open() 函式替換為 Mock 物件。
在測試方法中，呼叫 log_message 函式，並檢查 open() 和 write() 方法是否以預期引數被呼叫。

依賴注入（Dependency Injection）模式

依賴注入模式透過將類別的依賴以外部實體傳遞，而不是在類別內部建立，促進了鬆耦合、模組化和可測試性。

真實世界的例子

電器裝置與電源插座：各種電器可以插入不同的電源插座，而不需要直接和永久的接線。
相機鏡頭：攝影師可以根據環境更換相機鏡頭，而無需更換相機本身。

使用案例

在網頁應用程式中，將資料函式庫連線物件注入到倉函式庫或服務元件中，提高了模組化和可維護性。這使得在不同資料函式庫引擎或組態之間切換變得容易，而無需直接更改元件的程式碼。

使用 Mock 物件實作依賴注入模式

以下範例展示瞭如何使用依賴注入模式和 Mock 物件進行單元測試：

from typing import Protocol
from unittest.mock import MagicMock

class WeatherApiClient(Protocol):
    def fetch_weather(self, location: str) -> str:
        """取得指定地點的天氣資料"""
        ...

class RealWeatherApiClient:
    def fetch_weather(self, location: str) -> str:
        return f"Real weather data for {location}"

class WeatherService:
    def __init__(self, weather_api: WeatherApiClient):
        self.weather_api = weather_api
    
    def get_weather(self, location: str) -> str:
        return self.weather_api.fetch_weather(location)

class TestWeatherService(unittest.TestCase):
    def test_get_weather(self):
        # 建立 WeatherApiClient 的 Mock 物件
        mock_api = MagicMock(spec=WeatherApiClient)
        mock_api.fetch_weather.return_value = "Mocked weather data"
        
        # 將 Mock 物件注入 WeatherService
        service = WeatherService(mock_api)
        
        # 測試 get_weather 方法
        weather_data = service.get_weather("Taipei")
        self.assertEqual(weather_data, "Mocked weather data")
        
        # 檢查 fetch_weather 是否以正確引數被呼叫
        mock_api.fetch_weather.assert_called_once_with("Taipei")

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

內容解密：

定義 WeatherApiClient 協定，規範天氣 API 客戶端的介面。
RealWeatherApiClient 類別實作了 WeatherApiClient，提供真實的天氣資料取得功能。
WeatherService 類別依賴 WeatherApiClient，透過建構器注入。
在測試中，使用 MagicMock 建立 WeatherApiClient 的 Mock 物件，並設定其行為。
將 Mock 物件注入 WeatherService，進行單元測試。

玄貓

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

斷路器模式與 Mock 物件單元測試

分散式系統模式：斷路器模式詳解

斷路器模式的原理

現實世界的類別比

斷路器模式的應用場景

實作斷路器模式

程式碼解析：

輸出結果：

其他分散式系統模式

測試模式：開發可靠的分散式系統

技術需求

模擬物件模式

模擬物件模式的三個特點

與存根（Stub）的比較

實際例子

模擬物件模式的使用案例

實作模擬物件模式

圖表翻譯：

使用 Mock 物件進行單元測試的最佳實踐

關於 mock_open

內容解密：

關於 unittest.mock.patch

內容解密：

使用 Mock 物件進行單元測試

內容解密：

依賴注入（Dependency Injection）模式

真實世界的例子

使用案例

使用 Mock 物件實作依賴注入模式

內容解密：

玄貓

關於 `mock_open`

關於 `unittest.mock.patch`