tests/phase10/test_collisions.py

"""
Phase 11 — Pipeline Collision Test Runner
==========================================
Runs ALL collision test cases against the live API and produces
a structured JSON report with per-failure classification.
"""
import json
import sys
import time
import re
import requests
from pathlib import Path

# ── Configuration ──
API_URL = "https://bayan10-bayan-api.hf.space/api/analyze"
DATASET_PATH = Path(__file__).parent / "gold_datasets" / "pipeline_collision.json"
REPORT_PATH = Path(__file__).parent / "reports" / "collision_report.json"

def strip_diacritics(text):
    """Remove Arabic diacritics for comparison."""
    return re.sub(r'[\u064B-\u065F\u0670]', '', text)

def normalize_for_compare(text):
    """Normalize text for fuzzy comparison (strip diacritics + collapse spaces)."""
    t = strip_diacritics(text)
    t = re.sub(r'\s+', ' ', t).strip()
    return t

def classify_failure(sample, actual, suggestions):
    """Classify root cause for a failure."""
    expected = sample["expected"]
    inp = sample["input"]
    category = sample.get("category", "")

    # Check which words are wrong
    exp_words = set(expected.split())
    act_words = set(actual.split())
    inp_words = set(inp.split())
    missing_fixes = exp_words - act_words  # Expected words not in actual
    unwanted = act_words - exp_words       # Actual words not in expected

    # Check suggestion stages
    stages = [s.get('type', '') for s in suggestions]
    has_spelling = 'spelling' in stages
    has_grammar = 'grammar' in stages
    has_punctuation = 'punctuation' in stages

    # Determine root cause
    if category == "spelling_blocks_grammar":
        # Spelling fixed ه→ة but locked the range, grammar couldn't fix gender
        if any(s.get('type') == 'spelling' for s in suggestions):
            grammar_words_missed = [w for w in missing_fixes if w not in inp_words]
            if grammar_words_missed:
                return "STAGELOCKER", "spelling→grammar lock collision", grammar_words_missed
        return "MODEL", "Grammar model missed correction", list(missing_fixes)

    elif category == "grammar_drops_spelling":
        return "PIPELINE", "Grammar stage dropped spelling fix", list(missing_fixes)

    elif category == "spelling_grammar_overlap":
        return "PIPELINE", "Spelling and grammar overlap conflict", list(missing_fixes)

    elif category == "multi_stage_collision":
        if not has_grammar and missing_fixes:
            return "MODEL", "Grammar model missed correction", list(missing_fixes)
        elif has_spelling and not has_grammar:
            return "STAGELOCKER", "Spelling lock blocked grammar", list(missing_fixes)
        return "PIPELINE", "Multi-stage interaction failure", list(missing_fixes)

    else:
        return "UNKNOWN", f"Unclassified failure in category '{category}'", list(missing_fixes)


def main():
    # Load dataset
    if not DATASET_PATH.exists():
        print(f"❌ Dataset not found: {DATASET_PATH}")
        sys.exit(1)

    with open(DATASET_PATH, 'r', encoding='utf-8') as f:
        samples = json.load(f)

    print(f"[COLLISION] Running {len(samples)} test cases against {API_URL}")
    print(f"{'='*70}")

    results = []
    passed = 0
    failed = 0
    errors = 0

    for i, s in enumerate(samples):
        sid = s["id"]
        print(f"  [{i+1}/{len(samples)}] {sid} ({s.get('category','')})... ", end="", flush=True)

        try:
            t0 = time.time()
            r = requests.post(API_URL, json={"text": s["input"]}, timeout=120)
            elapsed_ms = int((time.time() - t0) * 1000)
            resp = r.json()
            actual = resp.get("corrected", "")
            suggestions = resp.get("suggestions", [])

            # Normalize for comparison (strip diacritics, collapse spaces)
            norm_actual = normalize_for_compare(actual)
            norm_expected = normalize_for_compare(s["expected"])

            result = {
                "id": sid,
                "category": s.get("category", ""),
                "input": s["input"],
                "expected": s["expected"],
                "actual": actual,
                "suggestions": suggestions,
                "elapsed_ms": elapsed_ms,
            }

            if norm_actual == norm_expected:
                result["verdict"] = "PASS"
                passed += 1
                print(f"✅ PASS ({elapsed_ms}ms)")
            else:
                result["verdict"] = "FAIL"
                component, detail, missing = classify_failure(s, actual, suggestions)
                result["root_cause_component"] = component
                result["root_cause_detail"] = detail
                result["missing_words"] = missing
                failed += 1
                print(f"❌ FAIL ({elapsed_ms}ms)")
                print(f"       Input:    {s['input']}")
                print(f"       Expected: {s['expected']}")
                print(f"       Actual:   {actual}")
                print(f"       Cause:    [{component}] {detail}")

            results.append(result)

        except Exception as e:
            errors += 1
            results.append({
                "id": sid, "category": s.get("category", ""),
                "verdict": "ERROR", "error": str(e),
            })
            print(f"💥 ERROR: {e}")

    # ── Summary ──
    total = len(samples)
    pass_rate = (passed / total * 100) if total > 0 else 0

    print(f"\n{'='*70}")
    print(f"COLLISION BENCHMARK RESULTS")
    print(f"{'='*70}")
    print(f"  Total:     {total}")
    print(f"  Passed:    {passed}")
    print(f"  Failed:    {failed}")
    print(f"  Errors:    {errors}")
    print(f"  Pass Rate: {pass_rate:.1f}%")

    # ── Root cause breakdown ──
    failures = [r for r in results if r.get("verdict") == "FAIL"]
    by_component = {}
    by_category = {}
    for r in failures:
        comp = r.get("root_cause_component", "UNKNOWN")
        cat = r.get("category", "unknown")
        by_component[comp] = by_component.get(comp, 0) + 1
        by_category[cat] = by_category.get(cat, 0) + 1

    if failures:
        print(f"\n  Root Cause by Component:")
        for comp, count in sorted(by_component.items(), key=lambda x: -x[1]):
            print(f"    {comp}: {count}")
        print(f"\n  Failures by Category:")
        for cat, count in sorted(by_category.items(), key=lambda x: -x[1]):
            print(f"    {cat}: {count}")

    # ── Save report ──
    REPORT_PATH.parent.mkdir(parents=True, exist_ok=True)
    report = {
        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "target": API_URL,
        "total": total,
        "passed": passed,
        "failed": failed,
        "errors": errors,
        "pass_rate": round(pass_rate, 1),
        "root_cause_by_component": by_component,
        "failures_by_category": by_category,
        "results": results,
    }
    with open(REPORT_PATH, 'w', encoding='utf-8') as f:
        json.dump(report, f, ensure_ascii=False, indent=2)
    print(f"\n[COLLISION] Report saved → {REPORT_PATH}")


if __name__ == "__main__":
    main()