#!/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()