Coverage for functions \ flipdare \ task \ report \ issue_reporter.py: 52%

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.issue_report_schema import IssueReportSchema 

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

22from flipdare.util.time_util import FirestoreTime 

23from flipdare.wrapper import IssueWrapper 

24 

25if TYPE_CHECKING: 

26 from flipdare.manager.db_manager import DbManager 

27 from flipdare.manager.backend_manager import BackendManager 

28 

29 

30class IssueReporter(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[IssueReportSchema]( 

43 job_type=AppJobType.REPORT_ISSUE_WAITING_ADMIN, 

44 schema_cls=IssueReportSchema, 

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

46 process_fn=self._issue_entry_processor, 

47 app_logger=self.app_logger, 

48 mailer=self.admin_mailer, 

49 ) 

50 return report.create_and_send() 

51 

52 def _issue_entry_processor(self, issue: IssueWrapper) -> IssueReportSchema: 

53 issue_id = issue.doc_id 

54 issue_type = issue.issue_type.value 

55 progress = issue.progress.value 

56 from_uid = issue.from_uid 

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

58 created_at = FirestoreTime.internal_str(issue.created_at_db) 

59 updated_at = FirestoreTime.internal_str(issue.updated_at_db) 

60 description = issue.description 

61 

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

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

64 else: 

65 backend_obj_link = "N/A" 

66 

67 return IssueReportSchema( 

68 { 

69 "from_uid": from_uid, 

70 "issue_type": issue_type, 

71 "progress": progress, 

72 "description": description, 

73 "backend_link": backend_link, 

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

75 "backend_obj_link": backend_obj_link, 

76 "created_at": created_at, 

77 "updated_at": updated_at, 

78 }, 

79 )