API 自動化測試與效能優化

from fastapi.testclient import TestClient
from optimize_description import app

# 初始化 FastAPI 測試客戶端
client = TestClient(app)
URL = "/summarized-description"

def test_summarize_description():
    """ 測試 API 是否能正確處理商品描述優化 """
    # 測試請求的 JSON 內容
    payload = {"description": "這款智能手錶具備心率監測與 GPS 追蹤，適合運動健身。"}
    
    # 發送 POST 請求到 API
    response = client.post(URL, json=payload)

    # 確保 API 回應狀態碼為 200（成功）
    assert response.status_code == 200
    
    # 解析回應內容為 JSON
    data = response.json()

    # 確保 API 回應中包含 `summarized_description` 欄位
    assert "summarized_description" in data  

    # 確保 `summarized_description` 內部包含必要欄位
    assert "title" in data["summarized_description"]  # 標題
    assert "summary" in data["summarized_description"]  # 簡要說明
    assert "bullet_points" in data["summarized_description"]  # 商品重點條列

    # 確保 `bullet_points` 為一個列表
    assert isinstance(data["summarized_description"]["bullet_points"], list)

def test_invalid_json():
    """ 測試 API 是否能正確處理非 JSON 格式的請求 """
    response = client.post(URL, content="invalid json")  # 直接傳遞非 JSON 字串

    # FastAPI 內建的 422 Unprocessable Entity 錯誤碼表示請求格式錯誤
    assert response.status_code == 422

def test_empty_description():
    """ 測試當輸入為空白時，API 是否能正確處理 """
    payload = {"description": ""}
    response = client.post(URL, json=payload)

    # API 應該回傳 400 Bad Request，因為 description 不能為空
    assert response.status_code == 400

    # 確保 API 回應的錯誤訊息正確
    data = response.json()
    assert data["detail"] == "❌ 商品描述不可為空"

import pytest
import time
from concurrent.futures import ThreadPoolExecutor
from fastapi.testclient import TestClient
from optimize_description import app

client = TestClient(app)  # 初始化 FastAPI 測試客戶端
URL = "/summarized-description"

@pytest.mark.benchmark
def test_async_api_performance(benchmark):
    """ 測試 API 在單一請求與多併發請求下的效能 """

    payload = {"description": "這款智能手錶具備心率監測與 GPS 追蹤，適合運動健身。"}

    def send_request():
        """ 發送 API 請求並計算執行時間 """
        start_time = time.perf_counter()  # 記錄請求開始時間
        response = client.post(URL, json=payload)  # 發送 API 請求
        end_time = time.perf_counter()  # 記錄請求結束時間

        assert response.status_code == 200  # 確保 API 回應正常
        return end_time - start_time  # 計算請求耗時

    avg_time = 0  # 初始化平均執行時間變數

    try:
        # **測試單一請求的處理時間**
        execution_time = benchmark(send_request)  # 使用 pytest-benchmark 測量單次請求時間

        # **測試多併發請求的效能**
        with ThreadPoolExecutor(max_workers=10) as executor:  # 使用 10 個執行緒模擬 30 個請求
            execution_times = list(executor.map(lambda _: send_request(), range(30)))

        # **計算 30 次請求的平均執行時間**
        avg_time = sum(execution_times) / len(execution_times)

    finally:
        # **將測試結果寫入紀錄檔**
        with open("benchmark.log", "a", encoding="utf-8") as f:
            f.write(f"30 個並發請求的平均執行時間: {avg_time:.4f} 秒\n")

    print(f"🚀 30 個並發請求的平均執行時間: {avg_time:.4f} 秒")

execution_time = benchmark(send_request)

with ThreadPoolExecutor(max_workers=10) as executor:
    execution_times = list(executor.map(lambda _: send_request(), range(30)))

with open("benchmark.log", "a", encoding="utf-8") as f:
    f.write(f"30 個並發請求的平均執行時間: {avg_time:.4f} 秒\n")

from aiocache import cached
from aiocache.serializers import JsonSerializer

@cached(ttl=3600, key_builder=lambda func, *args, **kwargs: f"{args[0]}", serializer=JsonSerializer())
async def optimize_description(text: str) -> str:
    ...

import asyncio
def test_cache_mechanism(monkeypatch):
    asyncio.run(optimize_description.cache.clear())

    call_count = {"calls": 0}

    class FakeChatCompletions:
        def create(self, *args, **kwargs):
            call_count["calls"] += 1
            return {"choices": [{"message": {"content": '{"summarized_description": {"title": "Test", "summary": "Summary", "bullet_points": ["A", "B"]}}'}}]}

    class FakeOpenAI:
        def __init__(self, api_key):
            self.api_key = api_key
            self.chat = FakeChatCompletions()

    monkeypatch.setattr(openai, "OpenAI", FakeOpenAI)

    payload = {"description": "測試快取"}
    
    response1 = client.post(URL, json=payload)
    assert response1.status_code == 200

    response2 = client.post(URL, json=payload)
    assert response2.status_code == 200

    assert call_count["calls"] == 1  # API 只應該被呼叫一次

response = client.chat.completions.create(  
    model="gpt-4o",
    response_format={"type": "json_object"},
    messages=[{"role": "system", "content": prompt}],
    max_tokens=500  # 限制回應長度
)