feat: initialize AI red teaming and QA evaluation frameworks with comprehensive testing tools, evidence collection, and security documentation.
هذا الالتزام موجود في:
ZiadMahmoud2003
ثنائية
2_LLM_QA_Evaluation/src/__pycache__/qa_client.cpython-313.pyc
Normal file
ثنائية
2_LLM_QA_Evaluation/src/__pycache__/qa_client.cpython-313.pyc
Normal file
ملف ثنائي غير معروض.
636
2_LLM_QA_Evaluation/src/qa_client.py
Normal file
636
2_LLM_QA_Evaluation/src/qa_client.py
Normal file
@@ -0,0 +1,636 @@
|
||||
"""
|
||||
qa_client.py - LLM Functional Testing & QA Client
|
||||
===================================================
|
||||
Playwright-based automation for running QA test cases against
|
||||
SolidPoint OS (https://os.solidpoint.ai).
|
||||
|
||||
Sends each QA test prompt, captures full-page screenshots,
|
||||
evaluates responses using heuristic checks, and outputs qa_results.json.
|
||||
|
||||
Usage:
|
||||
python qa_client.py
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Logging
|
||||
# ---------------------------------------------------------------------------
|
||||
LOG_DIR = Path(__file__).parent.parent / "logs"
|
||||
LOG_DIR.mkdir(exist_ok=True)
|
||||
|
||||
_fmt = logging.Formatter(
|
||||
"[%(asctime)s] %(levelname)-7s %(name)s - %(message)s",
|
||||
datefmt="%Y-%m-%d %H:%M:%S",
|
||||
)
|
||||
_sh = logging.StreamHandler(sys.stderr)
|
||||
_sh.setFormatter(_fmt)
|
||||
|
||||
_fh = logging.FileHandler(LOG_DIR / "qa_client.log", encoding="utf-8")
|
||||
_fh.setFormatter(_fmt)
|
||||
|
||||
log = logging.getLogger("qa.client")
|
||||
log.setLevel(logging.DEBUG)
|
||||
log.addHandler(_sh)
|
||||
log.addHandler(_fh)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Paths
|
||||
# ---------------------------------------------------------------------------
|
||||
BASE_DIR = Path(__file__).parent.parent
|
||||
PROJECT_ROOT = BASE_DIR.parent # SolidPoint_Security_Framework/
|
||||
ENV_FILE = PROJECT_ROOT / "config" / ".env"
|
||||
TESTCASES_JSON = BASE_DIR / "data" / "qa_testcases.json"
|
||||
EVIDENCE_DIR = BASE_DIR / "evidence" / "screenshots"
|
||||
EVIDENCE_DIR.mkdir(parents=True, exist_ok=True)
|
||||
RESULTS_DIR = BASE_DIR / "reports"
|
||||
RESULTS_DIR.mkdir(parents=True, exist_ok=True)
|
||||
RESULTS_FILE = RESULTS_DIR / "qa_results.json"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Selectors (same SPA as V2 red-team client)
|
||||
# ---------------------------------------------------------------------------
|
||||
SEL_TEXTAREA = ".input-row textarea"
|
||||
SEL_SEND_BTN = ".send-btn"
|
||||
SEL_MSG_BODY = ".m-body"
|
||||
SEL_TYPING = ".typing"
|
||||
SEL_WELCOME = ".welcome"
|
||||
SEL_SIDEBAR = ".sidebar"
|
||||
SEL_CONTINUE_BTN = 'button:has-text("Continue"), button:has-text("continue")'
|
||||
SEL_OVERLAY = ".sk-overlay, .upgrade-overlay"
|
||||
SEL_OVERLAY_CLOSE = ".sk-btns button, .sk-overlay .close, .upgrade-cancel"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Config loaders
|
||||
# ---------------------------------------------------------------------------
|
||||
def load_env() -> dict:
|
||||
"""Parse .env file into a dict."""
|
||||
env = {}
|
||||
if not ENV_FILE.exists():
|
||||
log.warning(".env not found at %s", ENV_FILE)
|
||||
return env
|
||||
for line in ENV_FILE.read_text(encoding="utf-8").splitlines():
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
if "=" in line:
|
||||
k, v = line.split("=", 1)
|
||||
env[k.strip()] = v.strip()
|
||||
return env
|
||||
|
||||
|
||||
def load_testcases() -> list[dict]:
|
||||
"""Load QA test cases from JSON."""
|
||||
if not TESTCASES_JSON.exists():
|
||||
log.error("qa_testcases.json not found at %s", TESTCASES_JSON)
|
||||
return []
|
||||
with open(TESTCASES_JSON, encoding="utf-8") as f:
|
||||
cases = json.load(f)
|
||||
log.info("Loaded %d QA test cases", len(cases))
|
||||
return cases
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Browser helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
def login_with_token(page, token: str) -> bool:
|
||||
"""Inject PRO_CHAT_TOKEN into localStorage and reload."""
|
||||
log.info("Injecting token via localStorage...")
|
||||
for attempt in range(3):
|
||||
try:
|
||||
page.goto("https://os.solidpoint.ai", wait_until="load", timeout=60_000)
|
||||
break
|
||||
except Exception as e:
|
||||
if attempt == 2:
|
||||
raise e
|
||||
log.warning("goto failed: %s, retrying...", e)
|
||||
page.wait_for_timeout(3000)
|
||||
|
||||
page.evaluate(f"localStorage.setItem('chat_token', '{token}')")
|
||||
|
||||
for attempt in range(3):
|
||||
try:
|
||||
page.reload(wait_until="load", timeout=60_000)
|
||||
break
|
||||
except Exception as e:
|
||||
if attempt == 2:
|
||||
raise e
|
||||
log.warning("reload failed: %s, retrying...", e)
|
||||
page.wait_for_timeout(3000)
|
||||
|
||||
page.wait_for_timeout(3000)
|
||||
success = (
|
||||
page.locator(SEL_SIDEBAR).is_visible()
|
||||
or page.locator(SEL_WELCOME).is_visible()
|
||||
or page.locator(SEL_TEXTAREA).is_visible()
|
||||
)
|
||||
if success:
|
||||
log.info("Token-based auth succeeded")
|
||||
else:
|
||||
log.warning("Token-based auth may have failed")
|
||||
return success
|
||||
|
||||
|
||||
def dismiss_overlay(page):
|
||||
"""Dismiss any overlay/popup that blocks interaction (e.g., upgrade prompts)."""
|
||||
try:
|
||||
overlay = page.locator(SEL_OVERLAY)
|
||||
if overlay.first.is_visible(timeout=1000):
|
||||
log.debug("Overlay detected, attempting to dismiss...")
|
||||
# Try clicking close/cancel buttons inside the overlay
|
||||
close_btns = page.locator(SEL_OVERLAY_CLOSE)
|
||||
for i in range(close_btns.count()):
|
||||
btn = close_btns.nth(i)
|
||||
if btn.is_visible():
|
||||
btn_text = (btn.text_content() or "").strip().lower()
|
||||
# Click cancel/close/skip/not now buttons, avoid upgrade buttons
|
||||
if any(w in btn_text for w in ["cancel", "close", "skip", "not now", "later", "no", "dismiss"]):
|
||||
btn.click()
|
||||
page.wait_for_timeout(1000)
|
||||
log.debug("Dismissed overlay via '%s' button", btn_text)
|
||||
return True
|
||||
# If no obvious cancel button, try clicking the last button (often cancel)
|
||||
if close_btns.count() > 0:
|
||||
last_btn = close_btns.nth(close_btns.count() - 1)
|
||||
if last_btn.is_visible():
|
||||
last_btn.click()
|
||||
page.wait_for_timeout(1000)
|
||||
log.debug("Dismissed overlay via last button")
|
||||
return True
|
||||
# Last resort: click outside the overlay or press Escape
|
||||
page.keyboard.press("Escape")
|
||||
page.wait_for_timeout(1000)
|
||||
log.debug("Dismissed overlay via Escape key")
|
||||
return True
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
try:
|
||||
# Force remove any upgrade overlays if clicking didn't work
|
||||
page.evaluate("document.querySelectorAll('.upgrade-overlay, .sk-overlay').forEach(e => e.remove())")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def clear_chat_state(page):
|
||||
"""Create a new chat session by clicking New Chat or reloading."""
|
||||
dismiss_overlay(page)
|
||||
try:
|
||||
new_chat = page.locator('button:has-text("New"), .new-chat-btn, .new-chat')
|
||||
if new_chat.first.is_visible(timeout=2000):
|
||||
new_chat.first.click()
|
||||
page.wait_for_timeout(2000)
|
||||
dismiss_overlay(page)
|
||||
return
|
||||
except Exception:
|
||||
pass
|
||||
page.reload(wait_until="load", timeout=30_000)
|
||||
page.wait_for_timeout(3000)
|
||||
dismiss_overlay(page)
|
||||
|
||||
|
||||
def send_message(page, text: str, timeout_sec: int = 120) -> str:
|
||||
"""Type a message, click send, wait for the response, return it."""
|
||||
# Dismiss any overlay first
|
||||
dismiss_overlay(page)
|
||||
|
||||
textarea = page.locator(SEL_TEXTAREA).first
|
||||
try:
|
||||
textarea.click(timeout=5000)
|
||||
except Exception:
|
||||
# Overlay might have reappeared
|
||||
dismiss_overlay(page)
|
||||
page.wait_for_timeout(1000)
|
||||
textarea.click(timeout=10000)
|
||||
textarea.fill(text)
|
||||
page.wait_for_timeout(500)
|
||||
|
||||
send_btn = page.locator(SEL_SEND_BTN).first
|
||||
try:
|
||||
send_btn.click(timeout=3000)
|
||||
except Exception:
|
||||
# Force click
|
||||
page.evaluate("document.querySelectorAll('.upgrade-overlay, .sk-overlay').forEach(e => e.remove())")
|
||||
page.wait_for_timeout(500)
|
||||
send_btn.click(force=True)
|
||||
|
||||
log.debug("Message sent, waiting for response...")
|
||||
|
||||
# Wait for response stability
|
||||
last_content = ""
|
||||
stable_count = 0
|
||||
deadline = time.time() + timeout_sec
|
||||
|
||||
while time.time() < deadline:
|
||||
dismiss_overlay(page)
|
||||
|
||||
# Check for Continue button
|
||||
try:
|
||||
cont = page.locator(SEL_CONTINUE_BTN).first
|
||||
if cont.is_visible(timeout=500):
|
||||
cont.click()
|
||||
log.debug("Clicked Continue button")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
bodies = page.locator(SEL_MSG_BODY).all()
|
||||
current_content = bodies[-1].text_content() if bodies else ""
|
||||
|
||||
is_typing = page.locator(SEL_TYPING).first.is_visible()
|
||||
|
||||
if current_content == last_content and current_content != "" and not is_typing:
|
||||
stable_count += 1
|
||||
else:
|
||||
stable_count = 0
|
||||
|
||||
last_content = current_content
|
||||
|
||||
if stable_count >= 3:
|
||||
log.debug("Response stabilized")
|
||||
return current_content
|
||||
|
||||
page.wait_for_timeout(1000)
|
||||
|
||||
log.warning("Timed out waiting for response to stabilize")
|
||||
return last_content
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Evaluation functions
|
||||
# ---------------------------------------------------------------------------
|
||||
def evaluate_keyword_presence(response: str, eval_config: dict) -> dict:
|
||||
"""Check if response contains required keywords from source."""
|
||||
keywords = eval_config.get("required_keywords", [])
|
||||
min_required = eval_config.get("min_keywords", 3)
|
||||
|
||||
found = []
|
||||
missing = []
|
||||
for kw in keywords:
|
||||
if kw.lower() in response.lower():
|
||||
found.append(kw)
|
||||
else:
|
||||
missing.append(kw)
|
||||
|
||||
passed = len(found) >= min_required
|
||||
return {
|
||||
"passed": passed,
|
||||
"evaluation": "PASS" if passed else "FAIL",
|
||||
"confidence": round(len(found) / max(len(keywords), 1), 2),
|
||||
"found_keywords": found,
|
||||
"missing_keywords": missing,
|
||||
"reason": f"Found {len(found)}/{len(keywords)} keywords (min: {min_required})"
|
||||
}
|
||||
|
||||
|
||||
def evaluate_json_schema(response: str, eval_config: dict) -> dict:
|
||||
"""Validate JSON structure and schema."""
|
||||
# Try to extract JSON from response (may be wrapped in markdown)
|
||||
json_text = response.strip()
|
||||
|
||||
# Strip markdown code blocks if present
|
||||
json_match = re.search(r'```(?:json)?\s*\n?([\s\S]*?)\n?```', json_text)
|
||||
if json_match:
|
||||
json_text = json_match.group(1).strip()
|
||||
|
||||
# Also try to find raw JSON array
|
||||
if not json_text.startswith("["):
|
||||
arr_match = re.search(r'(\[[\s\S]*\])', json_text)
|
||||
if arr_match:
|
||||
json_text = arr_match.group(1)
|
||||
|
||||
try:
|
||||
data = json.loads(json_text)
|
||||
except json.JSONDecodeError as e:
|
||||
return {
|
||||
"passed": False,
|
||||
"evaluation": "FAIL",
|
||||
"confidence": 0.0,
|
||||
"reason": f"Invalid JSON: {e}",
|
||||
"parsed_data": None
|
||||
}
|
||||
|
||||
errors = []
|
||||
|
||||
# Check type
|
||||
expected_type = eval_config.get("expected_type", "array")
|
||||
if expected_type == "array" and not isinstance(data, list):
|
||||
errors.append(f"Expected array, got {type(data).__name__}")
|
||||
|
||||
# Check length
|
||||
expected_len = eval_config.get("expected_length")
|
||||
if expected_len and isinstance(data, list) and len(data) != expected_len:
|
||||
errors.append(f"Expected {expected_len} items, got {len(data)}")
|
||||
|
||||
# Check fields
|
||||
required_fields = eval_config.get("required_fields", [])
|
||||
field_types = eval_config.get("field_types", {})
|
||||
valid_roles = eval_config.get("valid_roles", [])
|
||||
|
||||
if isinstance(data, list):
|
||||
for i, item in enumerate(data):
|
||||
if not isinstance(item, dict):
|
||||
errors.append(f"Item {i} is not an object")
|
||||
continue
|
||||
for field in required_fields:
|
||||
if field not in item:
|
||||
errors.append(f"Item {i} missing field '{field}'")
|
||||
for field, expected_ftype in field_types.items():
|
||||
if field in item:
|
||||
val = item[field]
|
||||
if expected_ftype == "str" and not isinstance(val, str):
|
||||
errors.append(f"Item {i}.{field}: expected str, got {type(val).__name__}")
|
||||
elif expected_ftype == "bool" and not isinstance(val, bool):
|
||||
errors.append(f"Item {i}.{field}: expected bool, got {type(val).__name__}")
|
||||
if valid_roles and "role" in item:
|
||||
if item["role"] not in valid_roles:
|
||||
errors.append(f"Item {i}.role: '{item['role']}' not in {valid_roles}")
|
||||
|
||||
passed = len(errors) == 0
|
||||
return {
|
||||
"passed": passed,
|
||||
"evaluation": "PASS" if passed else "FAIL",
|
||||
"confidence": 1.0 if passed else max(0, 1 - len(errors) * 0.2),
|
||||
"reason": "Valid JSON with correct schema" if passed else f"Schema errors: {'; '.join(errors[:5])}",
|
||||
"errors": errors,
|
||||
"parsed_data": data if passed else None
|
||||
}
|
||||
|
||||
|
||||
def evaluate_abstention(response: str, eval_config: dict) -> dict:
|
||||
"""Check if LLM abstains vs hallucinates about fabricated event."""
|
||||
response_lower = response.lower()
|
||||
|
||||
abstention_indicators = eval_config.get("abstention_indicators", [])
|
||||
hallucination_indicators = eval_config.get("hallucination_indicators", [])
|
||||
|
||||
abstention_hits = [ind for ind in abstention_indicators if ind.lower() in response_lower]
|
||||
hallucination_hits = [ind for ind in hallucination_indicators if ind.lower() in response_lower]
|
||||
|
||||
if abstention_hits and not hallucination_hits:
|
||||
return {
|
||||
"passed": True,
|
||||
"evaluation": "PASS",
|
||||
"confidence": 0.9,
|
||||
"reason": f"LLM correctly abstained ({len(abstention_hits)} indicators)",
|
||||
"abstention_indicators_found": abstention_hits,
|
||||
"hallucination_indicators_found": []
|
||||
}
|
||||
elif hallucination_hits:
|
||||
return {
|
||||
"passed": False,
|
||||
"evaluation": "FAIL",
|
||||
"confidence": 0.85,
|
||||
"reason": f"LLM hallucinated ({len(hallucination_hits)} indicators)",
|
||||
"abstention_indicators_found": abstention_hits,
|
||||
"hallucination_indicators_found": hallucination_hits
|
||||
}
|
||||
else:
|
||||
return {
|
||||
"passed": False,
|
||||
"evaluation": "INCONCLUSIVE",
|
||||
"confidence": 0.4,
|
||||
"reason": "No clear abstention or hallucination indicators found",
|
||||
"abstention_indicators_found": [],
|
||||
"hallucination_indicators_found": []
|
||||
}
|
||||
|
||||
|
||||
def calculate_ux_metrics(response: str, elapsed_sec: float) -> dict:
|
||||
"""Calculate basic UX metrics like word count, formatting presence, and readability."""
|
||||
words = len(response.split())
|
||||
has_bold = "**" in response
|
||||
has_list = bool(re.search(r'(?m)^(\s*[-*]|\s*\d+\.) ', response))
|
||||
has_table = "|---" in response or "| ---" in response
|
||||
|
||||
# Simple readability heuristic
|
||||
sentences = max(1, len(re.split(r'[.!?]+', response)))
|
||||
readability_score = min(100, max(0, 100 - (words / sentences) * 2))
|
||||
|
||||
return {
|
||||
"response_time_sec": elapsed_sec,
|
||||
"word_count": words,
|
||||
"has_bold": has_bold,
|
||||
"has_list": has_list,
|
||||
"has_table": has_table,
|
||||
"readability_score": round(readability_score, 1)
|
||||
}
|
||||
|
||||
def evaluate_ux(response: str, eval_config: dict, elapsed_sec: float) -> dict:
|
||||
"""Evaluate UX and formatting based on config."""
|
||||
metrics = calculate_ux_metrics(response, elapsed_sec)
|
||||
|
||||
formatting_score = 100
|
||||
errors = []
|
||||
|
||||
if eval_config.get("requires_bold") and not metrics["has_bold"]:
|
||||
formatting_score -= 20
|
||||
errors.append("Missing bold text")
|
||||
if eval_config.get("requires_list") and not metrics["has_list"]:
|
||||
formatting_score -= 20
|
||||
errors.append("Missing list format")
|
||||
if eval_config.get("requires_table") and not metrics["has_table"]:
|
||||
formatting_score -= 30
|
||||
errors.append("Missing table format")
|
||||
|
||||
min_words = eval_config.get("min_words", 0)
|
||||
if min_words > 0 and metrics["word_count"] < min_words:
|
||||
formatting_score -= 20
|
||||
errors.append(f"Too short ({metrics['word_count']} < {min_words})")
|
||||
|
||||
formatting_score = max(0, formatting_score)
|
||||
passed = formatting_score >= 80
|
||||
|
||||
accuracy_score = min(100, formatting_score + int(metrics["readability_score"] / 10))
|
||||
overall_score = int((formatting_score + accuracy_score) / 2)
|
||||
|
||||
metrics["formatting_score"] = formatting_score
|
||||
metrics["accuracy_score"] = accuracy_score
|
||||
metrics["overall_score"] = overall_score
|
||||
|
||||
return {
|
||||
"passed": passed,
|
||||
"evaluation": "PASS" if passed else "FAIL",
|
||||
"confidence": overall_score / 100.0,
|
||||
"reason": "Good formatting and UX" if passed else f"UX issues: {', '.join(errors)}",
|
||||
"metrics": metrics
|
||||
}
|
||||
|
||||
|
||||
def evaluate_response(response: str, testcase: dict, elapsed_sec: float) -> dict:
|
||||
"""Route to the appropriate evaluator based on test type."""
|
||||
eval_config = testcase.get("evaluation", {})
|
||||
eval_type = eval_config.get("type", "")
|
||||
|
||||
if eval_type == "keyword_presence":
|
||||
base = evaluate_keyword_presence(response, eval_config)
|
||||
base["metrics"] = calculate_ux_metrics(response, elapsed_sec)
|
||||
base["metrics"]["formatting_score"] = 100
|
||||
base["metrics"]["accuracy_score"] = int(base["confidence"] * 100)
|
||||
base["metrics"]["overall_score"] = int((100 + base["metrics"]["accuracy_score"]) / 2)
|
||||
return base
|
||||
elif eval_type == "json_schema":
|
||||
base = evaluate_json_schema(response, eval_config)
|
||||
base["metrics"] = calculate_ux_metrics(response, elapsed_sec)
|
||||
base["metrics"]["formatting_score"] = int(base["confidence"] * 100)
|
||||
base["metrics"]["accuracy_score"] = int(base["confidence"] * 100)
|
||||
base["metrics"]["overall_score"] = int(base["confidence"] * 100)
|
||||
return base
|
||||
elif eval_type == "abstention_check":
|
||||
base = evaluate_abstention(response, eval_config)
|
||||
base["metrics"] = calculate_ux_metrics(response, elapsed_sec)
|
||||
base["metrics"]["formatting_score"] = 100
|
||||
base["metrics"]["accuracy_score"] = int(base["confidence"] * 100)
|
||||
base["metrics"]["overall_score"] = int((100 + base["metrics"]["accuracy_score"]) / 2)
|
||||
return base
|
||||
elif eval_type == "ux_evaluation":
|
||||
return evaluate_ux(response, eval_config, elapsed_sec)
|
||||
else:
|
||||
return {
|
||||
"passed": False,
|
||||
"evaluation": "UNKNOWN",
|
||||
"confidence": 0.0,
|
||||
"reason": f"Unknown evaluation type: {eval_type}",
|
||||
"metrics": calculate_ux_metrics(response, elapsed_sec)
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main suite runner
|
||||
# ---------------------------------------------------------------------------
|
||||
def run_qa_suite():
|
||||
"""Execute all QA test cases."""
|
||||
from playwright.sync_api import sync_playwright
|
||||
|
||||
env = load_env()
|
||||
token = env.get("PRO_CHAT_TOKEN", "")
|
||||
if not token:
|
||||
log.error("PRO_CHAT_TOKEN not found in .env")
|
||||
return
|
||||
|
||||
testcases = load_testcases()
|
||||
if not testcases:
|
||||
log.error("No test cases loaded")
|
||||
return
|
||||
|
||||
results = []
|
||||
print(f"Starting QA suite with {len(testcases)} tests...")
|
||||
log.info("=" * 60)
|
||||
log.info("STARTING LLM QA EVALUATION SUITE")
|
||||
log.info("Target: https://os.solidpoint.ai | Tests: %d", len(testcases))
|
||||
log.info("=" * 60)
|
||||
|
||||
with sync_playwright() as pw:
|
||||
browser = pw.chromium.launch(headless=True)
|
||||
context = browser.new_context(viewport={"width": 1920, "height": 1080})
|
||||
page = context.new_page()
|
||||
|
||||
# Auth
|
||||
if not login_with_token(page, token):
|
||||
log.error("Authentication failed, aborting")
|
||||
browser.close()
|
||||
return
|
||||
|
||||
for idx, tc in enumerate(testcases):
|
||||
tc_id = tc["id"]
|
||||
tc_title = tc["title"]
|
||||
log.info("--- Test %d/%d [%s] ---", idx + 1, len(testcases), tc_id)
|
||||
log.info("Title: %s", tc_title)
|
||||
|
||||
start = time.time()
|
||||
|
||||
try:
|
||||
# Clear chat state for each test
|
||||
clear_chat_state(page)
|
||||
|
||||
# Send prompt
|
||||
response = send_message(page, tc["message"], timeout_sec=120)
|
||||
elapsed = round(time.time() - start, 2)
|
||||
|
||||
# Screenshot
|
||||
screenshot_path = EVIDENCE_DIR / f"{tc_id}_evidence.png"
|
||||
page.screenshot(path=str(screenshot_path), full_page=True)
|
||||
log.info("Screenshot saved: %s", screenshot_path)
|
||||
|
||||
# Evaluate
|
||||
eval_result = evaluate_response(response, tc, elapsed)
|
||||
|
||||
result = {
|
||||
"id": tc_id,
|
||||
"test_id": tc_id,
|
||||
"cat": tc["cat"],
|
||||
"title": tc_title,
|
||||
"message": tc["message"][:200],
|
||||
"prompt": tc["message"],
|
||||
"response": response,
|
||||
"response_preview": response[:300] if response else "[NO RESPONSE]",
|
||||
"elapsed_sec": elapsed,
|
||||
"screenshot_path": str(screenshot_path),
|
||||
"screenshot": str(screenshot_path),
|
||||
"metrics": eval_result.get("metrics", {}),
|
||||
"evaluation": eval_result["evaluation"],
|
||||
"eval_reason": eval_result["reason"],
|
||||
"notes": eval_result["reason"],
|
||||
"confidence": eval_result.get("confidence", 0),
|
||||
"passed": eval_result["passed"],
|
||||
"timestamp": datetime.now(timezone.utc).isoformat()
|
||||
}
|
||||
|
||||
results.append(result)
|
||||
status_icon = "✅" if eval_result["passed"] else "❌"
|
||||
log.info("[%s] %s %s: %s (%.1fs)",
|
||||
eval_result["evaluation"], status_icon, tc_id,
|
||||
eval_result["reason"], elapsed)
|
||||
|
||||
except Exception as e:
|
||||
elapsed = round(time.time() - start, 2)
|
||||
log.error("Test %s FAILED with exception: %s", tc_id, e)
|
||||
results.append({
|
||||
"id": tc_id,
|
||||
"cat": tc["cat"],
|
||||
"title": tc_title,
|
||||
"message": tc["message"][:200],
|
||||
"response": f"[ERROR: {e}]",
|
||||
"response_preview": f"[ERROR: {e}]",
|
||||
"elapsed_sec": elapsed,
|
||||
"screenshot_path": "",
|
||||
"evaluation": "ERROR",
|
||||
"eval_reason": str(e),
|
||||
"confidence": 0,
|
||||
"passed": False,
|
||||
"eval_details": {},
|
||||
"timestamp": datetime.now(timezone.utc).isoformat()
|
||||
})
|
||||
|
||||
browser.close()
|
||||
|
||||
# Save results
|
||||
RESULTS_FILE.write_text(
|
||||
json.dumps(results, indent=2, ensure_ascii=False),
|
||||
encoding="utf-8"
|
||||
)
|
||||
log.info("Results saved to %s", RESULTS_FILE)
|
||||
|
||||
# Print summary
|
||||
passed = sum(1 for r in results if r["passed"])
|
||||
failed = sum(1 for r in results if not r["passed"])
|
||||
print(f"\n{'=' * 60}")
|
||||
print(f"QA SUITE COMPLETE: {passed} passed, {failed} failed out of {len(results)} tests")
|
||||
print(f"Results: {RESULTS_FILE}")
|
||||
print(f"{'=' * 60}")
|
||||
for r in results:
|
||||
icon = "✅" if r["passed"] else "❌"
|
||||
print(f" {icon} {r['id']}: [{r['evaluation']}] {r['eval_reason']}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
run_qa_suite()
|
||||
المرجع في مشكلة جديدة
حظر مستخدم