Coverage for functions \ flipdare \ task \ report \ log_reporter.py: 63%

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 __future__ import annotations 

14 

15from typing import TYPE_CHECKING 

16from flipdare.service._service_provider import ServiceProvider 

17from flipdare.task.report.core.query_report import QueryReport 

18from flipdare.core.app_backend_link import AppBackendLink 

19from flipdare.result.output_result import OutputResult 

20from flipdare.generated.schema.report.log_report_schema import LogReportSchema 

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

22from flipdare.util.debug_util import stringify_debug 

23from flipdare.util.time_util import FirestoreTime 

24from flipdare.wrapper import AppLogWrapper 

25 

26if TYPE_CHECKING: 

27 from flipdare.manager.db_manager import DbManager 

28 from flipdare.manager.backend_manager import BackendManager 

29 

30__all__ = ["LogReporter"] 

31 

32 

33class LogReporter(ServiceProvider): 

34 def __init__( 

35 self, 

36 db_manager: DbManager, 

37 backend_manager: BackendManager, 

38 ) -> None: 

39 super().__init__( 

40 db_manager=db_manager, 

41 backend_manager=backend_manager, 

42 ) 

43 

44 def payment_critical_issues(self) -> OutputResult: 

45 report = QueryReport[LogReportSchema]( 

46 job_type=AppJobType.REPORT_PAYMENT_CRITICAL_ISSUES, 

47 schema_cls=LogReportSchema, 

48 query_fn=lambda: self.log_db.get_recent_payment_critical_issues(), 

49 process_fn=self._payment_entry_processor, 

50 app_logger=self.app_logger, 

51 mailer=self.admin_mailer, 

52 ) 

53 

54 return report.create_and_send() 

55 

56 def _payment_entry_processor(self, entry: AppLogWrapper) -> LogReportSchema: 

57 flag_id = entry.doc_id 

58 error_code = entry.error_code.value if entry.error_code is not None else "N/A" 

59 obj_id = entry.obj_id if entry.obj_id is not None else "N/A" 

60 source = entry.source if entry.source is not None else "N/A" 

61 cause = entry.message 

62 extra_info = stringify_debug(entry.extra) if entry.extra is not None else "N/A" 

63 backend_obj_link = AppBackendLink.PLEDGE.link(flag_id) if flag_id else "N/A" 

64 

65 return LogReportSchema( 

66 { 

67 "doc_id": flag_id, 

68 "error_code": error_code, 

69 "obj_id": obj_id, 

70 "source": source, 

71 "cause": cause, 

72 "extra_info": extra_info, 

73 "link": backend_obj_link, 

74 "created_at": FirestoreTime.internal_str(entry.created_at_db), 

75 "updated_at": FirestoreTime.internal_str(entry.updated_at_db), 

76 }, 

77 )