Coverage for functions \ flipdare \ job \ cron_validator.py: 78%

1#!/usr/bin/env python 

2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved. 

3# 

4# This file is part of Flipdare's proprietary software and contains 

5# confidential and copyrighted material. Unauthorised copying, 

6# modification, distribution, or use of this file is strictly 

7# prohibited without prior written permission from Flipdare Pty Ltd. 

8# 

9# This software includes third-party components licensed under MIT, 

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details. 

11# 

12 

13 

14# NOTE: This is different from trigger validators as no model is supplied. 

15 

16from firebase_functions import scheduler_fn 

17 

18from flipdare.app_log import LOG 

19from flipdare.generated.shared.backend.app_job_type import AppJobType 

20 

21 

22class CronValidator: 

23 """Validator for scheduled (cron) events.""" 

24 

25 def __init__( 

26 self, 

27 job_type: AppJobType, 

28 event: scheduler_fn.ScheduledEvent, 

29 ) -> None: 

30 self._job_type = job_type 

31 self._event = event 

32 self._errors = self._validate() 

33 

34 @property 

35 def event(self) -> scheduler_fn.ScheduledEvent: 

36 return self._event 

37 

38 @property 

39 def job_type(self) -> AppJobType: 

40 return self._job_type 

41 

42 def valid(self) -> bool: 

43 return len(self._errors) == 0 

44 

45 def _validate(self) -> list[str]: 

46 job_name = self.event.job_name 

47 errors: list[str] = [] 

48 job_value = self.job_type.value 

49 

50 if job_name is None: 

51 msg = f"Scheduled event {job_value} is missing job name: {self.event!s}" 

52 LOG().error(msg) 

53 errors.append(msg) 

54 elif job_name != job_value: 

55 msg = f"Scheduled event job name '{job_name}' does not match expected '{job_value}'" 

56 LOG().error(msg) 

57 errors.append(msg) 

58 

59 return errors 

60 

61 @property 

62 def errors(self) -> list[str] | None: 

63 return self._errors if len(self._errors) > 0 else None 

64 

65 def error_str(self) -> str | None: 

66 if self.valid(): 

67 return None 

68 

69 error_str = "\n".join(self._errors) 

70 return f"Invalid scheduled event for {self.job_type.value}\n{error_str}"