qingsongping-AIDevelop/scripts/verify_blackbox.py

import requests
import json
import time
import threading
import pymysql
import redis
import os

BASE = "http://127.0.0.1:8080/api/v1"
HEADERS = {"Content-Type": "application/json"}
DB = {
    "host": os.environ.get("DB_HOST", "127.0.0.1"),
    "port": int(os.environ.get("DB_PORT", "3307")),
    "user": os.environ.get("DB_USERNAME", "mydreamjob"),
    "password": os.environ["DB_PASSWORD"],
    "database": os.environ.get("DB_NAME", "mydreamjob")
}

def db_conn():
    return pymysql.connect(**DB, autocommit=True)

def get_code(phone, purpose="login"):
    conn = db_conn()
    with conn.cursor() as c:
        c.execute("SELECT code FROM sms_verification_codes WHERE phone=%s AND purpose=%s AND used=0 ORDER BY id DESC LIMIT 1",
                  (phone, purpose))
        row = c.fetchone()
    conn.close()
    return row[0] if row else None

def auth_header(token):
    return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}

def login(phone, role):
    r = redis.Redis(host='127.0.0.1', port=6379)
    r.delete(f'sms:rate:{phone}')
    r.delete(f'sms:lockout:{phone}')
    requests.post(f"{BASE}/auth/sms/send", params={"phone": phone, "purpose": "login"}, headers=HEADERS)
    time.sleep(2)
    code = get_code(phone)
    if not code: raise Exception(f"No code for {phone}")
    r = requests.post(f"{BASE}/auth/login", json={"phone": phone, "code": code, "purpose": "login"}, headers=HEADERS)
    if r.status_code != 200: raise Exception(f"Login failed: {r.status_code} {r.text}")
    return r.json()["data"]["accessToken"]

results = []

def test(name, method, url, headers, expected_status, body=None, check_fn=None):
    full_url = f"{BASE}{url}" if not url.startswith("http") else url
    if body:
        r = requests.request(method, full_url, json=body, headers=headers)
    else:
        r = requests.request(method, full_url, headers=headers)
    status = r.status_code
    passed = status == expected_status
    detail = ""
    try:
        resp = r.json()
        if check_fn: passed = passed and check_fn(resp)
        detail = resp.get("message", "") or str(resp)[:200]
    except:
        detail = r.text[:200]
    results.append((name, method, url, status, expected_status, "PASS" if passed else "FAIL", detail))
    if not passed:
        print(f"  FAIL: {name} - expected {expected_status}, got {status}, detail={detail}")
    return r

print("=== Login ===")
company_token = login("13886593964", "COMPANY")
print(f"  Company token: {company_token[:20]}...")
admin_token = login("19986593964", "ADMIN")
print(f"  Admin token: {admin_token[:20]}...")
other_company_token = login("15186593964", "COMPANY")
print(f"  Other company token: {other_company_token[:20]}...")
student_token = login("13986593964", "STUDENT")
print(f"  Student token: {student_token[:20]}...")

# 1. Create company profile
print("\n=== Test 1: Create company profile ===")
company_body = {"name": "黑盒测试企业", "description": "test", "industry": "IT", "address": "北京"}
r = requests.get(f"{BASE}/companies/me", headers=auth_header(company_token))
if r.status_code == 200:
    body = r.json()
    if body.get("data") is not None:
        test("1. Company exists", "GET", "/companies/me", auth_header(company_token), 200, None,
             lambda j: j.get("data", {}).get("id") is not None)
    else:
        test("1. Create company", "POST", "/companies/me", auth_header(company_token), 200, company_body,
             lambda j: j.get("data", {}).get("id") is not None)
elif r.status_code == 404:
    test("1. Create company", "POST", "/companies/me", auth_header(company_token), 200, company_body,
         lambda j: j.get("data", {}).get("id") is not None)
else:
    test("1. Create company failed", "GET", "/companies/me", auth_header(company_token), r.status_code, None)

# 2. Submit verification
print("\n=== Test 2: Submit verification ===")
r = test("2. Submit verification", "POST", f"{BASE}/companies/me/verification",
        auth_header(company_token), 200,
        {"materialsJson": "{\"license\":\"http://example.com/license.jpg\"}"},
        lambda j: j.get("data", {}).get("status") == "PENDING")
verification_id = r.json().get("data", {}).get("id")
print(f"  Verification ID: {verification_id}")

# 3. Query own verification
print("\n=== Test 3: Query own verification ===")
test("3. Query own verification", "GET", f"{BASE}/companies/me/verification",
     auth_header(company_token), 200, None,
     lambda j: j.get("data", {}).get("status") == "PENDING")

# 4. Non-admin accesses admin list -> 403
print("\n=== Test 4: Non-admin -> 403 ===")
test("4. Company access admin list", "GET", f"{BASE}/admin/companies/verifications",
     auth_header(company_token), 403)
test("4b. Student access admin list", "GET", f"{BASE}/admin/companies/verifications",
     auth_header(student_token), 403)

# 5. Admin query pending list
print("\n=== Test 5: Admin query pending list ===")
test("5. Admin list pending", "GET", f"{BASE}/admin/companies/verifications?status=PENDING",
     auth_header(admin_token), 200,
     None, lambda j: j.get("data", {}).get("total", 0) > 0)

# 6. Admin query detail
print("\n=== Test 6: Admin query detail ===")
test("6. Admin query detail", "GET", f"{BASE}/admin/companies/verifications/{verification_id}",
     auth_header(admin_token), 200,
     None, lambda j: j.get("data", {}).get("status") == "PENDING")

# 7. REJECT without reason -> 400
print("\n=== Test 7: REJECT without reason ===")
test("7. REJECT without reason", "POST", f"{BASE}/admin/companies/verifications/{verification_id}/decision",
     auth_header(admin_token), 400, {"decision": "REJECTED"})

# 8. Admin REJECT -> 200
print("\n=== Test 8: Admin REJECT ===")
test("8. Admin REJECT", "POST", f"{BASE}/admin/companies/verifications/{verification_id}/decision",
     auth_header(admin_token), 200,
     {"decision": "REJECTED", "reason": "营业执照不清晰，请重新上传"},
     lambda j: j.get("data", {}).get("status") == "REJECTED")

# 9. Company sees REJECTED
print("\n=== Test 9: Company sees REJECTED ===")
test("9. Company sees REJECTED", "GET", f"{BASE}/companies/me/verification",
     auth_header(company_token), 200,
     None, lambda j: j.get("data", {}).get("status") == "REJECTED")

# 10. Company resubmit -> RESUBMITTED
print("\n=== Test 10: Resubmit ===")
r = test("10. Resubmit", "POST", f"{BASE}/companies/me/verification",
        auth_header(company_token), 200,
        {"materialsJson": "{\"license\":\"http://example.com/license_v2.jpg\"}"},
        lambda j: j.get("data", {}).get("status") == "RESUBMITTED")
verification_id2 = r.json().get("data", {}).get("id")
print(f"  New Verification ID: {verification_id2}")

# 11. Admin APPROVE -> 200
print("\n=== Test 11: Admin APPROVE ===")
test("11. Admin APPROVE", "POST", f"{BASE}/admin/companies/verifications/{verification_id2}/decision",
     auth_header(admin_token), 200,
     {"decision": "APPROVED", "reason": "材料齐全，通过"},
     lambda j: j.get("data", {}).get("status") == "APPROVED")

# 12. Double APPROVE -> 409
print("\n=== Test 12: Double APPROVE -> 409 ===")
test("12. Double APPROVE", "POST", f"{BASE}/admin/companies/verifications/{verification_id2}/decision",
     auth_header(admin_token), 409, {"decision": "APPROVED"})

# 13. Approved company resubmit -> 409
print("\n=== Test 13: Approved resubmit -> 409 ===")
test("13. Approved resubmit", "POST", f"{BASE}/companies/me/verification",
     auth_header(company_token), 409, {"materialsJson": "{}"})

# 14. Other company views detail -> 403
print("\n=== Test 14: Other company -> 403 ===")
test("14. Other company views detail", "GET", f"{BASE}/admin/companies/verifications/{verification_id2}",
     auth_header(other_company_token), 403)

# 15. Query history
print("\n=== Test 15: Query history ===")
test("15. Query history", "GET", f"{BASE}/admin/companies/verifications/history?verificationId={verification_id2}",
     auth_header(admin_token), 200,
     None, lambda j: len(j.get("data", [])) >= 2)

# 16. Verify companies.verified updated
print("\n=== Test 16: companies.verified ===")
conn = db_conn()
with conn.cursor() as c:
    c.execute("SELECT verified FROM companies WHERE name=%s", ("黑盒测试企业",))
    row = c.fetchone()
conn.close()
verified = row[0] if row else 0
print(f"  verified={verified}")
results.append(("16. companies.verified", "DB", "", 0, 0, "PASS" if verified == 1 else "FAIL", f"verified={verified}"))

# 17. Verify notification sent
print("\n=== Test 17: Notification ===")
conn = db_conn()
with conn.cursor() as c:
    c.execute("SELECT COUNT(*) FROM notifications WHERE user_id=(SELECT id FROM users WHERE phone=%s) AND type LIKE %s",
              ("13886593964", "VERIFICATION_%"))
    count = c.fetchone()[0]
conn.close()
print(f"  Notification count: {count}")
results.append(("17. Notification", "DB", "", 0, 0, "PASS" if count >= 2 else "FAIL", f"count={count}"))

# 18. Concurrent verification
print("\n=== Test 18: Concurrent ===")
# Ensure other company has a company profile and no pending verifications
conc_body = {"name": "并发测试企业", "description": "test", "industry": "IT", "address": "上海"}
r_other_get = requests.get(f"{BASE}/companies/me", headers=auth_header(other_company_token))
company_id = None
if r_other_get.status_code == 200 and r_other_get.json().get("data") is not None:
    company_id = r_other_get.json()["data"]["id"]
elif r_other_get.status_code == 404 or (r_other_get.status_code == 200 and r_other_get.json().get("data") is None):
    r_create = requests.post(f"{BASE}/companies/me", json=conc_body, headers=auth_header(other_company_token))
    if r_create.status_code == 200:
        company_id = r_create.json()["data"]["id"]
    elif r_create.status_code == 409:
        # company exists, get it
        r_get = requests.get(f"{BASE}/companies/me", headers=auth_header(other_company_token))
        if r_get.status_code == 200 and r_get.json().get("data"):
            company_id = r_get.json()["data"]["id"]

if company_id is None:
    results.append(("18. Concurrent", "DB", "", 0, 0, "FAIL", "no company for other user"))
else:
    # Clean up any existing pending verifications for this company
    conn = db_conn()
    with conn.cursor() as c:
        c.execute("DELETE FROM company_verification_history WHERE verification_id IN (SELECT id FROM company_verifications WHERE company_id=%s)", (company_id,))
        c.execute("DELETE FROM company_verifications WHERE company_id=%s", (company_id,))
    conn.close()

    r_sub = requests.post(f"{BASE}/companies/me/verification",
        json={"materialsJson": "{}"}, headers=auth_header(other_company_token))
    if r_sub.status_code == 200:
        cid = r_sub.json().get("data", {}).get("id")
        if cid:
            def approve():
                requests.post(f"{BASE}/admin/companies/verifications/{cid}/decision",
                    {"decision": "APPROVED", "reason": "test"}, headers=auth_header(admin_token))
            t1 = threading.Thread(target=approve)
            t2 = threading.Thread(target=approve)
            t1.start(); t2.start(); t1.join(); t2.join()
            # Verify only one history entry for APPROVED
            conn = db_conn()
            with conn.cursor() as c:
                c.execute("SELECT COUNT(*) FROM company_verification_history WHERE verification_id=%s AND to_status='APPROVED'", (cid,))
                cnt = c.fetchone()[0]
            conn.close()
            print(f"  Concurrent APPROVED history count: {cnt}")
            results.append(("18. Concurrent", "DB", "", 0, 0, "PASS" if cnt <= 1 else "FAIL", f"concurrent count={cnt}"))
        else:
            results.append(("18. Concurrent", "DB", "", 0, 0, "FAIL", "no verification id"))
    else:
        results.append(("18. Concurrent", "DB", "", 0, 0, "FAIL", f"submit failed: {r_sub.status_code} {r_sub.text[:200]}"))

# Summary
print("\n" + "="*60)
print("RESULTS SUMMARY")
print("="*60)
passed = 0
failed = 0
for name, method, url, status, expected, result, detail in results:
    print(f"  [{result}] {name} | {method} {url} | {status}/{expected} {detail}")
    if result == "PASS": passed += 1
    else: failed += 1
print(f"\nTOTAL: {len(results)} tests, {passed} PASS, {failed} FAIL")