test_pushover.py

@@ -0,0 +1,322 @@
+#!/usr/bin/env python3
+"""Test script to verify Pushover push notification connectivity.
+
+This script tests Pushover API connectivity and sends test notifications
+to help verify that the configuration is correct.
+"""
+
+import sys
+import time
+from datetime import datetime
+
+# Load configuration from config.yaml
+try:
+    from alpine_bits_python.config_loader import load_config
+    from pushover_complete import PushoverAPI
+
+    print("Loading configuration from config.yaml...")
+    config = load_config()
+    pushover_config = config.get("pushover", {})
+
+    USER_KEY = pushover_config.get("user_key", "")
+    API_TOKEN = pushover_config.get("api_token", "")
+
+    # Get monitoring configuration
+    monitoring_config = pushover_config.get("monitoring", {})
+    daily_report_config = monitoring_config.get("daily_report", {})
+    error_alerts_config = monitoring_config.get("error_alerts", {})
+
+    DAILY_REPORT_ENABLED = daily_report_config.get("enabled", False)
+    DAILY_REPORT_PRIORITY = daily_report_config.get("priority", 0)
+    ERROR_ALERTS_ENABLED = error_alerts_config.get("enabled", False)
+    ERROR_ALERTS_PRIORITY = error_alerts_config.get("priority", 1)
+
+    print(f"✓ Configuration loaded successfully")
+    print(f"  User Key: {USER_KEY[:10]}... (hidden)" if USER_KEY else "  User Key: (not set)")
+    print(f"  API Token: {API_TOKEN[:10]}... (hidden)" if API_TOKEN else "  API Token: (not set)")
+    print(f"  Daily Reports: {'Enabled' if DAILY_REPORT_ENABLED else 'Disabled'} (priority: {DAILY_REPORT_PRIORITY})")
+    print(f"  Error Alerts: {'Enabled' if ERROR_ALERTS_ENABLED else 'Disabled'} (priority: {ERROR_ALERTS_PRIORITY})")
+    print()
+
+    if not USER_KEY or not API_TOKEN:
+        print("✗ Pushover credentials not configured!")
+        print()
+        print("Please add the following to your secrets.yaml:")
+        print("  PUSHOVER_USER_KEY: your-user-key-here")
+        print("  PUSHOVER_API_TOKEN: your-app-token-here")
+        print()
+        print("Get your credentials from https://pushover.net")
+        sys.exit(1)
+
+except Exception as e:
+    print(f"✗ Failed to load configuration: {e}")
+    print()
+    print("Make sure you have:")
+    print("1. config.yaml with pushover section")
+    print("2. secrets.yaml with PUSHOVER_USER_KEY and PUSHOVER_API_TOKEN")
+    sys.exit(1)
+
+
+def test_simple_notification() -> bool:
+    """Test sending a simple notification."""
+    print("Test 1: Sending simple test notification...")
+    try:
+        api = PushoverAPI(API_TOKEN)
+        api.send_message(
+            user=USER_KEY,
+            title="Pushover Test",
+            message=f"Test notification from AlpineBits server at {datetime.now().strftime('%H:%M:%S')}",
+        )
+        print("✓ Simple notification sent successfully")
+        print("  Check your Pushover device for the notification!")
+        return True
+    except Exception as e:
+        print(f"✗ Failed to send notification: {e}")
+        return False
+
+
+def test_priority_levels() -> bool:
+    """Test different priority levels."""
+    print("\nTest 2: Testing priority levels...")
+
+    priorities = [
+        (-2, "Lowest", "No alert, quiet notification"),
+        (-1, "Low", "No alert"),
+        (0, "Normal", "Standard notification"),
+        (1, "High", "Bypasses quiet hours"),
+    ]
+
+    success_count = 0
+    for i, (priority, name, description) in enumerate(priorities):
+        try:
+            api = PushoverAPI(API_TOKEN)
+            api.send_message(
+                user=USER_KEY,
+                title=f"Priority Test: {name}",
+                message=f"Testing priority {priority} - {description}",
+                priority=priority,
+            )
+            print(f"✓ Sent notification with priority {priority} ({name})")
+            success_count += 1
+
+            # Add delay between notifications to avoid rate limiting (except after last one)
+            if i < len(priorities) - 1:
+                time.sleep(1)
+        except Exception as e:
+            print(f"✗ Failed to send priority {priority} notification: {e}")
+
+    print(f"  {success_count}/{len(priorities)} priority notifications sent")
+    return success_count == len(priorities)
+
+
+def test_daily_report_format() -> bool:
+    """Test sending a message formatted like a daily report."""
+    print("\nTest 3: Testing daily report format...")
+
+    # Sample stats similar to what the app would generate
+    date_str = datetime.now().strftime("%Y-%m-%d")
+    stats = {
+        "reporting_period": {
+            "start": "2025-10-15 08:00:00",
+            "end": "2025-10-16 08:00:00",
+        },
+        "total_reservations": 12,
+        "hotels": [
+            {"hotel_name": "Bemelmans Post", "reservations": 5},
+            {"hotel_name": "Jagthof Kaltern", "reservations": 4},
+            {"hotel_name": "Residence Erika", "reservations": 3},
+        ],
+    }
+
+    # Build message similar to pushover_service.py
+    lines = [f"Report for {date_str}", ""]
+
+    period = stats.get("reporting_period", {})
+    if period:
+        start = period.get("start", "").split(" ")[1] if " " in period.get("start", "") else ""
+        end = period.get("end", "").split(" ")[1] if " " in period.get("end", "") else ""
+        if start and end:
+            lines.append(f"Period: {start} - {end}")
+
+    total = stats.get("total_reservations", 0)
+    lines.append(f"Total Reservations: {total}")
+
+    hotels = stats.get("hotels", [])
+    if hotels:
+        lines.append("")
+        lines.append("By Hotel:")
+        for hotel in hotels[:5]:
+            hotel_name = hotel.get("hotel_name", "Unknown")
+            count = hotel.get("reservations", 0)
+            if len(hotel_name) > 20:
+                hotel_name = hotel_name[:17] + "..."
+            lines.append(f"  • {hotel_name}: {count}")
+
+    message = "\n".join(lines)
+
+    try:
+        api = PushoverAPI(API_TOKEN)
+        api.send_message(
+            user=USER_KEY,
+            title=f"AlpineBits Daily Report - {date_str}",
+            message=message,
+            priority=DAILY_REPORT_PRIORITY,
+        )
+        print("✓ Daily report format notification sent successfully")
+        print(f"  Message preview:\n{message}")
+        return True
+    except Exception as e:
+        print(f"✗ Failed to send daily report notification: {e}")
+        return False
+
+
+def test_error_alert_format() -> bool:
+    """Test sending a message formatted like an error alert."""
+    print("\nTest 4: Testing error alert format...")
+
+    error_count = 3
+    title = f"🚨 AlpineBits Error Alert: {error_count} errors"
+
+    message = f"""Error Alert - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
+
+Alert Type: Test Alert
+Error Count: {error_count}
+Time Range: 14:30:00 to 14:45:00
+
+Sample errors (see logs for details):
+1. Database connection timeout
+2. SMTP connection failed
+3. API rate limit exceeded"""
+
+    try:
+        api = PushoverAPI(API_TOKEN)
+        api.send_message(
+            user=USER_KEY,
+            title=title,
+            message=message,
+            priority=ERROR_ALERTS_PRIORITY,
+        )
+        print("✓ Error alert format notification sent successfully")
+        return True
+    except Exception as e:
+        print(f"✗ Failed to send error alert notification: {e}")
+        return False
+
+
+def test_with_url() -> bool:
+    """Test notification with supplementary URL."""
+    print("\nTest 5: Testing notification with URL...")
+
+    try:
+        api = PushoverAPI(API_TOKEN)
+        api.send_message(
+            user=USER_KEY,
+            title="AlpineBits Server",
+            message="This notification includes a supplementary URL. Tap to open.",
+            url="https://github.com/anthropics/claude-code",
+            url_title="View on GitHub",
+        )
+        print("✓ Notification with URL sent successfully")
+        return True
+    except Exception as e:
+        print(f"✗ Failed to send notification with URL: {e}")
+        return False
+
+
+def validate_credentials() -> bool:
+    """Validate Pushover credentials."""
+    print("Test 0: Validating Pushover credentials...")
+    try:
+        api = PushoverAPI(API_TOKEN)
+        # Try to validate the user key
+        response = api.send_message(
+            user=USER_KEY,
+            title="Credential Validation",
+            message="If you receive this, your Pushover credentials are valid!",
+        )
+        print("✓ Credentials validated successfully")
+        return True
+    except Exception as e:
+        print(f"✗ Credential validation failed: {e}")
+        print()
+        print("Possible issues:")
+        print("1. Invalid API token (check your application settings)")
+        print("2. Invalid user key (check your user dashboard)")
+        print("3. Network connectivity issues")
+        print("4. Pushover service is down")
+        return False
+
+
+def main():
+    """Run all Pushover tests."""
+    print("=" * 70)
+    print("Pushover Push Notification Test Script")
+    print("=" * 70)
+    print()
+
+    # First validate credentials
+    if not validate_credentials():
+        print("\n" + "=" * 70)
+        print("FAILED: Cannot proceed without valid credentials")
+        print("=" * 70)
+        return 1
+
+    print()
+
+    # Run all tests
+    tests = [
+        ("Simple Notification", test_simple_notification),
+        ("Priority Levels", test_priority_levels),
+        ("Daily Report Format", test_daily_report_format),
+        ("Error Alert Format", test_error_alert_format),
+        ("Notification with URL", test_with_url),
+    ]
+
+    results = []
+    for i, (test_name, test_func) in enumerate(tests):
+        try:
+            result = test_func()
+            results.append((test_name, result))
+
+            # Add delay between tests to avoid rate limiting (except after last one)
+            if i < len(tests) - 1:
+                print("  (Waiting 1 second to avoid rate limiting...)")
+                time.sleep(1)
+        except Exception as e:
+            print(f"✗ Test '{test_name}' crashed: {e}")
+            results.append((test_name, False))
+
+    # Print summary
+    print("\n" + "=" * 70)
+    print("TEST SUMMARY")
+    print("=" * 70)
+
+    passed = sum(1 for _, result in results if result)
+    total = len(results)
+
+    for test_name, result in results:
+        status = "✓ PASS" if result else "✗ FAIL"
+        print(f"  {status}: {test_name}")
+
+    print()
+    print(f"Results: {passed}/{total} tests passed")
+
+    if passed == total:
+        print("\n✓ ALL TESTS PASSED!")
+        print("=" * 70)
+        print("\nYour Pushover configuration is working correctly.")
+        print("Check your Pushover device for all the test notifications.")
+        return 0
+    else:
+        print(f"\n⚠ {total - passed} TEST(S) FAILED")
+        print("=" * 70)
+        print("\nSome tests failed. Check the output above for details.")
+        return 1
+
+
+if __name__ == "__main__":
+    try:
+        sys.exit(main())
+    except KeyboardInterrupt:
+        print("\n\nTest cancelled by user")
+        sys.exit(1)