Create qa_runner

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}")