Coverage for app / services / alert_service.py: 100%

1import smtplib 

2import logging 

3from email.mime.text import MIMEText 

4from datetime import datetime, timezone 

5from typing import Optional 

6 

7from sqlalchemy.orm import Session 

8 

9from app.core.config import settings 

10from app.models.monitor import Monitor 

11 

12logger = logging.getLogger(__name__) 

13 

14class AlertService: 

15 """Handles alert state transitions and email dispatching.""" 

16 

17 def __init__(self, db: Session): 

18 self.db = db 

19 

20 def should_alert(self, monitor: Monitor, is_up: bool) -> Optional[str]: 

21 """ 

22 Determine if an alert should be sent based on state transition. 

23 Returns the new alert_status if an alert should fire, else None. 

24 """ 

25 if not is_up and monitor.alert_status == "UP": 

26 return "DOWN" 

27 elif is_up and monitor.alert_status == "DOWN": 

28 return "UP" 

29 return None 

30 

31 def update_monitor_state(self, monitor: Monitor, new_status: str) -> None: 

32 """Update monitor alert state and timestamp.""" 

33 monitor.alert_status = new_status 

34 monitor.last_alerted_at = datetime.now(timezone.utc) 

35 self.db.commit() 

36 

37 def send_email( 

38 self, 

39 monitor_id: str, 

40 url: str, 

41 recipient: str, 

42 status: str, 

43 ) -> bool: 

44 """Send alert email via SMTP. Returns success boolean.""" 

45 subject = f"[ALERT] {url} is {status}" 

46 body = ( 

47 f"Monitor Alert\n" 

48 f"URL: {url}\n" 

49 f"Status: {status}\n" 

50 f"Monitor ID: {monitor_id}\n" 

51 f"Time: {datetime.now(timezone.utc).isoformat()}" 

52 ) 

53 

54 msg = MIMEText(body) 

55 msg["Subject"] = subject 

56 msg["From"] = settings.from_email 

57 msg["To"] = recipient 

58 

59 try: 

60 with smtplib.SMTP(settings.smtp_host, settings.smtp_port) as server: 

61 if settings.smtp_user and settings.smtp_pass: 

62 server.login(settings.smtp_user, settings.smtp_pass) 

63 server.send_message(msg) 

64 return True 

65 except Exception as e: 

66 # Log structured error instead of a silent print 

67 logger.error(f"[AlertService] Failed to send email for monitor {monitor_id}: {e}", exc_info=True) 

68 return False 

69 

70 def get_recipient(self, monitor: Monitor) -> str: 

71 """Get alert recipient email for a monitor.""" 

72 if monitor.user and monitor.user.email: 

73 return monitor.user.email 

74 return "admin@example.com" 

75 

76 def process_ping_result( 

77 self, 

78 monitor_id: str, 

79 is_up: bool, 

80 status_code: Optional[int], 

81 response_ms: Optional[int], 

82 error_message: Optional[str], 

83 ) -> Optional[str]: 

84 """ 

85 Full pipeline: record ping, check state transition, send alert if needed. 

86 Returns the alert status sent (UP/DOWN) or None. 

87 """ 

88 from app.models.ping_log import PingLog 

89 

90 # Record the ping 

91 ping = PingLog( 

92 monitor_id=monitor_id, 

93 status_code=status_code, 

94 response_ms=response_ms, 

95 is_up=is_up, 

96 error_message=error_message, 

97 ) 

98 self.db.add(ping) 

99 self.db.flush() 

100 

101 # Get monitor with user relationship 

102 monitor = self.db.get(Monitor, monitor_id) 

103 if not monitor: 

104 self.db.commit() 

105 return None 

106 

107 new_status = self.should_alert(monitor, is_up) 

108 if new_status: 

109 self.update_monitor_state(monitor, new_status) 

110 recipient = self.get_recipient(monitor) 

111 self.send_email( 

112 str(monitor.id), 

113 str(monitor.url), 

114 recipient, 

115 new_status, 

116 ) 

117 return new_status 

118 else: 

119 self.db.commit() 

120 return None 

121 

122def send_alert_email(monitor_id: str, url: str, recipient: str, status: str) -> bool: 

123 """ 

124 Standalone function for Celery tasks. 

125 Creates a temporary session and sends email. 

126 """ 

127 from app.db.session import SyncSessionLocal 

128 

129 with SyncSessionLocal() as db: 

130 service = AlertService(db) 

131 return service.send_email(monitor_id, url, recipient, status)