Coverage for app / tasks / ping.py: 100%

1from datetime import datetime, timezone 

2from celery_worker import celery_app 

3from app.db.session import SyncSessionLocal 

4from app.services.ping_service import PingService 

5from app.services.alert_service import AlertService 

6 

7@celery_app.task(bind=True, max_retries=3) 

8def ping_url(self, monitor_id: str, url: str): 

9 with SyncSessionLocal() as db: 

10 alert_service = AlertService(db) 

11 ping_service = PingService(alert_service) 

12 

13 result = ping_service.process_monitor_check(monitor_id, url, db) 

14 

15 # NOTE: next_check_at is now set by the scheduler at dispatch time, 

16 # not by the worker at completion time. This eliminates interval drift. 

17 

18 return result