Coverage for crons/core_dynamic_crons.py: 73%

1from apscheduler.triggers.cron import CronTrigger 

2from apscheduler.triggers.interval import IntervalTrigger 

3 

4from chalicelib.core import telemetry 

5from chalicelib.core import weekly_report, jobs, health 

6 

7 

8async def run_scheduled_jobs() -> None: 

9 jobs.execute_jobs() 

10 

11 

12async def weekly_report_cron() -> None: 

13 weekly_report.cron() 

14 

15 

16async def telemetry_cron() -> None: 

17 telemetry.compute() 

18 

19 

20async def health_cron() -> None: 

21 health.cron() 

22 

23 

24async def weekly_health_cron() -> None: 

25 health.weekly_cron() 

26 

27 

28cron_jobs = [ 

29 {"func": telemetry_cron, "trigger": CronTrigger(day_of_week="*"), 

30 "misfire_grace_time": 60 * 60, "max_instances": 1}, 

31 {"func": run_scheduled_jobs, "trigger": CronTrigger(day_of_week="*", hour=0, minute=15), 

32 "misfire_grace_time": 20, "max_instances": 1}, 

33 {"func": weekly_report_cron, "trigger": CronTrigger(day_of_week="mon", hour=5), 

34 "misfire_grace_time": 60 * 60, "max_instances": 1}, 

35 {"func": health_cron, "trigger": IntervalTrigger(hours=0, minutes=30, start_date="2023-04-01 0:0:0", jitter=300), 

36 "misfire_grace_time": 60 * 60, "max_instances": 1}, 

37 {"func": weekly_health_cron, "trigger": CronTrigger(day_of_week="sun", hour=5), 

38 "misfire_grace_time": 60 * 60, "max_instances": 1} 

39]