Coverage for functions \ flipdare \ task \ report \ core \ cron_table_report.py: 100%

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 

14from collections.abc import Sequence 

15from typing import Any, override 

16from flipdare.app_log import LOG 

17from flipdare.backend.app_logger import AppLogger 

18from flipdare.constants import IS_DEBUG 

19from flipdare.mailer.admin_mailer import AdminMailer 

20from flipdare.generated.schema.report.id_report_schema import IdReportSchema 

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

22from flipdare.service.core.cron_processor import CronResultEntry 

23from flipdare.task.report.core.table_report import TableReport 

24 

25 

26class CronTableReport(TableReport[IdReportSchema]): 

27 __slots__ = ("_processed",) 

28 

29 def __init__( 

30 self, 

31 job_type: AppJobType, 

32 processed: Sequence[CronResultEntry], 

33 app_logger: AppLogger, 

34 mailer: AdminMailer, 

35 ) -> None: 

36 super().__init__( 

37 job_type=job_type, 

38 schema_cls=IdReportSchema, 

39 process_fn=self._id_process, 

40 app_logger=app_logger, 

41 mailer=mailer, 

42 ) 

43 self._processed = processed 

44 

45 def _id_process(self, entry: CronResultEntry) -> IdReportSchema: 

46 return IdReportSchema( 

47 { 

48 "doc_id": entry.doc_id, 

49 "outcome": entry.outcome.value, 

50 "message": entry.message, 

51 } 

52 ) 

53 

54 @override 

55 def table_data(self) -> Sequence[Any] | None: 

56 processed = self._processed 

57 if not processed or len(processed) == 0: 

58 # return None, so an error is triggered and and email is sent to admin. 

59 msg = f"CronResultReport {self.job_type.value} has no results to report" 

60 self.add_error(msg) 

61 return None 

62 

63 # sort cronReportEntry by outcome, then doc_id for easier reading 

64 sorted_entries = sorted( 

65 self._processed, 

66 key=lambda e: (e.outcome.value, e.doc_id), 

67 ) 

68 if IS_DEBUG: 

69 LOG().debug( 

70 f"CronResultReport {self.job_type.value} generated {len(sorted_entries)} results" 

71 ) 

72 

73 return sorted_entries