From 315781aa5ebd2f1ffbb9833fed289bc72582ca4f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=9C=A0=EC=88=98=EC=A0=95?= <163844218+Nuysj@users.noreply.github.com> Date: Tue, 5 May 2026 08:56:44 +0900 Subject: [PATCH 1/4] Create qa_runner --- tools/qa_runner | 148 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 148 insertions(+) create mode 100644 tools/qa_runner diff --git a/tools/qa_runner b/tools/qa_runner new file mode 100644 index 0000000..69c6075 --- /dev/null +++ b/tools/qa_runner @@ -0,0 +1,148 @@ +""" +E2E 테스트 자동 실행기 (팀 JSON 데이터셋 지원) +사용법: 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:5000" +DEFAULT_DATASET = "../datasets/sample_dataset_v2_balanced.json" +RESULTS_DIR = Path("results") + +def labels_to_action(labels: list) -> str: + 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: + return "MASK" + return "ALLOW" + +def run_test(base_url: str, tc: dict) -> dict: + expected_action = labels_to_action(tc.get("labels", [])) + expected_labels = tc.get("labels", []) + payload = {"text": tc["text"], "user_id": "qa_tester"} + try: + resp = requests.post(f"{base_url}/proxy", json=payload, 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", "") + actual_reason_codes = data.get("reason_codes", []) + action_pass = actual_action == expected_action + + if not expected_labels: + codes_pass = len(actual_reason_codes) == 0 + else: + codes_pass = any( + any(exp.lower() in act.lower() or act.lower() in exp.lower() + for act in actual_reason_codes) + for exp in expected_labels + ) + + overall_pass = action_pass and codes_pass + + return { + **tc, + "status": "PASS" if overall_pass else "FAIL", + "action_pass": action_pass, + "codes_pass": codes_pass, + "expected_action": expected_action, + "actual_action": actual_action, + "actual_reason_codes": actual_reason_codes, + "expected_labels": expected_labels, + "latency_ms": data.get("latency_ms", 0), + } + +def run_all(base_url: str, dataset_path: str) -> 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 자동 테스트 시작 | 총 {total}개 케이스") + print(f" 대상 서버: {base_url}") + print(f"{'='*60}\n") + + results = [] + for i, tc in enumerate(test_cases, 1): + result = run_test(base_url, tc) + 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" 테스트 결과 요약") + 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 케이스") + 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() + + results = run_all(args.url, args.dataset) + result_path = save_results(results) + print_summary(results) + print(f"리포트 생성: python diff_report.py --input {result_path}") From 1a457b493745f933b3b72a280153ecfa96bccbd1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=9C=A0=EC=88=98=EC=A0=95?= <163844218+Nuysj@users.noreply.github.com> Date: Thu, 24 Sep 2026 10:23:11 +0900 Subject: [PATCH 2/4] Update qa_runner --- tools/qa_runner | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/tools/qa_runner b/tools/qa_runner index 69c6075..2f83a4d 100644 --- a/tools/qa_runner +++ b/tools/qa_runner @@ -11,7 +11,7 @@ from datetime import datetime from pathlib import Path from collections import Counter -DEFAULT_URL = "http://localhost:5000" +DEFAULT_URL = "http://localhost:8000" DEFAULT_DATASET = "../datasets/sample_dataset_v2_balanced.json" RESULTS_DIR = Path("results") @@ -33,9 +33,9 @@ def labels_to_action(labels: list) -> str: def run_test(base_url: str, tc: dict) -> dict: expected_action = labels_to_action(tc.get("labels", [])) expected_labels = tc.get("labels", []) - payload = {"text": tc["text"], "user_id": "qa_tester"} + payload = {"message": tc["text"], "user_id": "qa_tester"} try: - resp = requests.post(f"{base_url}/proxy", json=payload, timeout=10) + resp = requests.post(f"{base_url}/proxy/chat", json=payload, timeout=10) resp.raise_for_status() data = resp.json() except Exception as e: @@ -49,7 +49,7 @@ def run_test(base_url: str, tc: dict) -> dict: } actual_action = data.get("action", "") - actual_reason_codes = data.get("reason_codes", []) + actual_reason_codes = data.get("reasons", []) action_pass = actual_action == expected_action if not expected_labels: @@ -61,6 +61,11 @@ def run_test(base_url: str, tc: dict) -> dict: for exp in expected_labels ) + # [수정] 프록시는 정상 입력에도 SAFE_INPUT 코드를 반환하므로, 이를 탐지 코드로 보지 않는다. + real_codes = [c for c in actual_reason_codes if not c.startswith("SAFE_")] + codes_pass_strict = codes_pass # 원본 판정 (비교용 보존) + if not expected_labels: + codes_pass = len(real_codes) == 0 overall_pass = action_pass and codes_pass return { @@ -68,11 +73,13 @@ def run_test(base_url: str, tc: dict) -> dict: "status": "PASS" if overall_pass else "FAIL", "action_pass": action_pass, "codes_pass": codes_pass, + "codes_pass_strict": codes_pass_strict, "expected_action": expected_action, "actual_action": actual_action, "actual_reason_codes": actual_reason_codes, "expected_labels": expected_labels, - "latency_ms": data.get("latency_ms", 0), + "expected_codes_list": expected_labels, + "latency_ms": data.get("audit_summary", {}).get("latency_ms", 0), } def run_all(base_url: str, dataset_path: str) -> list: From 836ce5b84add88c3272f9392edffe82fa1fa3a72 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=9C=A0=EC=88=98=EC=A0=95?= <163844218+Nuysj@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:05:45 +0900 Subject: [PATCH 3/4] Update qa_runner --- tools/qa_runner | 59 ++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 51 insertions(+), 8 deletions(-) diff --git a/tools/qa_runner b/tools/qa_runner index 2f83a4d..cf2dfbd 100644 --- a/tools/qa_runner +++ b/tools/qa_runner @@ -1,6 +1,7 @@ """ E2E 테스트 자동 실행기 (팀 JSON 데이터셋 지원) -사용법: python qa_runner.py --dataset ../datasets/sample_dataset_v2_balanced.json +신버전 대응: 로그인(토큰) 인증 자동 처리 추가 +사용법: python qa_runner.py --dataset datasets/sample_dataset_v2_balanced.json """ import json @@ -15,6 +16,36 @@ DEFAULT_URL = "http://localhost:8000" DEFAULT_DATASET = "../datasets/sample_dataset_v2_balanced.json" RESULTS_DIR = Path("results") +# [신규] QA 테스트 전용 계정 (서버에 자동 가입 후 로그인) +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} + + # 가입 시도 (이미 가입돼 있으면 409가 나는데, 무시하고 넘어간다) + 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(f"[인증] 로그인 성공, 토큰 발급 완료") + return token + except Exception as e: + print(f"[오류] 로그인 실패: {e}") + return None + + def labels_to_action(labels: list) -> str: if not labels: return "ALLOW" @@ -30,12 +61,13 @@ def labels_to_action(labels: list) -> str: return "MASK" return "ALLOW" -def run_test(base_url: str, tc: dict) -> dict: + +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, timeout=10) + 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: @@ -61,9 +93,9 @@ def run_test(base_url: str, tc: dict) -> dict: for exp in expected_labels ) - # [수정] 프록시는 정상 입력에도 SAFE_INPUT 코드를 반환하므로, 이를 탐지 코드로 보지 않는다. + # 프록시는 정상 입력에도 SAFE_INPUT 코드를 반환하므로, 이를 탐지 코드로 보지 않는다. real_codes = [c for c in actual_reason_codes if not c.startswith("SAFE_")] - codes_pass_strict = codes_pass # 원본 판정 (비교용 보존) + codes_pass_strict = codes_pass if not expected_labels: codes_pass = len(real_codes) == 0 overall_pass = action_pass and codes_pass @@ -82,7 +114,8 @@ def run_test(base_url: str, tc: dict) -> dict: "latency_ms": data.get("audit_summary", {}).get("latency_ms", 0), } -def run_all(base_url: str, dataset_path: str) -> list: + +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) @@ -94,7 +127,7 @@ def run_all(base_url: str, dataset_path: str) -> list: results = [] for i, tc in enumerate(test_cases, 1): - result = run_test(base_url, tc) + 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']}") @@ -102,6 +135,7 @@ def run_all(base_url: str, dataset_path: str) -> list: return results + def save_results(results: list) -> str: RESULTS_DIR.mkdir(exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") @@ -111,6 +145,7 @@ def save_results(results: list) -> str: 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") @@ -143,13 +178,21 @@ def print_summary(results: list): 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() - results = run_all(args.url, args.dataset) + 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}") From 699a5fba949db6af3fce5152f5bd3b23189debef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=9C=A0=EC=88=98=EC=A0=95?= <163844218+Nuysj@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:27:24 +0900 Subject: [PATCH 4/4] Update qa_runner --- tools/qa_runner | 56 +++++++++++++++++++++++-------------------------- 1 file changed, 26 insertions(+), 30 deletions(-) diff --git a/tools/qa_runner b/tools/qa_runner index cf2dfbd..46265a4 100644 --- a/tools/qa_runner +++ b/tools/qa_runner @@ -1,6 +1,11 @@ """ 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 """ @@ -16,7 +21,6 @@ DEFAULT_URL = "http://localhost:8000" DEFAULT_DATASET = "../datasets/sample_dataset_v2_balanced.json" RESULTS_DIR = Path("results") -# [신규] QA 테스트 전용 계정 (서버에 자동 가입 후 로그인) QA_EMAIL = "qa_tester@example.com" QA_PASSWORD = "qa-test-pass-0001" @@ -24,14 +28,11 @@ QA_PASSWORD = "qa-test-pass-0001" def get_auth_token(base_url: str) -> str | None: """QA 테스트용 계정을 가입(이미 있으면 무시) 후 로그인해 토큰을 받아온다.""" creds = {"email": QA_EMAIL, "password": QA_PASSWORD} - - # 가입 시도 (이미 가입돼 있으면 409가 나는데, 무시하고 넘어간다) 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() @@ -39,7 +40,7 @@ def get_auth_token(base_url: str) -> str | None: if not token: print("[오류] 로그인 응답에 access_token이 없습니다.") return None - print(f"[인증] 로그인 성공, 토큰 발급 완료") + print("[인증] 로그인 성공, 토큰 발급 완료") return token except Exception as e: print(f"[오류] 로그인 실패: {e}") @@ -47,6 +48,10 @@ def get_auth_token(base_url: str) -> str | 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_")] @@ -58,6 +63,9 @@ def labels_to_action(labels: list) -> str: if len(pii) >= 2: return "BLOCK" if pii: + # [v3] 계좌번호 단독 노출은 WARN (서버 실응답으로 확인함) + if set(pii) <= {"PII_ACCOUNT_DETECTED"}: + return "WARN" return "MASK" return "ALLOW" @@ -80,37 +88,25 @@ def run_test(base_url: str, tc: dict, headers: dict) -> dict: "expected_labels": expected_labels, } - actual_action = data.get("action", "") - actual_reason_codes = data.get("reasons", []) - action_pass = actual_action == expected_action - - if not expected_labels: - codes_pass = len(actual_reason_codes) == 0 - else: - codes_pass = any( - any(exp.lower() in act.lower() or act.lower() in exp.lower() - for act in actual_reason_codes) - for exp in expected_labels - ) - - # 프록시는 정상 입력에도 SAFE_INPUT 코드를 반환하므로, 이를 탐지 코드로 보지 않는다. - real_codes = [c for c in actual_reason_codes if not c.startswith("SAFE_")] - codes_pass_strict = codes_pass - if not expected_labels: - codes_pass = len(real_codes) == 0 - overall_pass = action_pass and codes_pass + 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": codes_pass, - "codes_pass_strict": codes_pass_strict, + "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), } @@ -121,7 +117,7 @@ def run_all(base_url: str, dataset_path: str, headers: dict) -> list: total = len(test_cases) print(f"\n{'='*60}") - print(f" QA 자동 테스트 시작 | 총 {total}개 케이스") + print(f" QA 자동 테스트 시작 (v3: action 기준 판정) | 총 {total}개 케이스") print(f" 대상 서버: {base_url}") print(f"{'='*60}\n") @@ -155,7 +151,7 @@ def print_summary(results: list): cat_fail = Counter(r["category"] for r in results if r["status"] == "FAIL") print(f"\n{'='*60}") - print(f" 테스트 결과 요약") + print(f" 테스트 결과 요약 (v3: action 일치 여부만으로 판정)") print(f"{'='*60}") print(f" 전체: {total}개") print(f" PASS: {passed}개 ({passed/total*100:.1f}%)") @@ -170,7 +166,7 @@ def print_summary(results: list): if failed > 0: print(f"\n{'='*60}") - print(" FAIL 케이스") + print(" FAIL 케이스 (진짜 action 불일치만 남음)") print(f"{'='*60}") for r in results: if r["status"] == "FAIL":