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

1import httpx 

2from datetime import datetime, timezone 

3from typing import Optional, Tuple 

4 

5from app.services.alert_service import AlertService 

6 

7 

8class PingService: 

9 """Handles HTTP health checks and result processing.""" 

10 

11 DEFAULT_TIMEOUT = 10.0 

12 

13 # Browser-like headers to avoid bot detection 

14 _HEADERS = { 

15 "User-Agent": ( 

16 "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " 

17 "AppleWebKit/537.36 (KHTML, like Gecko) " 

18 "Chrome/124.0.0.0 Safari/537.36" 

19 ), 

20 "Accept": ( 

21 "text/html,application/xhtml+xml,application/xml;" 

22 "q=0.9,image/avif,image/webp,*/*;q=0.8" 

23 ), 

24 "Accept-Language": "en-US,en;q=0.5", 

25 "Accept-Encoding": "gzip, deflate, br", 

26 "Sec-Fetch-Dest": "document", 

27 "Sec-Fetch-Mode": "navigate", 

28 "Sec-Fetch-Site": "none", 

29 "Sec-Fetch-User": "?1", 

30 "Upgrade-Insecure-Requests": "1", 

31 } 

32 

33 def __init__(self, alert_service: Optional[AlertService] = None): 

34 self.alert_service = alert_service 

35 

36 def check_url(self, url: str) -> Tuple[bool, Optional[int], Optional[int], Optional[str]]: 

37 """ 

38 Perform HTTP GET against URL. 

39 Returns: (is_up, status_code, response_ms, error_message) 

40 """ 

41 start = datetime.now(timezone.utc) 

42 is_up = False 

43 status_code = None 

44 response_ms = None 

45 error_message = None 

46 

47 try: 

48 with httpx.Client( 

49 timeout=self.DEFAULT_TIMEOUT, 

50 follow_redirects=True, 

51 headers=self._HEADERS, 

52 ) as client: 

53 response = client.get(url) 

54 status_code = response.status_code 

55 is_up = 200 <= status_code < 400 

56 response_ms = int( 

57 (datetime.now(timezone.utc) - start).total_seconds() * 1000 

58 ) 

59 except httpx.TimeoutException: 

60 error_message = "Request timed out" 

61 is_up = False 

62 except httpx.ConnectError: 

63 error_message = "Connection failed" 

64 is_up = False 

65 except Exception as e: 

66 error_message = str(e) 

67 is_up = False 

68 

69 return is_up, status_code, response_ms, error_message 

70 

71 def process_monitor_check(self, monitor_id: str, url: str, db) -> dict: 

72 """ 

73 Full check pipeline: ping URL, record result, handle alerts. 

74 Returns result summary dict. 

75 """ 

76 is_up, status_code, response_ms, error_message = self.check_url(url) 

77 

78 if self.alert_service: 

79 alert_status = self.alert_service.process_ping_result( 

80 monitor_id=monitor_id, 

81 is_up=is_up, 

82 status_code=status_code, 

83 response_ms=response_ms, 

84 error_message=error_message, 

85 ) 

86 else: 

87 # Fallback: just record ping without alerts 

88 from app.models.ping_log import PingLog 

89 ping = PingLog( 

90 monitor_id=monitor_id, 

91 status_code=status_code, 

92 response_ms=response_ms, 

93 is_up=is_up, 

94 error_message=error_message, 

95 ) 

96 db.add(ping) 

97 db.commit() 

98 alert_status = None 

99 

100 return { 

101 "monitor_id": monitor_id, 

102 "is_up": is_up, 

103 "status_code": status_code, 

104 "response_ms": response_ms, 

105 "error_message": error_message, 

106 "alert_sent": alert_status, 

107 }