747 lines
17 KiB
Python
747 lines
17 KiB
Python
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())
|