iT邦幫忙

2026 iThome 鐵人賽

DAY 24
1
Software Development

從脆弱腳本到可信任測試平台:自動化測試架構30天系列 第 24

Day 24|平行測試為什麼一開就全部壞掉?共享資源與 Race Condition 解決指南

  • 分享至 

  • xImage
  •  

下指令加上 -n 4 只需要 3 秒鐘;但排查平行執行時產生的靜態變數污染、共用帳號互踢與資料庫 Deadlock,卻需要對系統架構與狀態有極深的理解。」

大家好,我是 Jane,一個每天在第一線處理跨平台(Web & App)自動化測試、跟 CI/CD Pipeline 效能與併發 Issue 奮戰的自動化測試工程師(SDET)。

在上集 Day 23 中,我們展示了如何利用 pytest-xdist 多進程平行執行與 CI 分片(Sharding),將原本 1 小時的測試壓縮到 3 分鐘內。

但許多工程師在興高采烈地安裝完 pytest-xdist 並輸入 pytest -n 4 後,遇到的第一個悲劇往往是:

  • 「原本 100% 全綠的 CI,一開平行竟然瞬間爆了 40 個紅燈!」
  • 「有時這個案例過、有時那個案例失敗,重跑一次失敗的案例又變少了?!」

平行測試會把系統中隱藏最深的 「併發競爭(Race Condition)」「狀態耦合」 無情地暴露出來。

今天這篇文章,我們就來盤點:平行測試一開就全部壞掉的 5 大核心病因,以及第一線 SDET 該如何解決這些共享資源衝突!

一、 直擊病根:平行測試崩潰的 5 大主因

                    ┌─────────────────────────┐
                    │  平行測試崩潰 5 大病因    │
                    └────────────┬────────────┘
                                 │
     ┌───────────────────────────┼───────────────────────────┐
     ▼                           ▼                           ▼
【狀態與資料競爭】            【環境與帳號衝突】            【系統資源與檔案存取】
1. 全域/靜態變數互相塗抹      3. 共用帳號被強迫踢下線       5. 本地檔案/Log 存取衝突 (I/O)
2. 資料庫特定欄位衝突/死鎖    4. 測試環境硬體容量崩潰 (OOM)

1. 全域/靜態變數與 Class 屬性污染(Shared State)

在 Python 程式碼中使用了全域變數(Global Variables)或類別層級屬性(Class Attributes)儲存 Token 或臨時狀態。當多個 Worker 進程同時讀寫時,變數會被互相覆蓋塗抹。

2. 資料庫死鎖與唯一值衝突(DB Race Condition)

多個 Worker 剛好在同一毫秒嘗試往 DB 寫入相同的主鍵(Primary Key)、修改同一筆庫存量、或是觸發資料庫死鎖(Deadlock)。

3. 共用測試帳號互踢下線(Session Displacement)

Web 或 App 系統通常有「單一帳號不允許重複登入」的限制。當 Worker 1 與 Worker 2 同時使用 admin_test 登入時,Worker 1 的 Session 會立刻被 Worker 2 踢下線,導致 Worker 1 後續所有操作跳 HTTP 401 或轉跳登入頁。

4. 本地檔案與日誌存取衝突(File I/O Lock)

多個案例嘗試同時寫入同一個 test_report.log 或清理同一個 downloads/ 資料夾,引發 PermissionErrorFileExistsError

5. 測試環境(Staging Server)併發容量不足

CI 機器一口氣開了 8 個 Worker,同時發送高併發 HTTP 請求,直接把 Staging 環境的 CPU 或後端 API 伺服器打趴(噴出 502 Bad Gateway / 504 Timeout)。

二、 Python / pytest 實戰:4 大解套重構技巧

技巧 1:帳號與資料庫隔離(結合 Worker ID 動態雜湊)

徹底告別寫死帳號!利用 pytest-xdist 提供的 worker_id 確保每個 Worker 使用獨一無二的測試帳號與資料前綴:

Python

# tests/conftest.py
import pytest
import uuid

@pytest.fixture(scope="session")
def worker_id(request):
    """取得當前 xdist Worker 名稱 (如: gw0, gw1, gw2)"""
    config = getattr(request, "config", None)
    if config and hasattr(config, "workerinput"):
        return config.workerinput["workerid"]
    return "master"

@pytest.fixture(scope="function")
def isolated_account(worker_id, api_client):
    """
    動態為當前 Worker 產生專屬帳號,徹底避免「帳號互踢下線」!
    """
    # 帳號帶有 worker_id,例如: user_gw0_a1b2, user_gw1_c3d4
    unique_suffix = f"{worker_id}_{uuid.uuid4().hex[:4]}"
    account_payload = {
        "username": f"user_{unique_suffix}",
        "email": f"test_{unique_suffix}@example.com",
        "password": "SecurePassword123!"
    }

    account = api_client.create_user(account_payload)
    yield account
    # 測試結束後獨立清理
    api_client.delete_user(account["id"])

技巧 2:檔案與快取目錄鎖定(Avoid File Lock)

存取檔案、下載檔或 Log 時,務必搭配 Worker 獨立目錄:

Python

# tests/conftest.py
import os
import pytest

@pytest.fixture(scope="session")
def worker_download_dir(worker_id, tmp_path_factory):
    """
    利用 pytest tmp_path_factory 為每個 Worker 分配獨立的臨時下載資料夾
    """
    # pytest 會自動為每個 gw 建立獨立的臨時路徑
    download_dir = tmp_path_factory.mktemp(f"downloads_{worker_id}")
    return str(download_dir)

技巧 3:精準控制平行粒度(利用 pytest-xdist 分組策略)

某些存在極強內部依賴的測試模組,如果強行跨進程打散執行會非常容易失敗。我們可以透過 pytest-xdist--dist 參數或 @pytest.mark.xdist_group,指定特定測試必須在「同一 Worker」內順序執行:

方式 A:依檔案(File)分組

確保同一 .py 檔案內的測試案例全部由同一個 Worker 承接:

Bash

# --dist loadfile 確保同一個測試檔案集中在同一進程執行
pytest -n 4 --dist loadfile

方式 B:使用 xdist_group 標籤鎖定特定案例

如果某幾個案例絕對不能和其他案例平行,可強制將它們歸在同一個 Group:

Python

# tests/test_payment.py
import pytest

# 標記同屬於 payment_group 的案例,將會強制分配給『同一個 Worker』依序執行
@pytest.mark.xdist_group(name="payment_group")
def test_payment_step_1():
    ...

@pytest.mark.xdist_group(name="payment_group")
def test_payment_step_2():
    ...

技巧 4:單次 Setup 的全域互斥鎖(filelock 跨 Worker 同步)

有時候我們希望「整個測試套件只需執行一次 Session-level Setup」(例如:去遠端 DB 載入基礎城市選單列表),但平行執行時 4 個 Worker 會同時觸發 4 次。

我們可以使用 filelock 實作跨進程互斥鎖:

Python

# tests/conftest.py
import pytest
from filelock import FileLock

@pytest.fixture(scope="session", autouse=True)
def initialize_global_database_once(tmp_path_factory, worker_id):
    """
    跨多個 pytest-xdist Worker 進程,確保高成本的初始化『只會執行一次』!
    """
    if worker_id == "master":
        # 單執行緒模式直接初始化
        _do_global_setup()
        return

    # 多進程模式:利用 filelock 爭奪鎖
    root_tmp_dir = tmp_path_factory.getbasetemp().parent
    fn = root_tmp_dir / "global_setup.lock"

    with FileLock(str(fn) + ".lock"):
        flag_file = root_tmp_dir / "global_setup.done"
        if not flag_file.is_file():
            # 第一個搶到鎖的 Worker 負責執行初始化
            _do_global_setup()
            flag_file.write_text("done")

def _do_global_setup():
    print("\n[Global Setup] 正在初始化遠端測試環境基礎資料 (僅執行一次)...")

三、 第一線 SDET 的平行化檢核清單(Checklist)

在將自動化專案正式啟用平行執行前,請對照以下 4 個 Self-Check 問題:

  1. 「專案裡是否有任何 global 變數或 class 屬性在儲存 state?」 >> 全部重構成函數區域變數或 pytest Fixture。
  2. 「測試案例是否在共用固定帳號?」>> 全部重構成依據 worker_id / UUID 動態產生獨立帳號。
  3. 「是否有案例會清理全域資料庫表格(如 TRUNCATE TABLE users)?」 >> 嚴禁 TRUNCATE!改為僅刪除當次測試建立的特定列。
  4. 「環境(Staging DB / Server)能否承受併發負載?」 >> 根據伺服器效能合理設定 n 數量(建議為 CPU 核心數的 1~2 倍,勿盲目加大)。

明日預告

解決了平行測試的併發衝突與效能優化後,自動化測試已經能高速、穩定地在 CI/CD 管道中運作。

然而,當測試在 CI 上失敗時,「測試報告能提供多少關鍵資訊?」 決定了 SDET 能在 3 分鐘內定位問題,還是要花 2 小時抓 Log、重新猜測現場。

明天 Day 25,我們將深入研討:《 Day 25|測試失敗後,報告要提供什麼資訊?打造高品質的排查報告 》


上一篇
Day 23|如何縮短自動化測試執行時間:平行執行與極速優化實戰
下一篇
Day 25|測試失敗後,報告要提供什麼資訊?打造高品質的排查報告
系列文
從脆弱腳本到可信任測試平台:自動化測試架構30天25
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言