Coverage for functions \ flipdare \ mailer \ admin \ cron_email.py: 93%

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 

13from typing import override 

14from flipdare.mailer.app_email_params import AppEmailParams 

15from flipdare.mailer.app_email_type import AppEmailType 

16from flipdare.generated.schema.email.subject.admin.cron_subject_schema import CronSubjectSchema 

17from flipdare.result.outcome import Outcome 

18from flipdare.mailer._jinja_email_template import JinjaEmailTemplate 

19from flipdare.generated.schema.email.body.admin.cron_email_schema import CronEmailSchema 

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

21 

22__all__ = ["CronEmail"] 

23 

24 

25class CronEmail(JinjaEmailTemplate[CronEmailSchema]): 

26 SCHEMA_CLASS = CronEmailSchema 

27 

28 def __init__( 

29 self, 

30 cron_date: str, 

31 job_type: AppJobType, 

32 message: str, 

33 priority: str, 

34 errors: list[str] | None = None, 

35 html: str | None = None, 

36 text: str | None = None, 

37 ) -> None: 

38 data = CronEmailSchema( 

39 cron_date=cron_date, 

40 job_type=job_type, 

41 message=message, 

42 priority=priority, 

43 ) 

44 

45 outcome = Outcome.OK 

46 if errors is not None: 

47 data["error"] = "\n".join(errors) 

48 outcome = Outcome.ERROR 

49 if html is not None: 

50 data["html_report"] = html 

51 if text is not None: 

52 data["text_report"] = text 

53 

54 super().__init__( 

55 data=data, 

56 params=AppEmailParams[CronSubjectSchema]( 

57 email_type=AppEmailType.ADM_CRON_REPORT, 

58 schema=CronSubjectSchema( 

59 job_type=job_type.short_description, 

60 cron_date=cron_date, 

61 outcome=outcome.label, 

62 ), 

63 ), 

64 ) 

65 

66 @override 

67 def newline_fields(self) -> list[str]: 

68 return ["message", "error"] 

69 

70 @property 

71 @override 

72 def data(self) -> CronEmailSchema: 

73 assert isinstance(self._data, dict) # narrowing, we known we have a dict.. 

74 return self._data