diff --git a/tools/qa_runner b/tools/qa_runner new file mode 100644 index 0000000..46265a4 --- /dev/null +++ b/tools/qa_runner @@ -0,0 +1,194 @@ +""" +E2E 테스트 자동 실행기 (팀 JSON 데이터셋 지원) +v3: 신버전 API 대응 + - 인증(로그인) 자동 처리 + - 응답에서 reasons 필드가 사라진 것에 맞춰, reason_code 비교를 제외하고 + action(ALLOW/MASK/WARN/BLOCK) 일치 여부만으로 PASS/FAIL 판정 + - 계좌번호(PII_ACCOUNT_DETECTED) 단독 노출은 기대값을 WARN으로 변경 + (실제 서버 응답으로 직접 확인한 신버전 정책) +사용법: python qa_runner.py --dataset datasets/sample_dataset_v2_balanced.json +""" + +import json +import time +import argparse +import requests +from datetime import datetime +from pathlib import Path +from collections import Counter + +DEFAULT_URL = "http://localhost:8000" +DEFAULT_DATASET = "../datasets/sample_dataset_v2_balanced.json" +RESULTS_DIR = Path("results") + +QA_EMAIL = "qa_tester@example.com" +QA_PASSWORD = "qa-test-pass-0001" + + +def get_auth_token(base_url: str) -> str | None: + """QA 테스트용 계정을 가입(이미 있으면 무시) 후 로그인해 토큰을 받아온다.""" + creds = {"email": QA_EMAIL, "password": QA_PASSWORD} + try: + requests.post(f"{base_url}/auth/signup", json=creds, timeout=10) + except Exception as e: + print(f"[경고] 회원가입 요청 실패(무시하고 로그인 시도): {e}") + + try: + resp = requests.post(f"{base_url}/auth/login", json=creds, timeout=10) + resp.raise_for_status() + token = resp.json().get("access_token") + if not token: + print("[오류] 로그인 응답에 access_token이 없습니다.") + return None + print("[인증] 로그인 성공, 토큰 발급 완료") + return token + except Exception as e: + print(f"[오류] 로그인 실패: {e}") + return None + + +def labels_to_action(labels: list) -> str: + """팀 데이터셋 라벨 -> 기대 action. + [v3 변경] 계좌번호(PII_ACCOUNT_DETECTED) 단독 노출은 WARN으로 기대. + 실제 /proxy/chat 응답을 직접 호출해 확인한 신버전 정책을 반영했다. + """ + if not labels: + return "ALLOW" + inj = [l for l in labels if l.startswith("INJ_")] + pii = [l for l in labels if l.startswith("PII_")] + if inj: + return "BLOCK" + if "PII_RRN_DETECTED" in pii: + return "BLOCK" + if len(pii) >= 2: + return "BLOCK" + if pii: + # [v3] 계좌번호 단독 노출은 WARN (서버 실응답으로 확인함) + if set(pii) <= {"PII_ACCOUNT_DETECTED"}: + return "WARN" + return "MASK" + return "ALLOW" + + +def run_test(base_url: str, tc: dict, headers: dict) -> dict: + expected_action = labels_to_action(tc.get("labels", [])) + expected_labels = tc.get("labels", []) + payload = {"message": tc["text"], "user_id": "qa_tester"} + try: + resp = requests.post(f"{base_url}/proxy/chat", json=payload, headers=headers, timeout=10) + resp.raise_for_status() + data = resp.json() + except Exception as e: + return { + **tc, + "status": "ERROR", "error": str(e), + "expected_action": expected_action, + "actual_action": None, + "actual_reason_codes": [], + "expected_labels": expected_labels, + } + + actual_action = data.get("action", "") + # [v3] 신버전 응답에는 reasons 필드가 없음(action/content/input_masked/output_masked만 존재). + # reason_code 비교가 불가능하므로, 판정은 action 일치 여부만으로 한다. + actual_reason_codes = data.get("reasons", []) # 항상 빈 리스트일 수 있음(참고용으로만 기록) + action_pass = actual_action == expected_action + overall_pass = action_pass + + return { + **tc, + "status": "PASS" if overall_pass else "FAIL", + "action_pass": action_pass, + "codes_pass": None, # [v3] 더 이상 reason_code로 판정하지 않음 + "expected_action": expected_action, + "actual_action": actual_action, + "actual_reason_codes": actual_reason_codes, + "expected_labels": expected_labels, + "expected_codes_list": expected_labels, + "input_masked": data.get("input_masked"), + "output_masked": data.get("output_masked"), + "latency_ms": data.get("audit_summary", {}).get("latency_ms", 0), + } + + +def run_all(base_url: str, dataset_path: str, headers: dict) -> list: + with open(dataset_path, encoding="utf-8") as f: + test_cases = json.load(f) + + total = len(test_cases) + print(f"\n{'='*60}") + print(f" QA 자동 테스트 시작 (v3: action 기준 판정) | 총 {total}개 케이스") + print(f" 대상 서버: {base_url}") + print(f"{'='*60}\n") + + results = [] + for i, tc in enumerate(test_cases, 1): + result = run_test(base_url, tc, headers) + results.append(result) + icon = "OK" if result["status"] == "PASS" else ("FAIL" if result["status"] == "FAIL" else "ERR") + print(f"[{i:03d}/{total}] {icon} {tc['id']} | {result['expected_action']} vs {result['actual_action']} | {result['status']}") + time.sleep(0.05) + + return results + + +def save_results(results: list) -> str: + RESULTS_DIR.mkdir(exist_ok=True) + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + output_path = RESULTS_DIR / f"test_result_{timestamp}.json" + with open(output_path, "w", encoding="utf-8") as f: + json.dump(results, f, ensure_ascii=False, indent=2) + print(f"\n결과 저장: {output_path}") + return str(output_path) + + +def print_summary(results: list): + total = len(results) + passed = sum(1 for r in results if r["status"] == "PASS") + failed = sum(1 for r in results if r["status"] == "FAIL") + errors = sum(1 for r in results if r["status"] == "ERROR") + avg_lat = sum(r.get("latency_ms", 0) for r in results) / total if total else 0 + cat_fail = Counter(r["category"] for r in results if r["status"] == "FAIL") + + print(f"\n{'='*60}") + print(f" 테스트 결과 요약 (v3: action 일치 여부만으로 판정)") + print(f"{'='*60}") + print(f" 전체: {total}개") + print(f" PASS: {passed}개 ({passed/total*100:.1f}%)") + print(f" FAIL: {failed}개 ({failed/total*100:.1f}%)") + print(f" ERROR:{errors}개") + print(f" 평균 응답시간: {avg_lat:.1f}ms") + + if cat_fail: + print(f"\n 카테고리별 실패:") + for cat, cnt in cat_fail.most_common(): + print(f" {cat}: {cnt}건") + + if failed > 0: + print(f"\n{'='*60}") + print(" FAIL 케이스 (진짜 action 불일치만 남음)") + print(f"{'='*60}") + for r in results: + if r["status"] == "FAIL": + print(f" [{r['id']}] {r['text'][:40]}") + print(f" 기대: {r['expected_action']} 실제: {r['actual_action']}") + print(f"{'='*60}\n") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument("--url", default=DEFAULT_URL) + parser.add_argument("--dataset", default=DEFAULT_DATASET) + args = parser.parse_args() + + token = get_auth_token(args.url) + if not token: + print("\n[중단] 인증 토큰을 받지 못해 테스트를 진행할 수 없습니다.") + raise SystemExit(1) + + headers = {"Authorization": f"Bearer {token}"} + + results = run_all(args.url, args.dataset, headers) + result_path = save_results(results) + print_summary(results) + print(f"리포트 생성: python diff_report.py --input {result_path}")