File size: 2,159 Bytes
9512958
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
import pandas as pd
import requests
import json
import random

def load_dataset(dataset_name):
    return pd.read_parquet(f"hf://datasets/dwb2023/{dataset_name}/data/train-00000-of-00001.parquet")

def test_endpoint(base_url, endpoint, payload):
    url = f"{base_url}{endpoint}"
    print(f"\nTesting {endpoint}")
    print(f"Request: POST {url}")
    print(f"Payload: {json.dumps(payload, indent=2)}")
    try:
        response = requests.post(url, json=payload)
        print(f"Response Status: {response.status_code}")
        print(f"Response Body: {json.dumps(response.json(), indent=2)}")
        if response.status_code == 200:
            print(f"βœ… {endpoint} - Success")
        else:
            print(f"❌ {endpoint} - Failed")
    except requests.exceptions.RequestException as e:
        print(f"❌ {endpoint} - Error: {str(e)}")

def main(base_url):
    print(f"Testing Blackbird API at {base_url}\n")

    # Load datasets
    customers_df = load_dataset("blackbird-customers")
    orders_df = load_dataset("blackbird-orders")

    # Prepare test cases
    random_customer = customers_df.sample(1).iloc[0]
    random_order = orders_df.sample(1).iloc[0]
    processing_order = orders_df[orders_df['status'] == 'Processing'].sample(1).iloc[0]

    endpoints = [
        ("/get_user", {"key": "email", "value": random_customer['email']}),
        ("/update_user", {"user_id": random_customer['id'], "email": f"updated_{random_customer['email']}", "phone": f"updated_{random_customer['phone']}"}),
        ("/get_order_by_id", {"order_id": random_order['id']}),
        ("/get_customer_orders", {"customer_id": random_customer['id']}),
        ("/cancel_order", {"order_id": processing_order['id']}),
        ("/get_user_info", {"key": "email", "value": f"updated_{random_customer['email']}"})
    ]

    for endpoint, payload in endpoints:
        test_endpoint(base_url, endpoint, payload)

    # Test get_user with old email (should return 404)
    test_endpoint(base_url, "/get_user", {"key": "email", "value": random_customer['email']})

if __name__ == "__main__":
    base_url = "https://dwb2023-blackbird-svc.hf.space"
    main(base_url)