昨天我們建立了一個 FastAPI 電商應用，今天我們將為它編寫完整的 Locust 測試腳本，模擬真實使用者的購物流程，並且實際走一次測試分析。

測試目標

我們要測試的完整使用者流程：

1. 使用者註冊 - 建立新帳號
2. 使用者登入 - 獲取認證 Token
3. 瀏覽商品 - 查看商品列表和詳情
4. 購物行為 - 添加商品到購物車
5. 完成購買 - 結帳流程

專案結構

day15/
├── README.md
├── locustfile.py           # 主要測試檔案
├── user_behaviors.py       # 使用者行為定義
├── test_data.py           # 測試資料管理
└── requirements.txt       # 依賴套件

實作步驟

1. 安裝測試依賴

# requirements.txt
locust==2.17.0
requests==2.31.0
faker==20.1.0

2. 測試資料管理

# test_data.py
import random
from faker import Faker

fake = Faker('zh_TW')

class TestDataGenerator:
    """測試資料生成器"""
    
    # 預定義測試使用者
    EXISTING_USERS = [
        {"username": "testuser1", "password": "password123"},
        {"username": "testuser2", "password": "password123"},
    ]
    
    # 商品 ID 範圍
    PRODUCT_IDS = [1, 2, 3, 4]
    
    @staticmethod
    def generate_new_user():
        """生成新使用者資料"""
        username = fake.user_name()
        timestamp = str(random.randint(1000, 9999))
        return {
            "username": f"{username}_{timestamp}",
            "email": fake.email(),
            "password": "password123"
        }
    
    @staticmethod
    def get_random_existing_user():
        """獲取隨機現有使用者"""
        return random.choice(TestDataGenerator.EXISTING_USERS)
    
    @staticmethod
    def get_random_product_id():
        """獲取隨機商品 ID"""
        return random.choice(TestDataGenerator.PRODUCT_IDS)
    
    @staticmethod
    def get_random_quantity():
        """獲取隨機商品數量"""
        return random.randint(1, 5)

3. 使用者行為模式定義

# user_behaviors.py
import json
import random
from locust import HttpUser, task, between
from test_data import TestDataGenerator

class ShopUser(HttpUser):
    """電商使用者基礎類別"""
    wait_time = between(1, 3)
    host = "http://localhost:8000"
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.auth_token = None
        self.user_data = None
        self.cart_items = []
    
    def on_start(self):
        """使用者開始測試時的初始化"""
        self.user_data = TestDataGenerator.generate_new_user()
        self.register_and_login()
    
    def register_and_login(self):
        """註冊並登入流程"""
        # 註冊新使用者
        response = self.client.post("/register", json=self.user_data)
        
        if response.status_code == 200:
            # 註冊成功，立即登入
            login_data = {
                "username": self.user_data["username"],
                "password": self.user_data["password"]
            }
            
            with self.client.post("/login", json=login_data, catch_response=True) as response:
                if response.status_code == 200:
                    token_data = response.json()
                    self.auth_token = token_data["access_token"]
                    response.success()
                else:
                    response.failure(f"登入失敗: {response.status_code}")
    
    def get_auth_headers(self):
        """獲取認證標頭"""
        if self.auth_token:
            return {"Authorization": f"Bearer {self.auth_token}"}
        return {}
    
    @task(5)
    def browse_products(self):
        """瀏覽商品列表"""
        with self.client.get("/products", catch_response=True) as response:
            if response.status_code == 200:
                products = response.json()
                response.success()
                
                # 隨機查看商品詳情
                if products and random.random() < 0.3:  # 30% 機率
                    product_id = random.choice(products)["id"]
                    self.view_product_detail(product_id)
            else:
                response.failure(f"無法載入商品列表: {response.status_code}")
    
    def view_product_detail(self, product_id):
        """查看商品詳情"""
        with self.client.get(f"/products/{product_id}", catch_response=True) as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"無法載入商品詳情: {response.status_code}")
    
    @task(3)
    def add_to_cart(self):
        """添加商品到購物車"""
        if not self.auth_token:
            return
        
        product_id = TestDataGenerator.get_random_product_id()
        quantity = TestDataGenerator.get_random_quantity()
        
        cart_item = {
            "product_id": product_id,
            "quantity": quantity
        }
        
        with self.client.post(
            "/cart/items",
            json=cart_item,
            headers=self.get_auth_headers(),
            catch_response=True
        ) as response:
            if response.status_code == 200:
                self.cart_items.append(cart_item)
                response.success()
            else:
                response.failure(f"添加商品到購物車失敗: {response.status_code}")
    
    @task(2)
    def view_cart(self):
        """查看購物車"""
        if not self.auth_token:
            return
        
        with self.client.get(
            "/cart",
            headers=self.get_auth_headers(),
            catch_response=True
        ) as response:
            if response.status_code == 200:
                cart_data = response.json()
                response.success()
            else:
                response.failure(f"無法查看購物車: {response.status_code}")
    
    @task(1)
    def checkout(self):
        """結帳流程"""
        if not self.auth_token or len(self.cart_items) == 0:
            return
        
        with self.client.post(
            "/checkout",
            headers=self.get_auth_headers(),
            catch_response=True
        ) as response:
            if response.status_code == 200:
                order_data = response.json()
                response.success()
                
                # 結帳成功，清空購物車記錄
                self.cart_items = []
            else:
                response.failure(f"結帳失敗: {response.status_code}")

class BrowsingUser(HttpUser):
    """專門瀏覽商品的使用者（不需登入）"""
    wait_time = between(0.5, 2)
    host = "http://localhost:8000"
    weight = 3
    
    @task(1)
    def browse_products_only(self):
        """只瀏覽商品，不進行購買"""
        # 瀏覽商品列表
        self.client.get("/products")
        
        # 30% 機率查看商品詳情
        if random.random() < 0.3:
            product_id = TestDataGenerator.get_random_product_id()
            self.client.get(f"/products/{product_id}")

class RegisterOnlyUser(HttpUser):
    """只註冊不購買的使用者"""
    wait_time = between(2, 5)
    host = "http://localhost:8000"
    weight = 1
    
    @task(1)
    def register_and_browse(self):
        """註冊後簡單瀏覽"""
        user_data = TestDataGenerator.generate_new_user()
        
        # 註冊
        response = self.client.post("/register", json=user_data)
        
        if response.status_code == 200:
            # 註冊成功後瀏覽一下商品
            self.client.get("/products")

4. 主測試檔案

# locustfile.py
from locust import HttpUser, task, between, events
from user_behaviors import ShopUser, BrowsingUser, RegisterOnlyUser
import logging

# 設定日誌
logging.basicConfig(level=logging.INFO)

@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    """測試開始時的初始化"""
    logging.info("開始電商負載測試")
    logging.info(f"目標主機: {environment.host}")

@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
    """測試結束時的統計"""
    logging.info("電商負載測試結束")
    
    stats = environment.stats
    total_requests = stats.total.num_requests
    total_failures = stats.total.num_failures
    
    logging.info(f"總請求數: {total_requests}")
    logging.info(f"失敗請求數: {total_failures}")
    
    if total_requests > 0:
        success_rate = (total_requests - total_failures) / total_requests * 100
        logging.info(f"成功率: {success_rate:.2f}%")

# 設定不同類型使用者的權重
class CompleteShoppingUser(ShopUser):
    """完整購物流程使用者"""
    weight = 5

class QuickBrowsingUser(BrowsingUser):
    """快速瀏覽使用者"""
    weight = 3

class NewRegisterUser(RegisterOnlyUser):
    """新註冊使用者"""
    weight = 1

5. 負載測試場景

# test_scenarios.py
from locust import LoadTestShape

class StepLoadPattern(LoadTestShape):
    """階段式負載模式"""
    
    step_time = 60  # 每階段 60 秒
    step_load = 10  # 每階段增加 10 個使用者
    spawn_rate = 2  # 每秒生成 2 個使用者
    time_limit = 600  # 總測試時間 10 分鐘
    
    def tick(self):
        run_time = self.get_run_time()
        
        if run_time > self.time_limit:
            return None
        
        current_step = run_time // self.step_time + 1
        users = min(current_step * self.step_load, 50)  # 最大 50 個使用者
        
        return (users, self.spawn_rate)

執行測試

1. 準備環境

# 啟動 FastAPI 應用（在 day14 目錄）
cd ../day14
pip install fastapi uvicorn
uvicorn app.main:app --reload &

# 回到測試目錄
cd ../day15
pip install -r requirements.txt

2. 執行 Locust 測試

# 基本測試 - Web UI 模式
locust -f locustfile.py --host http://localhost:8000

# 命令列模式 - 30 個使用者，持續 3 分鐘
locust -f locustfile.py --host http://localhost:8000 \
  --users 30 --spawn-rate 5 --run-time 3m --headless

# 生成 HTML 報告
locust -f locustfile.py --host http://localhost:8000 \
  --users 20 --spawn-rate 3 --run-time 2m --headless \
  --html report.html --csv results

3. 測試結果分析

關注以下指標：

效能指標：

• 平均回應時間
• 95% 回應時間
• 每秒請求數 (RPS)
• 錯誤率

業務指標：

• 註冊成功率
• 登入成功率
• 購物轉換率
• 結帳成功率

進階測試配置

1. 不同測試場景

# peak_traffic.py - 尖峰流量測試
class PeakTrafficUser(HttpUser):
    wait_time = between(0.1, 0.5)  # 更短等待時間
    weight = 10

# mobile_user.py - 行動裝置使用者測試  
class MobileUser(HttpUser):
    wait_time = between(2, 8)  # 較長思考時間
    
    def on_start(self):
        self.client.headers.update({
            "User-Agent": "Mobile App/1.0"
        })

2. 分散式測試

# 啟動 Master 節點
locust -f locustfile.py --master --host http://localhost:8000

# 啟動 Worker 節點
locust -f locustfile.py --worker --master-host localhost

3. 配置檔測試

# locust_config.yml
host: "http://localhost:8000"
users: 50
spawn-rate: 5
run-time: "5m"
headless: true
html: "shop_test_report.html"
csv: "shop_test_results"

# 使用配置檔執行
locust -f locustfile.py --config locust_config.yml

監控與除錯

1. 即時監控

在測試過程中：

• 觀察 Locust Web UI 的即時圖表
• 檢查 FastAPI 應用的日誌輸出
• 監控系統資源使用情況

2. 常見問題處理

連線問題：

# 確認 FastAPI 應用正在運行
curl http://localhost:8000/products

# 檢查防火牆設定
netstat -tlnp | grep 8000

認證問題：

• 檢查 JWT Token 格式
• 確認 API 端點的認證要求

實際執行結果

注意: 本節將記錄實際執行測試的結果，包含效能數據、截圖和分析。

測試環境

• 測試機器：MacBook Air (M2, 2022)
• 作業系統：macOS Darwin 24.6.0 (ARM64)
• Python 版本：Python 3.10.18
• FastAPI 版本：0.104.1
• Locust 版本：2.38.0
• 測試時間：2025-08-21 17:46-17:49

測試配置

# 主要測試參數
使用者數量: 20 個並發使用者
增長速率: 2 users/second
測試時間: 2 分鐘
目標主機: http://localhost:8000

# 使用者類型權重配置
- CompleteShoppingUser: 權重 5 (完整購物流程)
- QuickBrowsingUser: 權重 3 (純瀏覽商品)
- ExistingUserShoppingUser: 權重 2 (既有用戶購物)
- NewRegisterUser: 權重 1 (新用戶註冊)

效能測試結果

1. 整體效能指標

2. 各 API 端點效能

3. 業務指標分析

測試截圖

Locust 測試結果

• [x] 測試已完成，生成了詳細的統計報告
• [x] HTML 報告已產生：final_report.html
• [x] CSV 結果文件：test_results_stats.csv, test_results_failures.csv

FastAPI 應用運行狀況

• [x] API 服務運行正常，響應良好
• [x] 所有端點都可正常訪問
• [x] 資料庫操作正常運行

效能分析

1. 瓶頸識別

發現的效能瓶頸：

• 庫存管理問題：購物車和結帳功能失敗率高達 41%，主要因為商品庫存不足
• 認證處理較慢：註冊和登入平均需要 230-240ms，因為密碼雜湊和 JWT 產生較耗時
• 併發庫存更新：多使用者同時購買時容易造成庫存衝突

改善建議：

• 庫存管理優化：增加商品初始庫存量，或實作庫存預留機制
• 認證效能優化：考慮降低 bcrypt 雜湊複雜度或使用快取機制
• 資料庫優化：加入適當的索引和併發控制
• 錯誤處理：改善庫存不足時的使用者體驗

2. 資源使用情況

3. 錯誤分析

常見錯誤類型：

1. 庫存不足錯誤 (43 次)：添加商品到購物車失敗: 400 - {"detail":"庫存不足"}
2. 購物車添加失敗 (77 次)：HTTPError('400 Client Error: Bad Request for url: /cart/items')
3. 結帳失敗 (30 次)：涉及 MacBook Pro、iPad Air、iMac 庫存不足

錯誤原因分析：

• 高併發購買：多使用者同時搶購熱門商品導致庫存快速耗盡
• 庫存設定過低：初始庫存無法應付測試負載
• 缺乏庫存預留機制：沒有為購物車商品預留庫存

解決方案：

• 增加初始庫存：將熱門商品庫存設定為更高數值
• 實作庫存預留：加入購物車時暫時預留庫存 15 分鐘
• 優化併發處理：使用資料庫鎖定或樂觀鎖避免超賣

測試結論

1. 系統穩定性

• 整體穩定：系統在 20 並發使用者下運行穩定，無當機或嚴重錯誤
• 錯誤可控：7.18% 的失敗率主要來自業務邏輯（庫存不足），非系統故障
• 響應一致：各端點響應時間穩定，沒有明顯的效能劣化

2. 效能表現

• 優異的讀取效能：商品瀏覽相關 API 響應時間 3-5ms，非常出色
• 認證效能可接受：註冊/登入 230-240ms 的時間在安全考量下屬合理範圍
• 寫入操作效能良好：購物車和結帳操作響應時間 4-6ms，效能優秀

3. 擴展性評估

• 水平擴展潛力：FastAPI 架構支援良好的水平擴展
• 資料庫瓶頸：SQLite 在高併發下可能成為瓶頸，建議升級至 PostgreSQL
• 緩存機會：商品資料適合加入 Redis 緩存提升效能

4. 改進建議

1. 庫存管理系統優化：實作分散式庫存管理和預留機制
2. 資料庫升級：從 SQLite 遷移至 PostgreSQL 支援更高併發
3. 加入緩存層：使用 Redis 緩存熱門商品和使用者工作階段
4. 監控告警：建立效能監控和庫存告警機制
5. 負載平衡：準備多實例部署和負載平衡配置

總結

今天我們為 FastAPI 電商應用建立了完整的 Locust 測試方案：

1. 多樣化使用者模式：完整購物、純瀏覽、僅註冊
2. 真實使用者行為：註冊→登入→瀏覽→購買→結帳
3. 靈活的測試配置：支援不同負載模式和測試場景
4. 詳細的效能分析：業務相關的效能指標追蹤
5. 分散式測試支援：大規模負載測試能力

透過這套測試系統，我們可以：

• 驗證應用在不同負載下的效能表現
• 識別系統的效能瓶頸
• 確保關鍵業務流程的穩定性
• 為系統優化提供數據支持