Coverage for functions \ flipdare \ task \ report \ payment_issue_reporter.py: 50%

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.payment_issue_report_schema import PaymentIssueReportSchema 

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

22from flipdare.util.time_util import FirestoreTime 

23from flipdare.wrapper import PaymentIssueWrapper 

24 

25if TYPE_CHECKING: 

26 from flipdare.manager.db_manager import DbManager 

27 from flipdare.manager.backend_manager import BackendManager 

28 

29 

30class PaymentIssueReporter(ServiceProvider): 

31 def __init__( 

32 self, 

33 db_manager: DbManager, 

34 backend_manager: BackendManager, 

35 ) -> None: 

36 super().__init__( 

37 db_manager=db_manager, 

38 backend_manager=backend_manager, 

39 ) 

40 

41 def issue_waiting_admin(self) -> OutputResult: 

42 report = QueryReport[PaymentIssueReportSchema]( 

43 job_type=AppJobType.REPORT_PAYMENT_ISSUE_WAITING_ADMIN, 

44 schema_cls=PaymentIssueReportSchema, 

45 query_fn=lambda: self.payment_issue_db.get_recent_waiting_admin(), 

46 process_fn=self._payment_issue_entry_processor, 

47 app_logger=self.app_logger, 

48 mailer=self.admin_mailer, 

49 ) 

50 return report.create_and_send() 

51 

52 def _payment_issue_entry_processor( 

53 self, 

54 issue: PaymentIssueWrapper, 

55 ) -> PaymentIssueReportSchema: 

56 issue_id = issue.doc_id 

57 issue_type = issue.issue_type.value if issue.issue_type is not None else "N/A" 

58 from_uid = issue.from_uid 

59 progress = issue.progress.value 

60 disputed_progress = issue.disputed_progress.value if issue.disputed_progress else "N/A" 

61 backend_link = AppBackendLink.FLAG.link(issue_id) if issue_id else "N/A" 

62 created_at = FirestoreTime.internal_str(issue.created_at_db) 

63 updated_at = FirestoreTime.internal_str(issue.updated_at_db) 

64 description = issue.description 

65 

66 if issue.obj_id is not None and issue.obj_type is not None: 

67 backend_obj_link = AppBackendLink.FLAG.link(issue.obj_id) 

68 else: 

69 backend_obj_link = "N/A" 

70 

71 return PaymentIssueReportSchema( 

72 { 

73 "from_uid": from_uid, 

74 "issue_type": issue_type, 

75 "progress": progress, 

76 "disputed_progress": disputed_progress, 

77 "description": description, 

78 "backend_link": backend_link, 

79 "obj_id": issue.obj_id if issue.obj_id is not None else "N/A", 

80 "backend_obj_link": backend_obj_link, 

81 "created_at": created_at, 

82 "updated_at": updated_at, 

83 }, 

84 )