#!/usr/bin/env python3 """ Test script for DMP Artists Participants API endpoint Based on nodejs-example/app.js pattern """ import os import sys import time import requests from typing import Optional, Dict, Any from pathlib import Path from pprint import pprint # Load .env file if it exists try: from dotenv import load_dotenv env_path = Path(__file__).parent / '.env' if env_path.exists(): load_dotenv(env_path) print(f"✓ Loaded environment from {env_path}") except ImportError: # python-dotenv not installed, will use system environment variables pass class Auth0TokenManager: """Manages Auth0 JWT token with caching""" def __init__(self, domain: str, client_id: str, client_secret: str, audience: str): self.domain = domain self.client_id = client_id self.client_secret = client_secret self.audience = audience self.token: Optional[str] = None self.expires_at: float = 0 def get_token(self) -> str: """Get JWT token, refresh if expired""" if self.token and time.time() < self.expires_at: print("Using cached JWT token") return self.token print("Requesting new JWT from Auth0...") url = f"https://{self.domain}/oauth/token" payload = { "client_id": self.client_id, "client_secret": self.client_secret, "audience": self.audience, "grant_type": "client_credentials" } response = requests.post(url, json=payload) response.raise_for_status() data = response.json() self.token = data["access_token"] # Cache for expires_in seconds (usually 86400 = 24h), minus 5 min buffer expires_in = data.get("expires_in", 3600) self.expires_at = time.time() + expires_in - 300 print(f"✓ Received new JWT from Auth0 (expires in {expires_in}s)") print(f" Token length: {len(self.token)} characters") # Save token to file token_file = Path(__file__).parent / 'jwt_token.txt' with open(token_file, 'w') as f: f.write(self.token) print(f"✓ Token saved to: {token_file}") print("\n" + "=" * 80) print("RAW JWT TOKEN (Full):") print("=" * 80) # Print token without any truncation sys.stdout.write(self.token) sys.stdout.write('\n') sys.stdout.flush() print("=" * 80 + "\n") return self.token class DMPParticipantsAPI: """Client for DMP Artists Participants API""" def __init__(self, api_url: str, auth_manager: Auth0TokenManager): self.api_url = api_url self.auth_manager = auth_manager def get_participants( self, vendor_id: Optional[int] = None, subaccount_id: Optional[int] = None, search: Optional[str] = None, limit: Optional[int] = None, offset: Optional[int] = None, retries: int = 3 ) -> Dict[str, Any]: """ Get artists participants with optional filters Args: vendor_id: Account vendor id subaccount_id: Account subaccount id search: Search by partial name match limit: Page size (1-200) offset: Page offset (>=0) retries: Number of retry attempts for 5xx errors """ params = {} if vendor_id is not None: params["vendorId"] = vendor_id if subaccount_id is not None: params["subaccountId"] = subaccount_id if search is not None: params["search"] = search if limit is not None: params["limit"] = limit if offset is not None: params["offset"] = offset for attempt in range(retries + 1): try: token = self.auth_manager.get_token() headers = { "Authorization": f"Bearer {token}", "Accept": "application/json" } print(f"\n→ Calling {self.api_url}") if params: print(f" Parameters: {params}") response = requests.get( self.api_url, headers=headers, params=params, timeout=30 ) print(f"← Response status: {response.status_code}") if response.status_code == 200: data = response.json() print(f"✓ Success! Total items: {data.get('total', 'N/A')}") return data elif response.status_code == 401: print("✗ Unauthorized - invalid or expired token") # Try refreshing token once if attempt == 0: print(" Refreshing token and retrying...") self.auth_manager.token = None # Force refresh continue response.raise_for_status() elif response.status_code == 403: print("✗ Forbidden - IP not in whitelist or insufficient permissions") response.raise_for_status() elif response.status_code >= 500: print(f"✗ Server error (attempt {attempt + 1}/{retries + 1})") if attempt < retries: delay = 2 ** attempt # Exponential backoff: 1s, 2s, 4s print(f" Retrying in {delay}s...") time.sleep(delay) continue response.raise_for_status() else: response.raise_for_status() except requests.exceptions.RequestException as e: print(f"✗ Request error: {e}") if attempt < retries and "500" in str(e): delay = 2 ** attempt print(f" Retrying in {delay}s...") time.sleep(delay) continue raise raise Exception("Max retries exceeded") def main(): """Main test function""" # Load configuration from environment variables AUTH0_DOMAIN = os.getenv("AUTH0_DOMAIN", "qa-orchard.auth0.com") AUTH0_CLIENT_ID = os.getenv("AUTH0_CLIENT_ID") AUTH0_CLIENT_SECRET = os.getenv("AUTH0_CLIENT_SECRET") AUTH0_AUDIENCE = os.getenv("AUTH0_AUDIENCE", "https://qa-fan-response-jwt-authorizer") API_URL = os.getenv("DMP_PARTICIPANTS_API_URL", "https://qa-fan-response-api.theorchard.io/external/roster") # Validate required environment variables if not AUTH0_CLIENT_ID or not AUTH0_CLIENT_SECRET: print("ERROR: Missing required environment variables") print("\nRequired:") print(" AUTH0_CLIENT_ID - Your Auth0 client ID") print(" AUTH0_CLIENT_SECRET - Your Auth0 client secret") print("\nOptional:") print(" AUTH0_DOMAIN - Auth0 domain (default: qa-orchard.auth0.com)") print(" AUTH0_AUDIENCE - Auth0 API audience (default: https://qa-fan-response-jwt-authorizer)") print(" DMP_PARTICIPANTS_API_URL - API endpoint URL") print("\nExample:") print(" export AUTH0_CLIENT_ID='your-client-id'") print(" export AUTH0_CLIENT_SECRET='your-client-secret'") print(" python test_dmp_participants.py") sys.exit(1) print("=" * 60) print("DMP Artists Participants API Test") print("=" * 60) print(f"Auth0 Domain: {AUTH0_DOMAIN}") print(f"Client ID: {AUTH0_CLIENT_ID[:8]}...") print(f"Audience: {AUTH0_AUDIENCE}") print(f"API URL: {API_URL}") print("=" * 60) # Initialize clients auth_manager = Auth0TokenManager( domain=AUTH0_DOMAIN, client_id=AUTH0_CLIENT_ID, client_secret=AUTH0_CLIENT_SECRET, audience=AUTH0_AUDIENCE ) api_client = DMPParticipantsAPI( api_url=API_URL, auth_manager=auth_manager ) try: # Test 1: Basic request (no parameters) # print("\n" + "=" * 60) # print("Test 1: Get participants (no parameters)") # print("=" * 60) # result = api_client.get_participants() # print(f"\nItems returned: {len(result.get('items', []))}") # print(f"Total count: {result.get('total', 'N/A')}") # print(f"Limit: {result.get('limit', 'N/A')}") # print(f"Offset: {result.get('offset', 'N/A')}") # # if result.get('items'): # print("\nFirst item sample:") # first_item = result['items'][0] # for key, value in list(first_item.items())[:5]: # Show first 5 fields # print(f" {key}: {value}") # # Test 2: Search query (if applicable) # print("\n" + "=" * 60) # print("Test 2: Search with query parameter") # print("=" * 60) # result = api_client.get_participants(search="test", limit=5) # print(f"\nItems returned: {len(result.get('items', []))}") # print(f"Total matching: {result.get('total', 'N/A')}") # # Test 3: Filter by vendor_id print("\n" + "=" * 60) print("Test 3: Filter by vendor_id 7123") print("=" * 60) result = api_client.get_participants(vendor_id=7123, limit=10) print(f"\nItems returned: {len(result.get('items', []))}") print(f"Total matching: {result.get('total', 'N/A')}") if result.get('items'): print("\nSample items:") for idx, item in enumerate(result['items'][:3], 1): print(f"\n Item {idx}:") for key, value in list(item.items())[:5]: print(f" {key}: {value}") # Test 4: Pagination print("\n" + "=" * 60) print("Test 4: Pagination (offset=10, limit=5)") print("=" * 60) result = api_client.get_participants(offset=10, limit=5) print(f"\nItems returned: {len(result.get('items', []))}") print(f"Offset: {result.get('offset', 'N/A')}") print("\n" + "=" * 60) print("✓ All tests completed successfully!") print("=" * 60) pprint(result) except Exception as e: print("\n" + "=" * 60) print(f"✗ Test failed: {e}") print("=" * 60) sys.exit(1) if __name__ == "__main__": main()