384 lines
12 KiB
Python
Executable file
384 lines
12 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""
|
|
Webhook Test Script
|
|
Emulates A1→A2 and A4 webhooks for testing without running full workflows
|
|
Does NOT write to database - sends webhook only
|
|
Compatible with Python 3.6+
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import time
|
|
import logging
|
|
import argparse
|
|
import json
|
|
|
|
# Add shared library to path
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
|
|
|
from shared.config_loader import load_config
|
|
from shared.notifier import Notifier
|
|
|
|
# Setup logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(levelname)s - %(message)s'
|
|
)
|
|
|
|
logger = logging.getLogger('WebhookTest')
|
|
|
|
# ============================================================================
|
|
# WEBHOOK URL CONFIGURATION - Change this to test different endpoints
|
|
# ============================================================================
|
|
# Set to None to use URL from .env / config.yaml
|
|
# Set to a URL string to override config (for quick testing)
|
|
WEBHOOK_URL_OVERRIDE = None
|
|
|
|
# Example URLs:
|
|
# WEBHOOK_URL_OVERRIDE = "https://hook.us1.make.celonis.com/3f9ztwl8qnljufo0l65utfv5wvvnt9m5" # Production
|
|
# WEBHOOK_URL_OVERRIDE = "https://webhook.site/your-unique-id" # Webhook.site testing
|
|
# WEBHOOK_URL_OVERRIDE = "http://localhost:5555/webhook" # Local testing
|
|
# WEBHOOK_URL_OVERRIDE = "https://httpbin.org/post" # HTTPBin testing
|
|
# ============================================================================
|
|
|
|
def generate_sample_assets(count=3, campaign_number="C00000000700"):
|
|
"""
|
|
Generate sample processed assets for A1→A2 webhook
|
|
|
|
Args:
|
|
count: Number of sample assets to generate
|
|
campaign_number: Campaign number for naming
|
|
|
|
Returns:
|
|
list of asset dicts
|
|
"""
|
|
sample_assets = []
|
|
asset_names = [
|
|
"Hero_Image.jpg",
|
|
"Product_Shot.png",
|
|
"Lifestyle_Photo.jpg",
|
|
"Video_30sec.mp4",
|
|
"Banner_Ad.gif"
|
|
]
|
|
|
|
for i in range(count):
|
|
sample_assets.append({
|
|
'asset_id': "{}{}{}{}".format(
|
|
"A" * 8, "B" * 8, "C" * 8, str(i).zfill(8)
|
|
),
|
|
'asset_name': asset_names[i % len(asset_names)],
|
|
'tracking_id': "{}{}{}{}{}{}".format(
|
|
chr(65 + (i % 26)),
|
|
chr(97 + (i % 26)),
|
|
str(i % 10),
|
|
chr(65 + ((i + 1) % 26)),
|
|
chr(97 + ((i + 2) % 26)),
|
|
str((i + 3) % 10)
|
|
),
|
|
'box_file_id': str(1234567890 + i),
|
|
'box_url': "https://app.box.com/file/{}".format(1234567890 + i)
|
|
})
|
|
|
|
return sample_assets
|
|
|
|
def send_a1_to_a2_webhook(notifier, config, campaign_number, campaign_name, asset_count):
|
|
"""
|
|
Send A1→A2 webhook (Campaign Going Live)
|
|
|
|
Args:
|
|
notifier: Notifier instance
|
|
config: Configuration dict
|
|
campaign_number: Campaign number (e.g., C00000000700)
|
|
campaign_name: Campaign name
|
|
asset_count: Number of assets to include
|
|
|
|
Returns:
|
|
bool: Success status
|
|
"""
|
|
# Generate fake campaign ID (hex format like DAM uses)
|
|
campaign_id = "7C3F44AFD87849489D3F5DB0976BD03C"
|
|
|
|
# Generate sample assets
|
|
processed_assets = generate_sample_assets(asset_count, campaign_number)
|
|
|
|
# Build payload (matches a1_to_a2_download.py:293-304)
|
|
payload = {
|
|
'campaign_id': campaign_id,
|
|
'campaign_number': campaign_number,
|
|
'campaign_name': campaign_name,
|
|
'old_status': 'A1',
|
|
'new_status': 'A2',
|
|
'live_campaign': 'YES', # A1→A2 campaigns are going live
|
|
'asset_count': len(processed_assets),
|
|
'processed_assets': processed_assets,
|
|
'timestamp': int(time.time())
|
|
}
|
|
|
|
logger.info("=" * 80)
|
|
logger.info("A1→A2 WEBHOOK (Campaign Going Live)")
|
|
logger.info("=" * 80)
|
|
logger.info("")
|
|
# Determine webhook URL (override or config)
|
|
webhook_url = WEBHOOK_URL_OVERRIDE if WEBHOOK_URL_OVERRIDE else config['webhooks']['campaign_status_update']['url']
|
|
|
|
logger.info("Campaign Details:")
|
|
logger.info(" Number: {}".format(campaign_number))
|
|
logger.info(" Name: {}".format(campaign_name))
|
|
logger.info(" Status: A1 → A2")
|
|
logger.info(" Live: YES")
|
|
logger.info(" Asset Count: {}".format(len(processed_assets)))
|
|
logger.info("")
|
|
if WEBHOOK_URL_OVERRIDE:
|
|
logger.info("Webhook URL: {} (OVERRIDDEN - not using config)".format(webhook_url))
|
|
else:
|
|
logger.info("Webhook URL: {} (from config.yaml)".format(webhook_url))
|
|
logger.info("")
|
|
logger.info("Payload Preview:")
|
|
logger.info(json.dumps(payload, indent=2))
|
|
logger.info("")
|
|
logger.info("=" * 80)
|
|
logger.info("")
|
|
|
|
# Send webhook
|
|
logger.info("Sending webhook...")
|
|
result = notifier.send_webhook(
|
|
url=webhook_url,
|
|
payload=payload
|
|
)
|
|
|
|
if result:
|
|
logger.info("✓ Webhook sent successfully!")
|
|
return True
|
|
else:
|
|
logger.error("✗ Webhook send failed")
|
|
return False
|
|
|
|
def send_a4_webhook(notifier, config, campaign_number, campaign_name):
|
|
"""
|
|
Send A4 webhook (Campaign NOT Going Live)
|
|
|
|
Args:
|
|
notifier: Notifier instance
|
|
config: Configuration dict
|
|
campaign_number: Campaign number (e.g., C00000000700)
|
|
campaign_name: Campaign name
|
|
|
|
Returns:
|
|
bool: Success status
|
|
"""
|
|
# Generate fake campaign ID (hex format like DAM uses)
|
|
campaign_id = "7C3F44AFD87849489D3F5DB0976BD03C"
|
|
|
|
# Build payload (matches a4_webhook_monitor.py:86-94)
|
|
payload = {
|
|
'campaign_id': campaign_id,
|
|
'campaign_number': campaign_number,
|
|
'campaign_name': campaign_name,
|
|
'status': 'A4',
|
|
'live_campaign': 'NO', # A4 = Not going live
|
|
'timestamp': int(time.time()),
|
|
'message': 'Campaign marked A4 - Not going live'
|
|
}
|
|
|
|
# Determine webhook URL (override or config)
|
|
webhook_url = WEBHOOK_URL_OVERRIDE if WEBHOOK_URL_OVERRIDE else config['webhooks']['campaign_status_update']['url']
|
|
|
|
logger.info("=" * 80)
|
|
logger.info("A4 WEBHOOK (Campaign NOT Going Live)")
|
|
logger.info("=" * 80)
|
|
logger.info("")
|
|
logger.info("Campaign Details:")
|
|
logger.info(" Number: {}".format(campaign_number))
|
|
logger.info(" Name: {}".format(campaign_name))
|
|
logger.info(" Status: A4")
|
|
logger.info(" Live: NO")
|
|
logger.info("")
|
|
if WEBHOOK_URL_OVERRIDE:
|
|
logger.info("Webhook URL: {} (OVERRIDDEN - not using config)".format(webhook_url))
|
|
else:
|
|
logger.info("Webhook URL: {} (from config.yaml)".format(webhook_url))
|
|
logger.info("")
|
|
logger.info("Payload Preview:")
|
|
logger.info(json.dumps(payload, indent=2))
|
|
logger.info("")
|
|
logger.info("=" * 80)
|
|
logger.info("")
|
|
|
|
# Send webhook
|
|
logger.info("Sending webhook...")
|
|
result = notifier.send_webhook(
|
|
url=webhook_url,
|
|
payload=payload
|
|
)
|
|
|
|
if result:
|
|
logger.info("✓ Webhook sent successfully!")
|
|
return True
|
|
else:
|
|
logger.error("✗ Webhook send failed")
|
|
return False
|
|
|
|
def main():
|
|
"""Main entry point"""
|
|
parser = argparse.ArgumentParser(
|
|
description='Test webhook sender - Emulates A1→A2 and A4 webhooks',
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
Examples:
|
|
# Send A1→A2 webhook (going live) with default values
|
|
python scripts/test_webhook_sender.py --type a1a2
|
|
|
|
# Send A4 webhook (not going live) with custom campaign
|
|
python scripts/test_webhook_sender.py --type a4 --campaign C00000000999 --name "SUMMER_PROMO"
|
|
|
|
# Send A1→A2 with custom campaign and 5 assets
|
|
python scripts/test_webhook_sender.py --type a1a2 --campaign C00000000123 --name "KINDER_JOY" --assets 5
|
|
|
|
# Preview payload without sending
|
|
python scripts/test_webhook_sender.py --type a1a2 --preview-only
|
|
"""
|
|
)
|
|
|
|
parser.add_argument(
|
|
'--type',
|
|
required=True,
|
|
choices=['a1a2', 'a4'],
|
|
help='Webhook type: a1a2 (going live) or a4 (not going live)'
|
|
)
|
|
|
|
parser.add_argument(
|
|
'--campaign',
|
|
default='C00000000700',
|
|
help='Campaign number (default: C00000000700)'
|
|
)
|
|
|
|
parser.add_argument(
|
|
'--name',
|
|
default='THE_FINAL_NUTELLA',
|
|
help='Campaign name (default: THE_FINAL_NUTELLA)'
|
|
)
|
|
|
|
parser.add_argument(
|
|
'--assets',
|
|
type=int,
|
|
default=3,
|
|
help='Number of assets for A1→A2 webhook (default: 3, max: 5)'
|
|
)
|
|
|
|
parser.add_argument(
|
|
'--preview-only',
|
|
action='store_true',
|
|
help='Preview payload without sending webhook'
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Validate asset count
|
|
if args.assets < 1:
|
|
args.assets = 1
|
|
if args.assets > 5:
|
|
args.assets = 5
|
|
|
|
logger.info("=" * 80)
|
|
logger.info("WEBHOOK TEST SCRIPT")
|
|
logger.info("=" * 80)
|
|
logger.info("")
|
|
logger.info("Configuration:")
|
|
logger.info(" Type: {}".format("A1→A2 (Going Live)" if args.type == 'a1a2' else "A4 (Not Going Live)"))
|
|
logger.info(" Campaign: {}".format(args.campaign))
|
|
logger.info(" Name: {}".format(args.name))
|
|
if args.type == 'a1a2':
|
|
logger.info(" Assets: {}".format(args.assets))
|
|
logger.info(" Preview Only: {}".format("Yes" if args.preview_only else "No"))
|
|
logger.info("")
|
|
|
|
# Load configuration
|
|
try:
|
|
config = load_config('config/config.yaml')
|
|
except Exception as e:
|
|
logger.error("Failed to load config: {}".format(str(e)))
|
|
logger.error("Make sure you run this from Python-Version directory:")
|
|
logger.error(" cd Python-Version")
|
|
logger.error(" python scripts/test_webhook_sender.py --type a1a2")
|
|
sys.exit(1)
|
|
|
|
# Check if webhook is enabled
|
|
if not config['webhooks']['campaign_status_update']['enabled']:
|
|
logger.warning("⚠ WARNING: Webhooks are disabled in config.yaml")
|
|
logger.warning("Set webhooks.campaign_status_update.enabled = true to enable")
|
|
logger.warning("")
|
|
|
|
# Initialize notifier
|
|
notifier = Notifier(config)
|
|
|
|
# Preview only mode
|
|
if args.preview_only:
|
|
logger.info("PREVIEW ONLY MODE - Webhook will NOT be sent")
|
|
logger.info("")
|
|
|
|
# Send appropriate webhook
|
|
try:
|
|
if args.type == 'a1a2':
|
|
if args.preview_only:
|
|
# Just show the payload
|
|
campaign_id = "7C3F44AFD87849489D3F5DB0976BD03C"
|
|
processed_assets = generate_sample_assets(args.assets, args.campaign)
|
|
payload = {
|
|
'campaign_id': campaign_id,
|
|
'campaign_number': args.campaign,
|
|
'campaign_name': args.name,
|
|
'old_status': 'A1',
|
|
'new_status': 'A2',
|
|
'live_campaign': 'YES',
|
|
'asset_count': len(processed_assets),
|
|
'processed_assets': processed_assets,
|
|
'timestamp': int(time.time())
|
|
}
|
|
logger.info("A1→A2 Webhook Payload:")
|
|
logger.info(json.dumps(payload, indent=2))
|
|
success = True
|
|
else:
|
|
success = send_a1_to_a2_webhook(
|
|
notifier, config, args.campaign, args.name, args.assets
|
|
)
|
|
else: # a4
|
|
if args.preview_only:
|
|
# Just show the payload
|
|
campaign_id = "7C3F44AFD87849489D3F5DB0976BD03C"
|
|
payload = {
|
|
'campaign_id': campaign_id,
|
|
'campaign_number': args.campaign,
|
|
'campaign_name': args.name,
|
|
'status': 'A4',
|
|
'live_campaign': 'NO',
|
|
'timestamp': int(time.time()),
|
|
'message': 'Campaign marked A4 - Not going live'
|
|
}
|
|
logger.info("A4 Webhook Payload:")
|
|
logger.info(json.dumps(payload, indent=2))
|
|
success = True
|
|
else:
|
|
success = send_a4_webhook(
|
|
notifier, config, args.campaign, args.name
|
|
)
|
|
|
|
logger.info("")
|
|
logger.info("=" * 80)
|
|
if args.preview_only:
|
|
logger.info("PREVIEW COMPLETE")
|
|
elif success:
|
|
logger.info("✓ TEST SUCCESSFUL")
|
|
else:
|
|
logger.info("✗ TEST FAILED")
|
|
logger.info("=" * 80)
|
|
|
|
sys.exit(0 if success else 1)
|
|
|
|
except Exception as e:
|
|
logger.error("Error: {}".format(str(e)))
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1)
|
|
|
|
if __name__ == '__main__':
|
|
main()
|