272 lines
11 KiB
Python
272 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
KINTEX 자동전시시스템 — 스모크/회귀 테스트 러너
|
|
(GUARDiA `scripts/check/run_full_test.py` 패턴 차용 — kintex 백엔드 8021 전용)
|
|
|
|
무엇을 하나:
|
|
1) GET /health 가 200 + status:UP 를 반환하는지(배포 검증 게이트)
|
|
2) 주요 라우터가 등록되어 있는지(200 또는 401/403 = "라우터 존재·인증 필요" = 정상)
|
|
3) (선택) 테스트 계정이 주어지면 로그인 시도 — 1차 로그인 응답 shape 만 확인
|
|
(2차 인증 OTP 는 자동화하지 않는다. verifyToken 존재만 검증)
|
|
|
|
보안:
|
|
- 시크릿 하드코딩 없음. SSH 접속정보·테스트계정은 모두 환경변수에서만 읽는다.
|
|
- 자격증명은 로그·리포트에 출력하지 않는다(마스킹).
|
|
- 실서버 파괴적 요청(생성/삭제)은 보내지 않는다 — 읽기(GET) + 미인증 로그인 프로브만.
|
|
|
|
실행:
|
|
# 로컬(개발 PC)에서 서버로 SSH 후 서버 내부 curl (GUARDiA 러너 방식)
|
|
# 필수 env: KINTEX_SSH_HOST, KINTEX_SSH_PASSWORD (또는 KINTEX_SSH_KEYFILE)
|
|
# 선택 env: KINTEX_SSH_USER(기본 root), KINTEX_BASE(기본 http://127.0.0.1:8021)
|
|
python tools/test/kintex_smoke_test.py
|
|
|
|
# 이미 서버(또는 같은 호스트)에서 실행 — SSH 없이 로컬 curl
|
|
python tools/test/kintex_smoke_test.py --local
|
|
|
|
# 원격 HTTP 직접(도메인 경유, 서버 접속 불가 시)
|
|
KINTEX_BASE=https://kintex.zioinfo.co.kr python tools/test/kintex_smoke_test.py --http
|
|
|
|
결과:
|
|
tools/test/_results/latest.json + YYYYmmdd-HHMMSS.json
|
|
종료코드 0 = 전체 통과, 1 = 실패 있음(CI 게이트용)
|
|
"""
|
|
import os
|
|
import sys
|
|
import json
|
|
import time
|
|
import shlex
|
|
import subprocess
|
|
import pathlib
|
|
from datetime import datetime
|
|
|
|
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
|
|
|
BASE = os.environ.get("KINTEX_BASE", "http://127.0.0.1:8021")
|
|
MODE = "ssh"
|
|
if "--local" in sys.argv:
|
|
MODE = "local"
|
|
elif "--http" in sys.argv:
|
|
MODE = "http"
|
|
|
|
# ── SSH 연결(원격 서버 내부 curl) — GUARDiA 러너 패턴, 단 시크릿은 env only ──────────
|
|
_ssh = None
|
|
if MODE == "ssh":
|
|
host = os.environ.get("KINTEX_SSH_HOST")
|
|
if not host:
|
|
print("❌ KINTEX_SSH_HOST 미설정. --local(같은 호스트) 또는 --http(도메인) 사용, "
|
|
"혹은 KINTEX_SSH_HOST/KINTEX_SSH_PASSWORD env 설정.")
|
|
sys.exit(2)
|
|
user = os.environ.get("KINTEX_SSH_USER", "root")
|
|
pw = os.environ.get("KINTEX_SSH_PASSWORD")
|
|
keyfile = os.environ.get("KINTEX_SSH_KEYFILE")
|
|
try:
|
|
import paramiko
|
|
except ImportError:
|
|
print("❌ paramiko 미설치: pip install paramiko (또는 --local/--http 모드)")
|
|
sys.exit(2)
|
|
_ssh = paramiko.SSHClient()
|
|
_ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
kwargs = dict(hostname=host, username=user, timeout=20)
|
|
if keyfile:
|
|
kwargs["key_filename"] = keyfile
|
|
elif pw:
|
|
kwargs["password"] = pw
|
|
else:
|
|
print("❌ KINTEX_SSH_PASSWORD 또는 KINTEX_SSH_KEYFILE 필요(값은 로그에 노출되지 않음).")
|
|
sys.exit(2)
|
|
_ssh.connect(**kwargs) # 자격증명은 여기서만 사용, 출력 안 함
|
|
|
|
|
|
def run(cmd, timeout=15):
|
|
"""서버(또는 로컬)에서 shell 명령 실행 → stdout 문자열."""
|
|
if MODE == "ssh":
|
|
_, o, _ = _ssh.exec_command(cmd, timeout=timeout)
|
|
return o.read().decode("utf-8", "replace").strip()
|
|
# local / http 모드는 로컬 셸에서 curl 실행
|
|
r = subprocess.run(["bash", "-c", cmd], capture_output=True, timeout=timeout + 5)
|
|
return (r.stdout + r.stderr).decode("utf-8", "replace").strip()
|
|
|
|
|
|
def http_get(path):
|
|
"""GET path → (body_str, http_code)."""
|
|
url = f"{BASE}{path}"
|
|
out = run(f"curl -s -m 8 -w ' HTTP_%{{http_code}}' {shlex.quote(url)}")
|
|
code = None
|
|
if " HTTP_" in out:
|
|
body, _, tail = out.rpartition(" HTTP_")
|
|
out = body.strip()
|
|
try:
|
|
code = int(tail.strip())
|
|
except ValueError:
|
|
pass
|
|
return out, code
|
|
|
|
|
|
def http_post(path, body_json):
|
|
url = f"{BASE}{path}"
|
|
out = run(
|
|
f"curl -s -m 8 -w ' HTTP_%{{http_code}}' -X POST {shlex.quote(url)} "
|
|
f"-H 'Content-Type: application/json' -d {shlex.quote(body_json)}"
|
|
)
|
|
code = None
|
|
if " HTTP_" in out:
|
|
b, _, tail = out.rpartition(" HTTP_")
|
|
out = b.strip()
|
|
try:
|
|
code = int(tail.strip())
|
|
except ValueError:
|
|
pass
|
|
return out, code
|
|
|
|
|
|
results = {
|
|
"run_id": datetime.now().strftime("%Y%m%d-%H%M%S"),
|
|
"timestamp": datetime.now().isoformat(),
|
|
"base": BASE, "mode": MODE,
|
|
"groups": {}, "failures": [], "total": 0, "passed": 0,
|
|
}
|
|
|
|
|
|
def group(name, checks):
|
|
"""checks: [(label, fn)->(ok:bool, detail:str)]"""
|
|
g = {"passed": 0, "failed": 0, "tests": []}
|
|
for label, fn in checks:
|
|
results["total"] += 1
|
|
try:
|
|
ok, detail = fn()
|
|
except Exception as e: # noqa: BLE001
|
|
ok, detail = False, f"exc: {type(e).__name__}"
|
|
if ok:
|
|
g["passed"] += 1
|
|
results["passed"] += 1
|
|
g["tests"].append({"label": label, "status": "PASS"})
|
|
else:
|
|
g["failed"] += 1
|
|
g["tests"].append({"label": label, "status": "FAIL", "detail": detail[:80]})
|
|
results["failures"].append({"group": name, "test": label, "detail": detail[:80]})
|
|
results["groups"][name] = g
|
|
icon = "✅" if g["failed"] == 0 else "⚠️"
|
|
print(f" {icon} {name:<26} {g['passed']}/{g['passed'] + g['failed']}")
|
|
for t in g["tests"]:
|
|
if t["status"] == "FAIL":
|
|
print(f" ❌ {t['label']}: {t.get('detail', '')}")
|
|
|
|
|
|
# ── 체크 헬퍼 ─────────────────────────────────────────────────────────────────
|
|
def check_health():
|
|
body, code = http_get("/health")
|
|
if code != 200:
|
|
return False, f"HTTP {code}"
|
|
try:
|
|
j = json.loads(body)
|
|
except Exception:
|
|
return False, f"non-json: {body[:40]}"
|
|
data = j.get("data", {}) if isinstance(j, dict) else {}
|
|
if data.get("status") == "UP" and data.get("service") == "kintex-backend":
|
|
return True, "UP"
|
|
return False, f"status={data.get('status')}"
|
|
|
|
|
|
def check_registered(path, method="GET", body=None):
|
|
"""라우터 등록 검증: 200/201 = OK, 401/403 = 존재·인증필요(정상), 404/501 = 미등록/미구현(실패)."""
|
|
def _fn():
|
|
if method == "GET":
|
|
resp, code = http_get(path)
|
|
else:
|
|
resp, code = http_post(path, body or "{}")
|
|
if code in (200, 201, 202):
|
|
return True, f"{code}"
|
|
if code in (400, 401, 403, 422):
|
|
# 라우터는 존재하되 인증/유효성으로 거절 = 등록 확인
|
|
return True, f"{code} (registered)"
|
|
if code == 404:
|
|
return False, "404 not registered"
|
|
if code == 501:
|
|
return False, "501 not implemented"
|
|
if code is None:
|
|
return False, "no response (down?)"
|
|
return False, f"HTTP {code}"
|
|
return _fn
|
|
|
|
|
|
def check_login_probe():
|
|
"""테스트 계정(env)이 있으면 1차 로그인 shape 확인. 없으면 미인증 프로브(400/401=라우터 정상)."""
|
|
email = os.environ.get("KINTEX_TEST_EMAIL")
|
|
pw = os.environ.get("KINTEX_TEST_PASSWORD")
|
|
if email and pw:
|
|
body = json.dumps({"email": email, "password": pw})
|
|
resp, code = http_post("/api/auth/login", body)
|
|
if code in (200, 401): # 200=성공(또는 OTP단계), 401=자격오류지만 라우터 정상
|
|
return True, f"{code}"
|
|
return False, f"HTTP {code}"
|
|
# 자격증명 없음 — 유효성 거절(400/401/422)로 라우터 등록만 확인 (계정 노출 없음)
|
|
resp, code = http_post("/api/auth/login", json.dumps({"email": "probe@example.com", "password": "x"}))
|
|
if code in (400, 401, 422):
|
|
return True, f"{code} (registered, no creds)"
|
|
if code == 200:
|
|
return True, "200"
|
|
return False, f"HTTP {code}"
|
|
|
|
|
|
# ── 실행 ─────────────────────────────────────────────────────────────────────
|
|
print("\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
|
|
print(f" KINTEX 스모크 테스트 [{datetime.now():%Y-%m-%d %H:%M:%S}] ({MODE} → {BASE})")
|
|
print("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n")
|
|
start = time.time()
|
|
|
|
print("[배포 게이트]")
|
|
group("health", [("GET /health", check_health)])
|
|
|
|
print("\n[인증]")
|
|
group("auth", [
|
|
("POST /api/auth/login (프로브)", check_login_probe),
|
|
("GET /api/auth/workspaces", check_registered("/api/auth/workspaces")),
|
|
("GET /api/auth/me", check_registered("/api/auth/me")),
|
|
])
|
|
|
|
print("\n[공통·시스템 관리]")
|
|
group("system", [
|
|
("GET /api/common/menus", check_registered("/api/common/menus")),
|
|
("GET /api/common/code-groups", check_registered("/api/common/code-groups")),
|
|
("GET /api/admin/menus", check_registered("/api/admin/menus")),
|
|
])
|
|
|
|
print("\n[공통 업무 레이어(WISE)]")
|
|
group("work", [
|
|
("GET /api/work/notices", check_registered("/api/work/notices")),
|
|
("GET /api/work/stats/worklog", check_registered("/api/work/stats/worklog")),
|
|
("GET /api/work/notifications/unread-count",
|
|
check_registered("/api/work/notifications/unread-count")),
|
|
])
|
|
|
|
print("\n[전시 코어 모듈(M2~M5) — 라우터 등록]")
|
|
# 행사 스코프 경로: 인증/권한으로 거절되더라도 404/501 이 아니면 등록된 것으로 본다.
|
|
group("exhibition_core", [
|
|
("M2 플로어플랜 layout",
|
|
check_registered("/api/events/1/halls/1/layout")),
|
|
("M3 부스 design",
|
|
check_registered("/api/events/1/booths/1/design")),
|
|
("M4 유틸리티/배선",
|
|
check_registered("/api/events/1/booths/1/utility")),
|
|
])
|
|
|
|
elapsed = round(time.time() - start, 1)
|
|
results["duration_sec"] = elapsed
|
|
fail = results["total"] - results["passed"]
|
|
|
|
print("\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
|
|
print(f" 결과: {results['passed']}/{results['total']} 통과 "
|
|
f"{'✅ 전체 통과' if fail == 0 else f'❌ {fail}개 실패'} ({elapsed}초)")
|
|
print("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
|
|
|
|
out_dir = pathlib.Path(__file__).resolve().parent / "_results"
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
for p in [out_dir / f"{results['run_id']}.json", out_dir / "latest.json"]:
|
|
with open(p, "w", encoding="utf-8") as fp:
|
|
json.dump(results, fp, ensure_ascii=False, indent=2, default=str)
|
|
print(f"\n결과 저장: {out_dir / 'latest.json'}")
|
|
|
|
if _ssh:
|
|
_ssh.close()
|
|
sys.exit(0 if fail == 0 else 1)
|