Dovodi_Backend/scripts/test_webhook.py

#!/usr/bin/env python3
"""
Test script for PlugNMeet webhook endpoint.

Usage:
    python scripts/test_webhook.py [event_type]
    
Event types:
    - room_finished
    - participant_joined
    - participant_left
    - end_recording
"""

import sys
import os
import json
import hmac
import hashlib
import requests
from datetime import datetime

# Django setup
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')

import django
django.setup()

from django.conf import settings


def calculate_signature(payload: str, secret: str) -> str:
    """Calculate HMAC SHA256 signature."""
    return hmac.new(
        secret.encode('utf-8'),
        payload.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()


def get_webhook_url() -> str:
    """Get webhook URL from settings or use default."""
    base_url = getattr(settings, 'BASE_URL', 'http://localhost:8000')
    return f"{base_url}/api/course/plugnmeet/webhook/"


def get_test_payload(event_type: str) -> dict:
    """Generate test payload for given event type."""
    timestamp = int(datetime.now().timestamp())
    
    payloads = {
        'room_finished': {
            "event": "room_finished",
            "id": "550e8400-e29b-41d4-a716-446655440000",
            "createdAt": timestamp,
            "room": {
                "sid": "room-123456",
                "identity": "test-room-20240101120000",
                "name": "Test Class",
                "maxParticipants": 100,
                "creationTime": timestamp - 3600,
                "metadata": "{}",
                "numParticipants": 0,
                "duration": 3600
            }
        },
        'participant_joined': {
            "event": "participant_joined",
            "id": "660e8400-e29b-41d4-a716-446655440001",
            "createdAt": timestamp,
            "room": {
                "sid": "room-123456",
                "identity": "test-room-20240101120000",
                "name": "Test Class"
            },
            "participant": {
                "sid": "participant-user-1",
                "identity": "1",
                "state": "ACTIVE",
                "name": "Test User",
                "metadata": '{"is_admin": false}',
                "permission": {
                    "canPublish": True,
                    "canPublishData": True,
                    "canSubscribe": True
                },
                "tracks": [],
                "joinedAt": timestamp
            }
        },
        'participant_left': {
            "event": "participant_left",
            "id": "770e8400-e29b-41d4-a716-446655440002",
            "createdAt": timestamp,
            "room": {
                "sid": "room-123456",
                "identity": "test-room-20240101120000",
                "name": "Test Class"
            },
            "participant": {
                "sid": "participant-user-1",
                "identity": "1",
                "state": "DISCONNECTED",
                "name": "Test User",
                "metadata": '{"is_admin": false}',
                "permission": {
                    "canPublish": True,
                    "canPublishData": True,
                    "canSubscribe": True
                },
                "tracks": [],
                "joinedAt": timestamp - 1800,
                "duration": 1800
            }
        },
        'end_recording': {
            "event": "end_recording",
            "id": "880e8400-e29b-41d4-a716-446655440003",
            "createdAt": timestamp,
            "room": {
                "sid": "room-123456",
                "identity": "test-room-20240101120000",
                "name": "Test Class"
            },
            "recording_info": {
                "recordingId": "rec-123456",
                "roomId": "test-room-20240101120000",
                "recordingType": "COMPOSITE",
                "fileName": "test-room-20240101120000.mp4",
                "duration": 3600,
                "status": "FINISHED"
            }
        }
    }
    
    return payloads.get(event_type)


def send_webhook(event_type: str, dry_run: bool = False):
    """Send webhook request to the endpoint."""
    # Get configuration
    webhook_url = get_webhook_url()
    api_secret = getattr(settings, 'PLUGNMEET_API_SECRET', '')
    
    if not api_secret:
        print("❌ Error: PLUGNMEET_API_SECRET not configured in settings")
        return False
    
    # Get test payload
    payload = get_test_payload(event_type)
    if not payload:
        print(f"❌ Error: Unknown event type '{event_type}'")
        print("Available event types: room_finished, participant_joined, participant_left, end_recording")
        return False
    
    # Convert payload to JSON string
    payload_json = json.dumps(payload, ensure_ascii=False)
    
    # Calculate signature
    signature = calculate_signature(payload_json, api_secret)
    
    # Print test information
    print("\n" + "="*80)
    print(f"🧪 Testing PlugNMeet Webhook: {event_type}")
    print("="*80)
    print(f"\n📍 URL: {webhook_url}")
    print(f"\n📦 Payload:")
    print(json.dumps(payload, indent=2, ensure_ascii=False))
    print(f"\n🔐 Signature: {signature[:20]}...")
    
    if dry_run:
        print("\n⚠️  DRY RUN - Not sending actual request")
        return True
    
    # Send request
    try:
        headers = {
            'Content-Type': 'application/webhook+json',
            'Hash-Token': signature,
        }
        
        print("\n📤 Sending request...")
        response = requests.post(
            webhook_url,
            data=payload_json.encode('utf-8'),
            headers=headers,
            timeout=10
        )
        
        print(f"\n✅ Response Status: {response.status_code}")
        print(f"📄 Response Body:")
        try:
            print(json.dumps(response.json(), indent=2, ensure_ascii=False))
        except:
            print(response.text)
        
        if response.status_code == 200:
            print("\n✅ Webhook test successful!")
            return True
        else:
            print(f"\n❌ Webhook test failed with status {response.status_code}")
            return False
            
    except requests.exceptions.RequestException as e:
        print(f"\n❌ Error sending request: {e}")
        return False
    finally:
        print("\n" + "="*80 + "\n")


def main():
    """Main function."""
    if len(sys.argv) < 2:
        print("Usage: python scripts/test_webhook.py [event_type]")
        print("\nAvailable event types:")
        print("  - room_finished")
        print("  - participant_joined")
        print("  - participant_left")
        print("  - end_recording")
        print("\nOptions:")
        print("  --dry-run    Show payload without sending request")
        sys.exit(1)
    
    event_type = sys.argv[1]
    dry_run = '--dry-run' in sys.argv
    
    success = send_webhook(event_type, dry_run=dry_run)
    sys.exit(0 if success else 1)


if __name__ == '__main__':
    main()