作為基礎設施的工具,Pulumi 主要可以採用以下三種方式進行測試,而除了驗證的範疇外,在程式語言的支援上也有所不同,下方的表格總結了三種方式主要的差異:

表格一:Pulumi 測試方法比較
單元測試屬性測試整合測試
實際部署基礎設施是*
需要 Pulumi CLI
執行時間毫秒
語言同 Pulumi 支援的程式語言Node.js 或 Python所有語言
驗證對象資源輸入資源輸入和輸出外部端點

*: 由於屬性測試的檢測會分別在 preview 及 update 兩個時間點進行,在 preview 階段的檢測並不會真正的部署資源,然而屬性測試無法透過 Mock 進行模擬,因此部分的檢測則需在 update 階段實際發生部署後才發生效用。

本篇文章主要針對單元測試進行說明,若對屬性測試有興趣的讀者,可以參考我們的另外一篇文章:為你的自動化基礎設施拉上保險系列之一

我們可以從表格一中看到,Pulumi 單元測試可使用的語言與它本身所支援的程式語言相同,而另一常見的 IaC 工具 Terraform 常搭配的 Terratest,則需對 GO 語言有一定的熟悉度。相較之下,Pulumi 讓用戶可以在面對開發與測試時所需要的能力一致,不用在不同語言間進行切換,並可以基於該程式語言自帶或第三方提供的單元測試框架來撰寫測試,這也稍稍降低了在 IaC 中實踐 TDD(測試驅動開發)的門檻。

接下來的範例將基於先前文章自動化建置 GCP HTTP 負載平衡器所建立的基礎設施進行單元測試,驗證資源的配置是否正確。完整的實作程式碼請至 GitHub 1查看,由於本篇內容範疇為單元測試,因此僅會針對 tests 資料夾進行描述,關於基礎設施的程式碼實作則請參閱先前的文章。

.
├── ...
├── tests
│   ├── __init__.py
│   ├── mocks.py
│   ├── test_backend_server.py
│   ├── test_firewall.py
│   └── test_lb.py
└── ...

備註: 為使測試資料夾下的所有測試自動被識別和執行,檔案名建議以 test 作為開頭。

前置作業

在單元測試框架的選擇上,我們所採用的為 Python 自帶的 unittest(或稱為 PyUnit),因此無須額外安裝套件。

Pulumi 程式說明

1. 使用 Mock

在單元測試中,並不需要真的部屬資源,所以必須先透過 Mock 的方式模擬真正的執行及外部的依賴:

mocks.py

import pulumi

class RuntimeMocks(pulumi.runtime.Mocks):
    def new_resource(self, type_, name, inputs, provider, id_):
        # For the UT of firewall to verify LB address is in whitelist or not
        if name == "load-balancer-address":
            address = {
                "address": "1.2.3.4"
            }
            return [name + '_id', dict(inputs, **address)]
        else:
            return [name + '_id', inputs]
    def call(self, token, args, provider):
        return {}

pulumi.runtime.set_mocks(RuntimeMocks())

補充說明:由於單元測試並非實際執行,因此資源部分的輸出屬性可能為 undefined。但因為我們後續的測試案例需要依賴 Load Balancer 的 IP 來做些驗證,所以在 new_resource() 的模擬中,額外針對該資源的建立定義了 address 回傳值。

並在每個測試檔案呼叫 Pulumi 執行前,匯入上面所建立的 Mock 物件:

from tests import mocks

2. 撰寫測試

透過繼承 unittest.TestCase 基礎類別來建立新的測試案例,為使每個單元測試可以自動被識別,測試方法需以 test 為開頭,每個測試檔案的內容結構如下:

import unittest
from tests import mocks

class TestingXXX(unittest.TestCase):
    # Test Case 1
    @pulumi.runtime.test
    def test_a(self):
        ...
    # Test Case 2
    @pulumi.runtime.test
    def test_b(self):
        ...
...

在了解 unittest 的框架後,我們針對範例中的幾個主要的基礎設施元件撰寫了單元測試:

  1. 後端機器 (👉 test_backend_server.py
    • ✓ [測試案例 1.] 虛擬機器須打上 “webserver” 的標籤
  2. 防火牆 (👉 test_firewall.py
    • ✓ [測試案例 1.] 允許的 port 只有 TCP 80
    • ✓ [測試案例 2.] 允許的來源 IP 只有 Load Balancer 及 Health check
  3. Load Balancer(👉 test_lb.py
    • ✓ [測試案例 1.] URL Map 的 host 配置(i.e., 對所有 host 的請求皆對應到期望的 Path Matcher)
    • ✓ [測試案例 2.] URL Map 的 path matcher 配置(i.e., 根據 path 的請求對應到期望的後端服務)

由於 Pulumi 的資源屬性皆為 outputs,且許多屬性都是異步計算,因此需要透過 apply 接收一個回呼函式(callback function),當異步執行得到結果後,再將真正的屬性值傳遞給回呼函式。

以防火牆的測試案例test_firewall_ports為例:

import network_base

test_network_base = network_base.setup()

class TestingFirewall(unittest.TestCase):
    # check 1: Firewall should only allow TCP port 80
    @pulumi.runtime.test
    def test_firewall_ports(self):
        def check_ports(args):
            ...
        return pulumi.Output.all(test_network_base["compute_firewall"].allows).apply(check_ports)
    ...

測試案例 2.1(允許的 port 只有 TCP 80)需針對防火牆資源的 allows 屬性進行驗證。若在回呼函式check_ports()中印出接收的參數,可以看到屬性 allows 具體的值為:

# print(args[0])
[{'protocol': 'tcp', 'ports': ['80']}]

由於 allows 可以含多組 protocol 與 ports 的配對,為方便進行結果驗證,所以我們稍微將結果進行重組,並透過 assertCountEqual(actual, expected) 來驗證實際結果與預期結果具有相同的對象(不考慮順序):

# check 1: Firewall should only allow TCP port 80
@pulumi.runtime.test
def test_firewall_ports(self):
    def check_ports(args):
        actual_ports = []
        expected_ports = ["tcp_80"]
        for allow in args[0]:
            for port in allow["ports"]:
                actual_ports.append(allow["protocol"] + "_" + port)
        self.assertCountEqual(actual_ports, expected_ports)
    return pulumi.Output.all(test_network_base["compute_firewall"].allows).apply(check_ports)

而在防火牆的另一個測試案例中,則是針對了 source_ranges 進行驗證。由於先前透過 Mock 模擬了 Load Balancer 的 IP(值為 1.2.3.4),因此預期的 source_ranges 值則應為 模擬 Load Balancer IP(1.2.3.4) 及 Health Check 來源 IP(35.191.0.0/16、130.211.0.0/22):

# check 2: Firewall should only accept the request from health probe and LB
@pulumi.runtime.test
def test_source_ranges(self):
    def check_source_ranges(args):
        actual_source_ranges = args[0]
        expected_source_ranges = ["1.2.3.4", "35.191.0.0/16", "130.211.0.0/22"]
        self.assertCountEqual(actual_source_ranges, expected_source_ranges)
    return pulumi.Output.all(test_network_base["compute_firewall"].source_ranges).apply(check_source_ranges)

其餘的測試案例也是依相同的邏輯,因此這邊就不再贅述。


3. 執行測試

完成測試案例的撰寫後,於範例的根目錄執行測試指令:

# -v: 輸出詳細訊息
$ python -m unittest -v

# 測試結果
test_server_tags (tests.test_backend_server.TestingNginx) ... ok
test_firewall_ports (tests.test_firewall.TestingFirewall) ... ok
test_source_ranges (tests.test_firewall.TestingFirewall) ... ok
test_urlmap_host_rules (tests.test_lb.TestingLoadBalancer) ... ok
test_urlmap_path_matchers (tests.test_lb.TestingLoadBalancer) ... ok

----------------------------------------------------------------------
Ran 5 tests in 0.514s

OK

後記

隨著 IaC 及自動化流水線的發展,基礎設施的管理人員可以更一致化且快速的配置資源,然而基礎設施的變更往往牽一髮動全身(例如:DNS 資源記錄的配置錯誤可能造成所有服務請求失敗),因此在快速迭代中,持續提供可信賴的基礎設施絕對是首要關鍵。但由於基礎設施的建置大量依賴了外部資源,在沒有測試環境的情況下,相較於一般的程式開發,要能夠在本地端實踐測試也較為困難。Pulumi 的單元測試,除了讓基礎設施的管理人員可以基於自身熟悉的測試框架撰寫測試外,也協助人員可以在本地端或流水線中藉由模擬的方式完成基礎的單元驗證,並阻斷非預期的變更。


1. Pulumi 完整程式碼