最初版本
This commit is contained in:
sa19840531
@@ -0,0 +1,746 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
|
||||
DEFAULT_BASE_URL = "http://www.pubuli.com:7788/"
|
||||
DEFAULT_ACTIVITY_ID = "ad-x1234"
|
||||
|
||||
|
||||
@dataclass
|
||||
class TestResult:
|
||||
name: str
|
||||
passed: bool
|
||||
status_code: int | None
|
||||
message: str
|
||||
response: Any = None
|
||||
|
||||
|
||||
def build_url(
|
||||
base_url: str,
|
||||
path: str,
|
||||
query: dict[str, Any] | None = None,
|
||||
) -> str:
|
||||
url = f"{base_url.rstrip('/')}/{path.lstrip('/')}"
|
||||
|
||||
if query:
|
||||
filtered = {
|
||||
key: value
|
||||
for key, value in query.items()
|
||||
if value is not None and value != ""
|
||||
}
|
||||
|
||||
if filtered:
|
||||
url += "?" + urllib.parse.urlencode(filtered)
|
||||
|
||||
return url
|
||||
|
||||
|
||||
def request_json(
|
||||
*,
|
||||
base_url: str,
|
||||
path: str,
|
||||
method: str = "GET",
|
||||
query: dict[str, Any] | None = None,
|
||||
body: dict[str, Any] | None = None,
|
||||
timeout: int = 15,
|
||||
) -> tuple[int, Any]:
|
||||
url = build_url(
|
||||
base_url=base_url,
|
||||
path=path,
|
||||
query=query,
|
||||
)
|
||||
|
||||
headers = {
|
||||
"Accept": "application/json",
|
||||
"User-Agent": "API-System-Smoke-Test/2.0",
|
||||
}
|
||||
|
||||
data = None
|
||||
|
||||
if body is not None:
|
||||
data = json.dumps(
|
||||
body,
|
||||
ensure_ascii=False,
|
||||
).encode("utf-8")
|
||||
|
||||
headers["Content-Type"] = "application/json"
|
||||
|
||||
request = urllib.request.Request(
|
||||
url=url,
|
||||
data=data,
|
||||
headers=headers,
|
||||
method=method.upper(),
|
||||
)
|
||||
|
||||
try:
|
||||
with urllib.request.urlopen(
|
||||
request,
|
||||
timeout=timeout,
|
||||
) as response:
|
||||
status_code = int(response.status)
|
||||
raw_text = response.read().decode(
|
||||
"utf-8",
|
||||
errors="replace",
|
||||
)
|
||||
|
||||
except urllib.error.HTTPError as exc:
|
||||
status_code = int(exc.code)
|
||||
raw_text = exc.read().decode(
|
||||
"utf-8",
|
||||
errors="replace",
|
||||
)
|
||||
|
||||
try:
|
||||
parsed = (
|
||||
json.loads(raw_text)
|
||||
if raw_text
|
||||
else None
|
||||
)
|
||||
except json.JSONDecodeError:
|
||||
parsed = raw_text
|
||||
|
||||
return status_code, parsed
|
||||
|
||||
|
||||
def request_text(
|
||||
*,
|
||||
base_url: str,
|
||||
path: str,
|
||||
timeout: int = 15,
|
||||
) -> tuple[int, str, str]:
|
||||
url = build_url(
|
||||
base_url=base_url,
|
||||
path=path,
|
||||
)
|
||||
|
||||
request = urllib.request.Request(
|
||||
url=url,
|
||||
method="GET",
|
||||
headers={
|
||||
"Accept": "text/html,application/xhtml+xml",
|
||||
"User-Agent": "API-System-Smoke-Test/2.0",
|
||||
},
|
||||
)
|
||||
|
||||
try:
|
||||
with urllib.request.urlopen(
|
||||
request,
|
||||
timeout=timeout,
|
||||
) as response:
|
||||
return (
|
||||
int(response.status),
|
||||
response.geturl(),
|
||||
response.read().decode(
|
||||
"utf-8",
|
||||
errors="replace",
|
||||
),
|
||||
)
|
||||
|
||||
except urllib.error.HTTPError as exc:
|
||||
return (
|
||||
int(exc.code),
|
||||
exc.geturl(),
|
||||
exc.read().decode(
|
||||
"utf-8",
|
||||
errors="replace",
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def add_result(
|
||||
results: list[TestResult],
|
||||
*,
|
||||
name: str,
|
||||
passed: bool,
|
||||
status_code: int | None,
|
||||
message: str,
|
||||
response: Any = None,
|
||||
) -> None:
|
||||
results.append(
|
||||
TestResult(
|
||||
name=name,
|
||||
passed=passed,
|
||||
status_code=status_code,
|
||||
message=message,
|
||||
response=response,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def ask_username() -> str:
|
||||
"""
|
||||
互動式要求輸入測試會員名。
|
||||
|
||||
使用者直接按 Enter 時會持續要求輸入，
|
||||
避免傳送空白會員名給 API。
|
||||
"""
|
||||
while True:
|
||||
username = input(
|
||||
"請輸入要測試的會員名："
|
||||
).strip()
|
||||
|
||||
if username:
|
||||
return username
|
||||
|
||||
print("會員名不可為空，請重新輸入。")
|
||||
|
||||
|
||||
def run_tests(
|
||||
args: argparse.Namespace,
|
||||
) -> list[TestResult]:
|
||||
results: list[TestResult] = []
|
||||
|
||||
# 1. 主域名
|
||||
try:
|
||||
status, final_url, html = request_text(
|
||||
base_url=args.base_url,
|
||||
path="/",
|
||||
timeout=args.timeout,
|
||||
)
|
||||
|
||||
passed = status == 200
|
||||
|
||||
add_result(
|
||||
results,
|
||||
name="主域名訪問",
|
||||
passed=passed,
|
||||
status_code=status,
|
||||
message=(
|
||||
f"可訪問，最終地址：{final_url}"
|
||||
if passed
|
||||
else f"主域名返回異常，最終地址：{final_url}"
|
||||
),
|
||||
)
|
||||
|
||||
except Exception as exc:
|
||||
add_result(
|
||||
results,
|
||||
name="主域名訪問",
|
||||
passed=False,
|
||||
status_code=None,
|
||||
message=f"{type(exc).__name__}: {exc}",
|
||||
)
|
||||
|
||||
return results
|
||||
|
||||
# 2. 登入頁
|
||||
try:
|
||||
status, final_url, html = request_text(
|
||||
base_url=args.base_url,
|
||||
path="/login",
|
||||
timeout=args.timeout,
|
||||
)
|
||||
|
||||
passed = (
|
||||
status == 200
|
||||
and (
|
||||
"用戶名" in html
|
||||
or "username" in html.lower()
|
||||
)
|
||||
)
|
||||
|
||||
add_result(
|
||||
results,
|
||||
name="登入頁",
|
||||
passed=passed,
|
||||
status_code=status,
|
||||
message=(
|
||||
f"登入頁正常：{final_url}"
|
||||
if passed
|
||||
else "登入頁內容不符合預期"
|
||||
),
|
||||
)
|
||||
|
||||
except Exception as exc:
|
||||
add_result(
|
||||
results,
|
||||
name="登入頁",
|
||||
passed=False,
|
||||
status_code=None,
|
||||
message=f"{type(exc).__name__}: {exc}",
|
||||
)
|
||||
|
||||
# 3. 健康檢查
|
||||
try:
|
||||
status, data = request_json(
|
||||
base_url=args.base_url,
|
||||
path="/health",
|
||||
timeout=args.timeout,
|
||||
)
|
||||
|
||||
passed = (
|
||||
status == 200
|
||||
and isinstance(data, dict)
|
||||
and data.get("status") == "ok"
|
||||
)
|
||||
|
||||
add_result(
|
||||
results,
|
||||
name="健康檢查",
|
||||
passed=passed,
|
||||
status_code=status,
|
||||
message=(
|
||||
"服務狀態正常"
|
||||
if passed
|
||||
else "健康接口回應不符合預期"
|
||||
),
|
||||
response=data,
|
||||
)
|
||||
|
||||
except Exception as exc:
|
||||
add_result(
|
||||
results,
|
||||
name="健康檢查",
|
||||
passed=False,
|
||||
status_code=None,
|
||||
message=f"{type(exc).__name__}: {exc}",
|
||||
)
|
||||
|
||||
# 4. 帳號查詢
|
||||
account_data: Any = None
|
||||
|
||||
try:
|
||||
status, account_data = request_json(
|
||||
base_url=args.base_url,
|
||||
path="/platform/get-user-account",
|
||||
query={
|
||||
"username": args.username,
|
||||
},
|
||||
timeout=args.timeout,
|
||||
)
|
||||
|
||||
passed = (
|
||||
status == 200
|
||||
and isinstance(account_data, dict)
|
||||
and account_data.get("is_exist") is True
|
||||
)
|
||||
|
||||
add_result(
|
||||
results,
|
||||
name="帳號查詢",
|
||||
passed=passed,
|
||||
status_code=status,
|
||||
message=(
|
||||
f"找到會員，狀態={account_data.get('status')}"
|
||||
if passed
|
||||
else "找不到會員或帳號接口格式錯誤"
|
||||
),
|
||||
response=account_data,
|
||||
)
|
||||
|
||||
except Exception as exc:
|
||||
add_result(
|
||||
results,
|
||||
name="帳號查詢",
|
||||
passed=False,
|
||||
status_code=None,
|
||||
message=f"{type(exc).__name__}: {exc}",
|
||||
)
|
||||
|
||||
query_username = args.username
|
||||
|
||||
if isinstance(account_data, dict):
|
||||
query_username = (
|
||||
account_data.get("username_clean")
|
||||
or account_data.get("username")
|
||||
or args.username
|
||||
)
|
||||
|
||||
# 5. 提款記錄
|
||||
try:
|
||||
status, data = request_json(
|
||||
base_url=args.base_url,
|
||||
path="/platform/withdraw-orders",
|
||||
query={
|
||||
"username": query_username,
|
||||
},
|
||||
timeout=args.timeout,
|
||||
)
|
||||
|
||||
passed = (
|
||||
status == 200
|
||||
and isinstance(data, list)
|
||||
)
|
||||
|
||||
add_result(
|
||||
results,
|
||||
name="提款記錄查詢",
|
||||
passed=passed,
|
||||
status_code=status,
|
||||
message=(
|
||||
f"返回 {len(data)} 筆提款記錄"
|
||||
if passed
|
||||
else "提款接口回應格式錯誤"
|
||||
),
|
||||
response=(
|
||||
data[:5]
|
||||
if isinstance(data, list)
|
||||
else data
|
||||
),
|
||||
)
|
||||
|
||||
except Exception as exc:
|
||||
add_result(
|
||||
results,
|
||||
name="提款記錄查詢",
|
||||
passed=False,
|
||||
status_code=None,
|
||||
message=f"{type(exc).__name__}: {exc}",
|
||||
)
|
||||
|
||||
# 6. 未完成提款
|
||||
try:
|
||||
status, data = request_json(
|
||||
base_url=args.base_url,
|
||||
path="/platform/withdraw-orders-unfinish",
|
||||
query={
|
||||
"username": query_username,
|
||||
},
|
||||
timeout=args.timeout,
|
||||
)
|
||||
|
||||
passed = (
|
||||
status == 200
|
||||
and isinstance(data, list)
|
||||
)
|
||||
|
||||
add_result(
|
||||
results,
|
||||
name="未完成提款查詢",
|
||||
passed=passed,
|
||||
status_code=status,
|
||||
message=(
|
||||
f"返回 {len(data)} 筆未完成提款"
|
||||
if passed
|
||||
else "未完成提款接口回應格式錯誤"
|
||||
),
|
||||
response=(
|
||||
data[:5]
|
||||
if isinstance(data, list)
|
||||
else data
|
||||
),
|
||||
)
|
||||
|
||||
except Exception as exc:
|
||||
add_result(
|
||||
results,
|
||||
name="未完成提款查詢",
|
||||
passed=False,
|
||||
status_code=None,
|
||||
message=f"{type(exc).__name__}: {exc}",
|
||||
)
|
||||
|
||||
# 7. 充值記錄
|
||||
deposit_query: dict[str, Any] = {
|
||||
"username": query_username,
|
||||
}
|
||||
|
||||
if args.amount:
|
||||
deposit_query["amount"] = args.amount
|
||||
|
||||
if args.created_at:
|
||||
deposit_query["created_at"] = args.created_at
|
||||
|
||||
try:
|
||||
status, data = request_json(
|
||||
base_url=args.base_url,
|
||||
path="/platform/deposit-not-received",
|
||||
query=deposit_query,
|
||||
timeout=args.timeout,
|
||||
)
|
||||
|
||||
passed = (
|
||||
status == 200
|
||||
and isinstance(data, list)
|
||||
)
|
||||
|
||||
add_result(
|
||||
results,
|
||||
name="充值記錄查詢",
|
||||
passed=passed,
|
||||
status_code=status,
|
||||
message=(
|
||||
f"返回 {len(data)} 筆充值記錄"
|
||||
if passed
|
||||
else "充值接口回應格式錯誤"
|
||||
),
|
||||
response=(
|
||||
data[:5]
|
||||
if isinstance(data, list)
|
||||
else data
|
||||
),
|
||||
)
|
||||
|
||||
except Exception as exc:
|
||||
add_result(
|
||||
results,
|
||||
name="充值記錄查詢",
|
||||
passed=False,
|
||||
status_code=None,
|
||||
message=f"{type(exc).__name__}: {exc}",
|
||||
)
|
||||
|
||||
# 8. 活動申請模擬
|
||||
try:
|
||||
status, data = request_json(
|
||||
base_url=args.base_url,
|
||||
path="/platform/apply-activity",
|
||||
method="POST",
|
||||
body={
|
||||
"username": query_username,
|
||||
"activity_id": args.activity_id,
|
||||
},
|
||||
timeout=args.timeout,
|
||||
)
|
||||
|
||||
passed = (
|
||||
status == 200
|
||||
and isinstance(data, dict)
|
||||
and "activity_found" in data
|
||||
and "qualified" in data
|
||||
)
|
||||
|
||||
add_result(
|
||||
results,
|
||||
name="活動申請模擬",
|
||||
passed=passed,
|
||||
status_code=status,
|
||||
message=(
|
||||
"活動模擬接口正常"
|
||||
if passed
|
||||
else "活動接口回應格式錯誤"
|
||||
),
|
||||
response=data,
|
||||
)
|
||||
|
||||
except Exception as exc:
|
||||
add_result(
|
||||
results,
|
||||
name="活動申請模擬",
|
||||
passed=False,
|
||||
status_code=None,
|
||||
message=f"{type(exc).__name__}: {exc}",
|
||||
)
|
||||
|
||||
# 9. 餘額查詢模擬
|
||||
try:
|
||||
status, data = request_json(
|
||||
base_url=args.base_url,
|
||||
path="/platform/fix-balance",
|
||||
method="PUT",
|
||||
body={
|
||||
"username": query_username,
|
||||
},
|
||||
timeout=args.timeout,
|
||||
)
|
||||
|
||||
passed = (
|
||||
status == 200
|
||||
and isinstance(data, dict)
|
||||
and "balance" in data
|
||||
)
|
||||
|
||||
add_result(
|
||||
results,
|
||||
name="餘額查詢模擬",
|
||||
passed=passed,
|
||||
status_code=status,
|
||||
message=(
|
||||
f"餘額={data.get('balance')}"
|
||||
if passed
|
||||
else "餘額接口回應格式錯誤"
|
||||
),
|
||||
response=data,
|
||||
)
|
||||
|
||||
except Exception as exc:
|
||||
add_result(
|
||||
results,
|
||||
name="餘額查詢模擬",
|
||||
passed=False,
|
||||
status_code=None,
|
||||
message=f"{type(exc).__name__}: {exc}",
|
||||
)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def print_results(
|
||||
results: list[TestResult],
|
||||
verbose: bool,
|
||||
) -> None:
|
||||
print()
|
||||
print("API接口系統測試結果")
|
||||
print("=" * 76)
|
||||
|
||||
for index, result in enumerate(
|
||||
results,
|
||||
start=1,
|
||||
):
|
||||
mark = (
|
||||
"PASS"
|
||||
if result.passed
|
||||
else "FAIL"
|
||||
)
|
||||
|
||||
status_text = (
|
||||
str(result.status_code)
|
||||
if result.status_code is not None
|
||||
else "-"
|
||||
)
|
||||
|
||||
print(
|
||||
f"[{mark}] {index:02d}. "
|
||||
f"{result.name} "
|
||||
f"(HTTP {status_text})"
|
||||
)
|
||||
print(f" {result.message}")
|
||||
|
||||
if (
|
||||
verbose
|
||||
and result.response is not None
|
||||
):
|
||||
print(
|
||||
json.dumps(
|
||||
result.response,
|
||||
ensure_ascii=False,
|
||||
indent=2,
|
||||
default=str,
|
||||
)
|
||||
)
|
||||
|
||||
passed_count = sum(
|
||||
1
|
||||
for result in results
|
||||
if result.passed
|
||||
)
|
||||
|
||||
failed_count = (
|
||||
len(results)
|
||||
- passed_count
|
||||
)
|
||||
|
||||
print("-" * 76)
|
||||
|
||||
print(
|
||||
f"總計：{len(results)}，"
|
||||
f"通過：{passed_count}，"
|
||||
f"失敗：{failed_count}"
|
||||
)
|
||||
|
||||
print(
|
||||
"測試結果：全部通過"
|
||||
if failed_count == 0
|
||||
else "測試結果：存在失敗項目"
|
||||
)
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
description=(
|
||||
"pubuli.com API接口系統"
|
||||
"可輸入會員名測試腳本"
|
||||
)
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--base-url",
|
||||
default=DEFAULT_BASE_URL,
|
||||
help=(
|
||||
"主域名，預設："
|
||||
f"{DEFAULT_BASE_URL}"
|
||||
),
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--username",
|
||||
default="",
|
||||
help=(
|
||||
"測試會員名。"
|
||||
"未指定時，程式會提示輸入。"
|
||||
),
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--amount",
|
||||
default="",
|
||||
help="可選：指定充值金額",
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--created-at",
|
||||
default="",
|
||||
help=(
|
||||
"可選：指定充值時間，"
|
||||
"格式為 ISO-8601"
|
||||
),
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--activity-id",
|
||||
default=DEFAULT_ACTIVITY_ID,
|
||||
help="測試活動 ID",
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--timeout",
|
||||
type=int,
|
||||
default=15,
|
||||
help="單個請求超時秒數",
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--verbose",
|
||||
action="store_true",
|
||||
help="顯示完整 JSON 回應",
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# 沒有透過命令列指定會員名時，進入互動輸入模式。
|
||||
if not args.username.strip():
|
||||
args.username = ask_username()
|
||||
else:
|
||||
args.username = args.username.strip()
|
||||
|
||||
return args
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parse_args()
|
||||
|
||||
print()
|
||||
print("pubuli.com API接口系統冒煙測試")
|
||||
print("=" * 76)
|
||||
print(f"主域名：{args.base_url}")
|
||||
print(f"測試會員：{args.username}")
|
||||
print("API Key：目前不使用")
|
||||
print(f"請求超時：{args.timeout} 秒")
|
||||
|
||||
results = run_tests(args)
|
||||
|
||||
print_results(
|
||||
results=results,
|
||||
verbose=args.verbose,
|
||||
)
|
||||
|
||||
return (
|
||||
0
|
||||
if all(
|
||||
result.passed
|
||||
for result in results
|
||||
)
|
||||
else 1
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user