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)