added portals, quest tracking, discord monitor etc etc

discord-rare-monitor/test_websocket.py

@@ -0,0 +1,169 @@
+#!/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)