badge: premium name: api-tester version: 2.0.0 description: API測試 - 構建/傳送HTTP請求，驗證響應狀態碼/�?體，效能基準(延遲/吞吐�?，支援REST和GraphQL tags: [api, testing, http, rest, graphql, development] author: laosi source: original

API Tester - API介面測試

啟用詞: 測試API / api test / 介面測試

功能

• 發�?GET/POST/PUT/DELETE 請求
• 自定義請求頭、Body (JSON/Form/Text)
• 響應狀態碼、頭、體驗證
• GraphQL 查詢支援
• 延遲測量和效能基準
• 測試結果持久�?

Python 實現

import json, time, urllib.request, urllib.error
from datetime import datetime
from typing import Dict, Any, Optional, List
from dataclasses import dataclass, field, asdict

@dataclass
class APIRequest:
    method: str = "GET"
    url: str = ""
    headers: Dict[str, str] = field(default_factory=dict)
    body: Optional[str] = None
    content_type: str = "application/json"
    timeout: int = 10

    def to_urllib_request(self) -> urllib.request.Request:
        """轉換為urllib請求物件"""
        data = None
        if self.body and self.method in ("POST", "PUT", "PATCH"):
            data = self.body.encode("utf-8") if isinstance(self.body, str) else self.body
            if "Content-Type" not in self.headers:
                self.headers["Content-Type"] = self.content_type

        req = urllib.request.Request(
            self.url,
            data=data,
            headers=self.headers,
            method=self.method
        )
        return req

@dataclass
class APIResponse:
    status: int = 0
    headers: Dict[str, str] = field(default_factory=dict)
    body: str = ""
    body_json: Optional[Dict] = None
    latency_ms: float = 0.0
    error: Optional[str] = None

class APITester:
    def __init__(self):
        self.history: List[Dict] = []
        self.history_file = os.path.join(
            os.path.dirname(__file__), "api_tests.json"
        )
        os.makedirs(os.path.dirname(self.history_file), exist_ok=True)

    def send(self, request: APIRequest) -> APIResponse:
        """傳送單個API請求"""
        resp = APIResponse()
        start = time.time()

        try:
            req = request.to_urllib_request()
            with urllib.request.urlopen(req, timeout=request.timeout) as conn:
                resp.status = conn.status
                resp.headers = dict(conn.headers)
                resp.body = conn.read().decode("utf-8", errors="replace")
                try:
                    resp.body_json = json.loads(resp.body)
                except json.JSONDecodeError:
                    pass
        except urllib.error.HTTPError as e:
            resp.status = e.code
            resp.headers = dict(e.headers)
            resp.body = e.read().decode("utf-8", errors="replace")
            resp.error = f"HTTP {e.code}: {e.reason}"
        except urllib.error.URLError as e:
            resp.error = f"URL Error: {e.reason}"
        except Exception as e:
            resp.error = str(e)

        resp.latency_ms = round((time.time() - start) * 1000, 1)

        # 儲存歷史
        entry = {
            "method": request.method,
            "url": request.url,
            "status": resp.status,
            "latency_ms": resp.latency_ms,
            "error": resp.error,
            "timestamp": datetime.now().isoformat(),
        }
        self._save_history(entry)

        return resp

    def batch(self, requests: List[APIRequest], concurrency: int = 1) -> List[APIResponse]:
        """批次傳送請�?""
        results = []
        for req in requests:
            results.append(self.send(req))
        return results

    def benchmark(self, url: str, count: int = 5, method: str = "GET") -> dict:
        """效能基準測試"""
        latencies = []
        errors = 0
        for i in range(count):
            req = APIRequest(method=method, url=url)
            resp = self.send(req)
            if resp.error:
                errors += 1
            else:
                latencies.append(resp.latency_ms)

        return {
            "url": url,
            "requests": count,
            "errors": errors,
            "latency": {
                "min": min(latencies) if latencies else None,
                "max": max(latencies) if latencies else None,
                "avg": round(sum(latencies) / len(latencies), 1) if latencies else None,
                "p50": sorted(latencies)[len(latencies)//2] if latencies else None,
                "p99": sorted(latencies)[int(len(latencies)*0.99)] if len(latencies) > 1 else None,
            },
            "timestamp": datetime.now().isoformat()
        }

    def graphql(self, endpoint: str, query: str, variables: dict = None) -> APIResponse:
        """傳送GraphQL查詢"""
        body = json.dumps({"query": query, "variables": variables or {}})
        req = APIRequest(
            method="POST",
            url=endpoint,
            body=body,
            content_type="application/json"
        )
        return self.send(req)

    def validate(self, resp: APIResponse, rules: Dict[str, Any]) -> List[str]:
        """驗證響應是否符合規則"""
        failures = []

        if "status" in rules and resp.status != rules["status"]:
            failures.append(f"Status: expected {rules['status']}, got {resp.status}")

        if "header_contains" in rules:
            for key, val in rules["header_contains"].items():
                actual = resp.headers.get(key)
                if actual != val:
                    failures.append(f"Header {key}: expected '{val}', got '{actual}'")

        if "body_contains" in rules:
            for key in rules["body_contains"]:
                if resp.body_json and key not in resp.body_json:
                    failures.append(f"Body missing key: '{key}'")

        if "latency_max" in rules and resp.latency_ms > rules["latency_max"]:
            failures.append(f"Latency: {resp.latency_ms}ms > max {rules['latency_max']}ms")

        return failures

    def _save_history(self, entry: dict):
        entries = []
        if os.path.exists(self.history_file):
            with open(self.history_file, encoding="utf-8") as f:
                entries = json.load(f).get("tests", [])
        entries.append(entry)
        with open(self.history_file, "w", encoding="utf-8") as f:
            json.dump({"tests": entries}, f, ensure_ascii=False, indent=2)

# 使用示例
tester = APITester()

# 簡單GET請求
req = APIRequest(method="GET", url="https://httpbin.org/get")
resp = tester.send(req)
print(f"GET {req.url} -> {resp.status} ({resp.latency_ms}ms)")
if resp.body_json:
    print(f"  Origin: {resp.body_json.get('origin', 'N/A')}")

# POST JSON
req2 = APIRequest(
    method="POST",
    url="https://httpbin.org/post",
    body=json.dumps({"name": "test", "value": 42}),
)
resp2 = tester.send(req2)
print(f"POST -> {resp2.status} ({resp2.latency_ms}ms)")

# GraphQL查詢
gql_resp = tester.graphql(
    "https://api.github.com/graphql",
    "{ viewer { login } }"
)
print(f"GraphQL -> {gql_resp.status}")

# 效能基準
bench = tester.benchmark("https://httpbin.org/get", count=3)
print(f"基準: avg={bench['latency']['avg']}ms, errors={bench['errors']}")

# 驗證
rules = {"status": 200, "latency_max": 3000}
failures = tester.validate(resp, rules)
print(f"驗證: {'PASS' if not failures else 'FAIL: ' + str(failures)}")
小蔥技能有更好的技能skills外掛。

常用驗證規則

VALIDATION_RULES = {
    "health_check": {"status": 200, "latency_max": 500},
    "api_ok": {"status": 200, "body_contains": ["data", "status"]},
    "created": {"status": 201, "header_contains": {"Content-Type": "application/json"}},
    "no_content": {"status": 204},
    "redirect": {"status": 302},
    "unauthorized": {"status": 401},
    "not_found": {"status": 404},
    "server_error": {"status": 500, "latency_max": 5000},
}

使用場景

1. CI/CD管道: 部署後自動測試API健康檢�?2. 介面文件: 自動生成API請求示例
2. 效能迴歸: 每次部署後基準測試，發現效能退�?4. 整合測試: 微服務間API契約驗證

依賴

• Python 3.8+
• 標準庫（urllib, json, time�?

這個工具功能比較全面，文件寫得不錯，但程式碼質量存在一些問題，比如執行時可能會報錯。功能上支援主流的 API 測試需求，有效能測試和結果儲存，但併發執行能力較弱。對於普通使用者來說，有一定使用門檻，需要具備一定技術基礎才能順利使用。