ManualTrader/tests/test_alarm_system.py

#!/usr/bin/env python3
"""
Test suite for the Price Alarm System

Tests alarm creation, management, triggering, and integration.
"""

import unittest
import tempfile
import os
from datetime import datetime
from alarm_manager import AlarmManager


class TestAlarmManager(unittest.TestCase):
    """Test the AlarmManager class."""
    
    def setUp(self):
        """Set up test fixtures."""
        # Use temporary file for testing
        self.temp_file = tempfile.NamedTemporaryFile(delete=False, suffix='.json')
        self.temp_file.close()
        self.alarm_manager = AlarmManager(self.temp_file.name)
    
    def tearDown(self):
        """Clean up after tests."""
        # Remove temporary file
        if os.path.exists(self.temp_file.name):
            os.unlink(self.temp_file.name)
    
    def test_create_alarm_above(self):
        """Test creating an alarm for price going above current price."""
        # Current price $50, target $55 (above)
        alarm = self.alarm_manager.create_alarm("BTC", 55.0, 50.0)
        
        self.assertEqual(alarm['id'], 1)
        self.assertEqual(alarm['token'], 'BTC')
        self.assertEqual(alarm['target_price'], 55.0)
        self.assertEqual(alarm['current_price_when_set'], 50.0)
        self.assertEqual(alarm['direction'], 'above')
        self.assertEqual(alarm['status'], 'active')
    
    def test_create_alarm_below(self):
        """Test creating an alarm for price going below current price."""
        # Current price $50, target $45 (below)
        alarm = self.alarm_manager.create_alarm("ETH", 45.0, 50.0)
        
        self.assertEqual(alarm['id'], 1)
        self.assertEqual(alarm['token'], 'ETH')
        self.assertEqual(alarm['target_price'], 45.0)
        self.assertEqual(alarm['current_price_when_set'], 50.0)
        self.assertEqual(alarm['direction'], 'below')
        self.assertEqual(alarm['status'], 'active')
    
    def test_multiple_alarms_increment_id(self):
        """Test that multiple alarms get incrementing IDs."""
        alarm1 = self.alarm_manager.create_alarm("BTC", 55.0, 50.0)
        alarm2 = self.alarm_manager.create_alarm("ETH", 3500.0, 3000.0)
        alarm3 = self.alarm_manager.create_alarm("BTC", 45.0, 50.0)
        
        self.assertEqual(alarm1['id'], 1)
        self.assertEqual(alarm2['id'], 2)
        self.assertEqual(alarm3['id'], 3)
    
    def test_get_alarms_by_token(self):
        """Test filtering alarms by token."""
        self.alarm_manager.create_alarm("BTC", 55.0, 50.0)
        self.alarm_manager.create_alarm("ETH", 3500.0, 3000.0)
        self.alarm_manager.create_alarm("BTC", 45.0, 50.0)
        
        btc_alarms = self.alarm_manager.get_alarms_by_token("BTC")
        eth_alarms = self.alarm_manager.get_alarms_by_token("ETH")
        
        self.assertEqual(len(btc_alarms), 2)
        self.assertEqual(len(eth_alarms), 1)
        self.assertEqual(btc_alarms[0]['token'], 'BTC')
        self.assertEqual(eth_alarms[0]['token'], 'ETH')
    
    def test_remove_alarm(self):
        """Test removing an alarm by ID."""
        alarm = self.alarm_manager.create_alarm("BTC", 55.0, 50.0)
        alarm_id = alarm['id']
        
        # Verify alarm exists
        self.assertIsNotNone(self.alarm_manager.get_alarm_by_id(alarm_id))
        
        # Remove alarm
        result = self.alarm_manager.remove_alarm(alarm_id)
        self.assertTrue(result)
        
        # Verify alarm is gone
        self.assertIsNone(self.alarm_manager.get_alarm_by_id(alarm_id))
    
    def test_remove_nonexistent_alarm(self):
        """Test removing an alarm that doesn't exist."""
        result = self.alarm_manager.remove_alarm(999)
        self.assertFalse(result)
    
    def test_check_alarms_trigger_above(self):
        """Test triggering an alarm when price goes above target."""
        # Create alarm: trigger when BTC goes above $55
        self.alarm_manager.create_alarm("BTC", 55.0, 50.0)
        
        # Price data: BTC is now $56 (above $55)
        price_data = {"BTC": 56.0}
        triggered = self.alarm_manager.check_alarms(price_data)
        
        self.assertEqual(len(triggered), 1)
        self.assertEqual(triggered[0]['token'], 'BTC')
        self.assertEqual(triggered[0]['triggered_price'], 56.0)
        self.assertEqual(triggered[0]['status'], 'triggered')
    
    def test_check_alarms_trigger_below(self):
        """Test triggering an alarm when price goes below target."""
        # Create alarm: trigger when ETH goes below $45
        self.alarm_manager.create_alarm("ETH", 45.0, 50.0)
        
        # Price data: ETH is now $44 (below $45)
        price_data = {"ETH": 44.0}
        triggered = self.alarm_manager.check_alarms(price_data)
        
        self.assertEqual(len(triggered), 1)
        self.assertEqual(triggered[0]['token'], 'ETH')
        self.assertEqual(triggered[0]['triggered_price'], 44.0)
        self.assertEqual(triggered[0]['status'], 'triggered')
    
    def test_check_alarms_no_trigger(self):
        """Test that alarms don't trigger when conditions aren't met."""
        # Create alarm: trigger when BTC goes above $55
        self.alarm_manager.create_alarm("BTC", 55.0, 50.0)
        
        # Price data: BTC is now $54 (still below $55)
        price_data = {"BTC": 54.0}
        triggered = self.alarm_manager.check_alarms(price_data)
        
        self.assertEqual(len(triggered), 0)
        
        # Verify alarm is still active
        active_alarms = self.alarm_manager.get_all_active_alarms()
        self.assertEqual(len(active_alarms), 1)
        self.assertEqual(active_alarms[0]['status'], 'active')
    
    def test_check_alarms_exact_price(self):
        """Test that alarms trigger at exact target price."""
        # Create alarm: trigger when BTC goes above $55
        self.alarm_manager.create_alarm("BTC", 55.0, 50.0)
        
        # Price data: BTC is exactly $55
        price_data = {"BTC": 55.0}
        triggered = self.alarm_manager.check_alarms(price_data)
        
        self.assertEqual(len(triggered), 1)
        self.assertEqual(triggered[0]['triggered_price'], 55.0)
    
    def test_alarm_only_triggers_once(self):
        """Test that triggered alarms don't trigger again."""
        # Create alarm
        self.alarm_manager.create_alarm("BTC", 55.0, 50.0)
        
        # Trigger alarm
        price_data = {"BTC": 56.0}
        triggered1 = self.alarm_manager.check_alarms(price_data)
        self.assertEqual(len(triggered1), 1)
        
        # Try to trigger again with even higher price
        price_data = {"BTC": 60.0}
        triggered2 = self.alarm_manager.check_alarms(price_data)
        self.assertEqual(len(triggered2), 0)  # Should not trigger again
    
    def test_multiple_alarms_selective_trigger(self):
        """Test that only applicable alarms trigger."""
        # Create multiple alarms
        self.alarm_manager.create_alarm("BTC", 55.0, 50.0)  # Above $55
        self.alarm_manager.create_alarm("ETH", 45.0, 50.0)  # Below $45
        self.alarm_manager.create_alarm("BTC", 40.0, 50.0)  # Below $40
        
        # Price data: BTC $56 (triggers first), ETH $46 (doesn't trigger)
        price_data = {"BTC": 56.0, "ETH": 46.0}
        triggered = self.alarm_manager.check_alarms(price_data)
        
        self.assertEqual(len(triggered), 1)
        self.assertEqual(triggered[0]['token'], 'BTC')
        self.assertEqual(triggered[0]['target_price'], 55.0)
    
    def test_persistence(self):
        """Test that alarms persist across manager instances."""
        # Create alarm
        alarm = self.alarm_manager.create_alarm("BTC", 55.0, 50.0)
        alarm_id = alarm['id']
        
        # Create new manager instance with same file
        new_manager = AlarmManager(self.temp_file.name)
        
        # Check that alarm persists
        restored_alarm = new_manager.get_alarm_by_id(alarm_id)
        self.assertIsNotNone(restored_alarm)
        self.assertEqual(restored_alarm['token'], 'BTC')
        self.assertEqual(restored_alarm['target_price'], 55.0)
        self.assertEqual(restored_alarm['status'], 'active')
    
    def test_format_alarm_list_empty(self):
        """Test formatting empty alarm list."""
        message = self.alarm_manager.format_alarm_list([])
        self.assertIn("No alarms found", message)
    
    def test_format_alarm_list_with_alarms(self):
        """Test formatting alarm list with alarms."""
        self.alarm_manager.create_alarm("BTC", 55.0, 50.0)
        self.alarm_manager.create_alarm("ETH", 45.0, 50.0)
        
        alarms = self.alarm_manager.get_all_active_alarms()
        message = self.alarm_manager.format_alarm_list(alarms)
        
        self.assertIn("BTC", message)
        self.assertIn("ETH", message)
        self.assertIn("ID 1", message)
        self.assertIn("ID 2", message)
        self.assertIn("55.00", message)
        self.assertIn("45.00", message)
    
    def test_format_triggered_alarm(self):
        """Test formatting triggered alarm notification."""
        # Create and trigger alarm
        self.alarm_manager.create_alarm("BTC", 55.0, 50.0)
        price_data = {"BTC": 56.0}
        triggered = self.alarm_manager.check_alarms(price_data)
        
        message = self.alarm_manager.format_triggered_alarm(triggered[0])
        
        self.assertIn("Price Alert Triggered", message)
        self.assertIn("BTC", message)
        self.assertIn("56.00", message)
        self.assertIn("55.00", message)
        self.assertIn("50.00", message)
    
    def test_statistics(self):
        """Test alarm statistics."""
        # Create some alarms
        self.alarm_manager.create_alarm("BTC", 55.0, 50.0)
        self.alarm_manager.create_alarm("BTC", 45.0, 50.0)
        self.alarm_manager.create_alarm("ETH", 3500.0, 3000.0)
        
        # Trigger one alarm
        price_data = {"BTC": 56.0}
        self.alarm_manager.check_alarms(price_data)
        
        stats = self.alarm_manager.get_statistics()
        
        self.assertEqual(stats['total_active'], 2)  # 2 still active
        self.assertEqual(stats['total_triggered'], 1)  # 1 triggered
        self.assertEqual(stats['tokens_tracked'], 2)  # BTC and ETH
        self.assertEqual(stats['token_breakdown']['BTC'], 1)  # 1 BTC alarm left
        self.assertEqual(stats['token_breakdown']['ETH'], 1)  # 1 ETH alarm


if __name__ == '__main__':
    unittest.main()