labor-arbitration-system/test_e2e.py

#!/usr/bin/env python3
"""End-to-end test for labor arbitration system: upload -> extract -> portrait -> similar"""

import json
import requests
import sys

BASE = "http://localhost:8000"
NLP = "http://localhost:8001"


def test_health():
    print("=== 0. Health Check ===")
    for name, url in [("Backend", f"{BASE}/health"), ("NLP", f"{NLP}/health")]:
        try:
            r = requests.get(url, timeout=5)
            print(f"  {name}: {r.json()}")
        except Exception as e:
            print(f"  {name}: ERROR - {e}")
            return False
    return True


def test_upload():
    print("\n=== 1. Upload Case ===")
    text = (
        "申请人:张明，男，汉族，1990年5月15日出生。\n"
        "被申请人:北京恒达科技有限公司，住所北京市海淀区中关村大街1号。\n\n"
        "请求事项：\n"
        "1、请求支付拖欠工资8000元。\n"
        "2、请求支付违法解除劳动合同赔偿金24000元。\n\n"
        "事实与理由：申请人于2020年3月1日入职被申请人处，担任软件工程师，"
        "月工资12000元。2023年6月被申请人无故辞退申请人，"
        "且拖欠2023年5月、6月工资合计8000元。\n\n"
        "依据《劳动合同法》第四十七条、第八十七条规定。"
    )
    files = {"files": ("test_case.txt", text.encode("utf-8"), "text/plain")}
    data = {"case_name": "张明诉北京恒达科技劳动仲裁案"}

    try:
        r = requests.post(f"{BASE}/api/cases/upload", files=files, data=data, timeout=60)
        result = r.json()
        print(f"  Status: {r.status_code}")

        # Extract case_id from various possible locations
        case_id = result.get("id") or result.get("case_id")
        if not case_id and "case" in result:
            case_id = result["case"].get("id")
        if not case_id:
            case_id = result.get("case_id_from_task")

        print(f"  Case ID: {case_id}")

        if "elements" in result:
            el = result["elements"]
            print(f"  Case Cause: {el.get('case_cause', 'N/A')}")
            print(f"  Applicant: {el.get('applicant_name', 'N/A')}")
            print(f"  Respondent: {el.get('respondent_name', 'N/A')}")

        return case_id
    except Exception as e:
        print(f"  ERROR: {e}")
        import traceback
        traceback.print_exc()
        return None


def test_elements(case_id):
    print(f"\n=== 2. Get Elements (case {case_id}) ===")
    try:
        r = requests.get(f"{BASE}/api/cases/{case_id}/elements", timeout=30)
        result = r.json()
        print(f"  Status: {r.status_code}")

        # Show top-level fields
        show_keys = [
            "case_cause", "applicant_name", "respondent_name",
            "entry_date", "leave_date", "month_salary",
            "worker_position", "contract_type", "law_refs",
            "primary_cause_type",
        ]
        for k in show_keys:
            v = result.get(k, "N/A")
            print(f"  {k}: {v}")

        # Show claims
        claims = result.get("claims", {})
        if claims:
            print(f"  claims: {claims}")

        # Check element table
        table = result.get("case_elements_table", {})
        if table:
            print(f"  Element table: {table.get('field_count', 0)} fields")
        return result
    except Exception as e:
        print(f"  ERROR: {e}")
        return None


def test_portrait(case_id):
    print(f"\n=== 3. Get Portrait (case {case_id}) ===")
    try:
        r = requests.get(f"{BASE}/api/cases/{case_id}/portrait", timeout=30)
        result = r.json()
        print(f"  Status: {r.status_code}")

        if "legal_score" in result:
            print(f"  Legal Score: {result['legal_score']}/100")
        if "fact_score" in result:
            print(f"  Fact Score: {result['fact_score']}/100")
        if "risk_score" in result:
            print(f"  Risk Score: {result['risk_score']}/100")
        if "risk_level" in result:
            print(f"  Risk Level: {result['risk_level']}")
        if "tags" in result:
            print(f"  Tags: {result['tags'][:5] if isinstance(result['tags'], list) else result['tags']}")
        if "keywords" in result:
            kws = result["keywords"]
            if isinstance(kws, list):
                print(f"  Keywords (top 10): {[kw['word'] for kw in kws[:10]]}")
        return result
    except Exception as e:
        print(f"  ERROR: {e}")
        return None


def test_similar(case_id):
    print(f"\n=== 4. Similar Cases (case {case_id}) ===")
    try:
        r = requests.post(
            f"{BASE}/api/cases/{case_id}/similar",
            json={"top_k": 3},
            timeout=30,
        )
        result = r.json()
        print(f"  Status: {r.status_code}")

        if isinstance(result, list):
            print(f"  Found {len(result)} similar cases")
            for i, case in enumerate(result[:3]):
                sim = case.get("similarity", case.get("score", 0))
                name = case.get("case_name", case.get("name", "unknown"))
                print(f"  [{i+1}] {name} (similarity: {sim})")
        elif isinstance(result, dict):
            cases = result.get("cases", result.get("results", []))
            print(f"  Found {len(cases)} similar cases")
        return result
    except Exception as e:
        print(f"  ERROR: {e}")
        return None


def test_nlp_service():
    print("\n=== 5. NLP Service Direct Test ===")
    text = (
        "申请人:张明，男，汉族，1990年5月15日出生。"
        "被申请人:北京恒达科技有限公司，住所北京市海淀区中关村大街1号。"
        "请求事项：1、请求支付拖欠工资8000元。"
        "2、请求支付违法解除劳动合同赔偿金24000元。"
        "事实与理由：申请人于2020年3月1日入职被申请人处，担任软件工程师，"
        "月工资12000元。2023年6月被申请人无故辞退申请人。"
        "依据《劳动合同法》第四十七条、第八十七条规定。"
    )
    try:
        r = requests.post(f"{NLP}/extract", json={"text": text}, timeout=60)
        result = r.json()
        print(f"  Status: {r.status_code}")
        cause = result.get("case_cause", {}).get("type", "N/A")
        print(f"  Case Cause: {cause}")
        parties = result.get("parties", {})
        print(f"  Applicant: {parties.get('applicant_name', 'N/A')}")
        print(f"  Respondent: {parties.get('respondent_name', 'N/A')}")
        facts = result.get("facts", {})
        print(f"  Entry Date: {facts.get('entry_date', 'N/A')}")
        print(f"  Month Salary: {facts.get('month_salary', 'N/A')}")
        return result
    except Exception as e:
        print(f"  ERROR: {e}")
        return None


def main():
    print("=" * 60)
    print("LABOR ARBITRATION SYSTEM - E2E TEST")
    print("=" * 60)

    if not test_health():
        print("\n[FAIL] Services not running!")
        print("Start them first:")
        print("  cd backend && uvicorn app.main:app --port 8000")
        print("  cd nlp-service && uvicorn app.main:app --port 8001")
        sys.exit(1)

    case_id = test_upload()
    if case_id:
        test_elements(case_id)
        test_portrait(case_id)
        test_similar(case_id)

    test_nlp_service()

    print("\n" + "=" * 60)
    print("E2E TEST COMPLETE")
    print("=" * 60)
    print(f"Frontend: http://localhost:5173")
    print(f"Backend:  http://localhost:8000")
    print(f"NLP:      http://localhost:8001")
    print(f"API Docs: http://localhost:8000/docs")


if __name__ == "__main__":
    main()