169 lines
No EOL
6 KiB
Python
169 lines
No EOL
6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script for Discord Rare Monitor Bot WebSocket functionality.
|
|
Tests WebSocket connection and message parsing without requiring Discord token.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import sys
|
|
from datetime import datetime
|
|
|
|
import websockets
|
|
|
|
# Configure logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
|
handlers=[logging.StreamHandler(sys.stdout)]
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Test configuration
|
|
WEBSOCKET_URL = "ws://localhost:8765/ws/position" # Test against local instance
|
|
COMMON_RARE_KEYWORDS = ["Crystal", "Jewel", "Pearl", "Elixir", "Kit"]
|
|
|
|
|
|
def classify_rare(rare_name: str) -> str:
|
|
"""Classify rare as 'common' or 'great' based on keywords."""
|
|
if any(keyword in rare_name for keyword in COMMON_RARE_KEYWORDS) and "Frore Crystal" not in rare_name:
|
|
return "common"
|
|
else:
|
|
return "great"
|
|
|
|
|
|
async def test_websocket_connection(duration: int = 30):
|
|
"""Test WebSocket connection and message processing for specified duration."""
|
|
logger.info(f"🔗 Testing WebSocket connection to: {WEBSOCKET_URL}")
|
|
logger.info(f"⏱️ Test duration: {duration} seconds")
|
|
|
|
message_count = 0
|
|
rare_count = 0
|
|
start_time = asyncio.get_event_loop().time()
|
|
|
|
try:
|
|
async with websockets.connect(WEBSOCKET_URL) as websocket:
|
|
logger.info("✅ WebSocket connected successfully")
|
|
|
|
while True:
|
|
try:
|
|
# Check if test duration has elapsed
|
|
current_time = asyncio.get_event_loop().time()
|
|
if current_time - start_time >= duration:
|
|
logger.info(f"⏰ Test duration of {duration} seconds completed")
|
|
break
|
|
|
|
# Receive message with timeout
|
|
message = await asyncio.wait_for(websocket.recv(), timeout=1.0)
|
|
message_count += 1
|
|
|
|
# Parse JSON
|
|
try:
|
|
data = json.loads(message)
|
|
msg_type = data.get('type', 'unknown')
|
|
|
|
# Log message types (for first few messages)
|
|
if message_count <= 10:
|
|
logger.info(f"📨 Message {message_count}: type={msg_type}")
|
|
|
|
# Process rare events
|
|
if msg_type == 'rare':
|
|
rare_count += 1
|
|
rare_name = data.get('name', 'Unknown Rare')
|
|
character_name = data.get('character_name', 'Unknown Character')
|
|
rare_type = classify_rare(rare_name)
|
|
|
|
logger.info(f"🎯 RARE DISCOVERED: {rare_name} by {character_name} ({rare_type})")
|
|
logger.info(f" Location: {data.get('ew', 'N/A')}E, {data.get('ns', 'N/A')}N")
|
|
logger.info(f" Timestamp: {data.get('timestamp', 'N/A')}")
|
|
|
|
except json.JSONDecodeError:
|
|
# Ignore invalid JSON
|
|
continue
|
|
|
|
except asyncio.TimeoutError:
|
|
# No message received in timeout period, continue
|
|
continue
|
|
|
|
logger.info(f"📊 Test Summary:")
|
|
logger.info(f" Total messages received: {message_count}")
|
|
logger.info(f" Rare events detected: {rare_count}")
|
|
logger.info(f" Test duration: {duration} seconds")
|
|
|
|
except websockets.exceptions.ConnectionClosed:
|
|
logger.warning("⚠️ WebSocket connection closed")
|
|
return False
|
|
except ConnectionRefusedError:
|
|
logger.error("❌ Connection refused - is Dereth Tracker running?")
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"❌ WebSocket error: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
async def test_rare_classification():
|
|
"""Test rare classification logic."""
|
|
logger.info("🧪 Testing rare classification logic...")
|
|
|
|
test_cases = [
|
|
("Dark Heart", "great"),
|
|
("Pyreal Crystal", "common"),
|
|
("Frore Crystal", "great"), # Special case - should be great despite having "Crystal"
|
|
("Ruby Jewel", "common"),
|
|
("Ancient Relic", "great"),
|
|
("Mana Elixir", "common"),
|
|
("Health Kit", "common"),
|
|
("Sunstone", "great")
|
|
]
|
|
|
|
for rare_name, expected in test_cases:
|
|
result = classify_rare(rare_name)
|
|
status = "✅" if result == expected else "❌"
|
|
logger.info(f" {status} {rare_name}: {result} (expected {expected})")
|
|
|
|
|
|
def create_test_rare_message(name: str, character: str) -> str:
|
|
"""Create a test rare message for testing."""
|
|
return json.dumps({
|
|
"type": "rare",
|
|
"character_name": character,
|
|
"name": name,
|
|
"timestamp": datetime.now().isoformat() + "Z",
|
|
"ew": 12.34,
|
|
"ns": -56.78,
|
|
"z": 10.5
|
|
})
|
|
|
|
|
|
async def main():
|
|
"""Main test function."""
|
|
logger.info("🚀 Starting Discord Rare Monitor Bot WebSocket Tests")
|
|
|
|
# Test 1: Rare classification logic
|
|
await test_rare_classification()
|
|
|
|
# Test 2: WebSocket connection (if available)
|
|
logger.info("\n" + "="*50)
|
|
logger.info("Testing WebSocket connection...")
|
|
|
|
success = await test_websocket_connection(duration=30)
|
|
|
|
if success:
|
|
logger.info("✅ All tests completed successfully!")
|
|
else:
|
|
logger.warning("⚠️ WebSocket test failed - this is expected if Dereth Tracker is not running")
|
|
|
|
logger.info("🏁 Test suite complete")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
asyncio.run(main())
|
|
except KeyboardInterrupt:
|
|
logger.info("🛑 Test interrupted by user")
|
|
except Exception as e:
|
|
logger.error(f"❌ Test failed with error: {e}")
|
|
sys.exit(1) |