Coverage for functions \ flipdare \ task \ report \ restriction_reporter.py: 86%

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.constants import NO_DOC_ID 

19from flipdare.core.app_backend_link import AppBackendLink 

20from flipdare.result.output_result import OutputResult 

21from flipdare.generated.schema.report.restriction_report_schema import RestrictionReportSchema 

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

23from flipdare.util.time_util import FirestoreTime 

24from flipdare.wrapper import RestrictionWrapper 

25 

26if TYPE_CHECKING: 

27 from flipdare.manager.db_manager import DbManager 

28 from flipdare.manager.backend_manager import BackendManager 

29 

30__all__ = ["RestrictionReporter"] 

31 

32 

33class RestrictionReporter(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 auto_not_permanent(self) -> OutputResult: 

45 job_type = AppJobType.REPORT_RESTRICT_AUTO_NOT_PERMANENT 

46 report = QueryReport[RestrictionReportSchema]( 

47 job_type=job_type, 

48 schema_cls=RestrictionReportSchema, 

49 query_fn=lambda: self.restriction_db.get_auto_not_permanent(), 

50 process_fn=self._restriction_entry_processor, 

51 app_logger=self.app_logger, 

52 mailer=self.admin_mailer, 

53 ) 

54 

55 return report.create_and_send() 

56 

57 def auto_permanent(self) -> OutputResult: 

58 report = QueryReport[RestrictionReportSchema]( 

59 job_type=AppJobType.REPORT_RESTRICT_AUTO_PERMANENT, 

60 schema_cls=RestrictionReportSchema, 

61 query_fn=lambda: self.restriction_db.get_auto_permanent(), 

62 process_fn=self._restriction_entry_processor, 

63 app_logger=self.app_logger, 

64 mailer=self.admin_mailer, 

65 ) 

66 

67 return report.create_and_send() 

68 

69 def inactive(self) -> OutputResult: 

70 report = QueryReport[RestrictionReportSchema]( 

71 job_type=AppJobType.REPORT_RESTRICT_INACTIVE, 

72 schema_cls=RestrictionReportSchema, 

73 query_fn=lambda: self.restriction_db.get_inactive_actions(), 

74 process_fn=self._restriction_entry_processor, 

75 app_logger=self.app_logger, 

76 mailer=self.admin_mailer, 

77 ) 

78 

79 return report.create_and_send() 

80 

81 def _restriction_entry_processor( 

82 self, 

83 restriction: RestrictionWrapper, 

84 ) -> RestrictionReportSchema: 

85 stopwatch_start = "N/A" 

86 stopwatch_end = "N/A" 

87 stopwatch = restriction.stopwatch 

88 if stopwatch is not None: 

89 stopwatch_start = stopwatch.started_at_formatted() 

90 stopwatch_end = stopwatch.expires_at_formatted() 

91 

92 doc_id = restriction.doc_id 

93 backend_link = AppBackendLink.RESTRICTION.link(doc_id) if doc_id != NO_DOC_ID else "N/A" 

94 created_at = FirestoreTime.internal_str(restriction.created_at_db) 

95 updated_at = FirestoreTime.internal_str(restriction.updated_at_db) 

96 

97 return RestrictionReportSchema( 

98 { 

99 "restriction_id": doc_id, 

100 "user_id": restriction.uid, 

101 "in_danger": str(restriction.in_danger), 

102 "flag_type": restriction.flag_type.value, 

103 "status": restriction.status.value, 

104 "reason": restriction.reason, 

105 "action": restriction.action.value, 

106 "duration": restriction.duration.value, 

107 "category": restriction.category.value, 

108 "stopwatch_start": stopwatch_start, 

109 "stopwatch_end": stopwatch_end, 

110 "restriction_link": backend_link, 

111 "created_at": created_at, 

112 "updated_at": updated_at, 

113 }, 

114 )