Coverage for functions \ flipdare \ voting \ ballot_manager.py: 78%

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 flipdare.app_log import LOG 

15from flipdare.constants import IS_DEBUG 

16from flipdare.generated import AppErrorCode, AppJobType, DareStatus, PledgeStatus 

17from flipdare.generated.shared.firestore_collections import FirestoreCollections 

18from flipdare.service._service_provider import ServiceProvider 

19from flipdare.voting.ballot import Ballot, BallotOutcome, VoteTally 

20from flipdare.wrapper import DareWrapper 

21 

22__all__ = ["BallotManager", "get_ballot_manager"] 

23 

24 

25def get_ballot_manager() -> "BallotManager": 

26 return BallotManager.instance() 

27 

28 

29class BallotManager(ServiceProvider): 

30 

31 def __init__(self) -> None: 

32 super().__init__() 

33 

34 def count_votes(self, dare: DareWrapper) -> BallotOutcome | None: 

35 dare_id = dare.doc_id 

36 if not dare_id: 

37 msg = "Dare has no document ID, cannot evaluate ballot" 

38 self.app_logger.unexpected_code_path( 

39 job_type=AppJobType.CR_DARE_VOTE, 

40 collection=FirestoreCollections.DARE, 

41 message=msg, 

42 data={"dare_data": dare.to_dict()}, 

43 ) 

44 return None 

45 if dare.status != DareStatus.VOTING: 

46 if IS_DEBUG: 

47 LOG().debug(f"Dare {dare.doc_id} is not in VOTING status, cannot vote") 

48 return None 

49 

50 tally = self._get_vote_tally(dare) 

51 if tally is None: 

52 LOG().error(f"Could not retrieve vote tally for dare {dare_id}, aborting ballot count") 

53 return None 

54 if tally.empty: 

55 if IS_DEBUG: 

56 msg = f"Vote tally for dare {dare_id} is empty, cannot proceed with ballot count" 

57 LOG().debug(msg) 

58 return None 

59 

60 ballot = Ballot(tally) 

61 within_voting_period = dare.within_voting_period 

62 outcome = ballot.count_votes(within_voting_period) 

63 

64 if IS_DEBUG: 

65 msg = f"Ballot count for dare {dare_id}: {outcome!s} with stats: {ballot.stats_str()}" 

66 LOG().debug(msg) 

67 

68 result = outcome.result 

69 

70 # count votes takes into acconut the voting period, 

71 if within_voting_period and ( 

72 result.not_enough_votes 

73 or result.is_tie 

74 or result.is_rejected 

75 or result.is_auto_rejected 

76 ): 

77 # we are within the voting period, and we have either inconclusive or reject, 

78 # so we wait until the voting period expires to make the final decision 

79 if IS_DEBUG: 

80 msg = ( 

81 f"Ballot result for dare {dare_id} is {result} still within the voting period," 

82 " waiting until expiration to finalize" 

83 ) 

84 LOG().debug(msg) 

85 return outcome 

86 

87 # if we get here we are either outside the voting period 

88 # or have a conclusive result within the voting period (i.e. accepted or auto-accepted), 

89 # so we can finalize the result and update the dare status accordingly 

90 algorithm = outcome.algorithm 

91 if algorithm is None: 

92 # NOTE: we must have an algorithm for compliance. 

93 msg = ( 

94 f"Ballot result for dare {dare_id} has no algorithm decision, " 

95 f"cannot update dare status; {result!s}" 

96 ) 

97 self.app_logger.unexpected_code_path( 

98 job_type=AppJobType.CR_DARE_VOTE, 

99 collection=FirestoreCollections.DARE, 

100 message=msg, 

101 data={"dare_id": dare_id, "ballot_result": str(result)}, 

102 ) 

103 return None 

104 

105 # update the dare/db. 

106 dare.set_vote_result(outcome) 

107 updates = dare.get_updates() 

108 if not updates: 

109 cause = f"No updates to apply for dare {dare_id} with decision {result}: {ballot.stats_str()}" 

110 self.app_logger.unexpected_code_path( 

111 job_type=AppJobType.CR_DARE_VOTE, 

112 collection=FirestoreCollections.DARE, 

113 message=cause, 

114 data={ 

115 "dare_id": dare_id, 

116 "decision": str(result), 

117 "ballot_stats": ballot.stats_str(), 

118 }, 

119 ) 

120 return outcome 

121 

122 dare_db = self.dare_db 

123 dare_db.update(dare_id, updates) 

124 if IS_DEBUG: 

125 msg = f"Dare {dare_id} has status {dare.status} after voting round with outcome: {outcome!s}" 

126 LOG().debug(msg) 

127 

128 return outcome 

129 

130 def _get_vote_tally(self, dare: DareWrapper) -> VoteTally | None: 

131 dare_id = dare.doc_id 

132 try: 

133 pledge_db = self.pledge_db 

134 pledges = pledge_db.get_pledges_for_dare(dare_id) 

135 if len(pledges) == 0: 

136 LOG().error(f"No pledges found for dare {dare_id}, cannot tally votes") 

137 return None 

138 

139 approved = sum(1 for p in pledges if p.status == PledgeStatus.VOTE_APPROVED) 

140 rejected = sum(1 for p in pledges if p.status == PledgeStatus.VOTE_REJECTED) 

141 undecided = sum(1 for p in pledges if p.status == PledgeStatus.UNDECIDED) 

142 

143 tally = VoteTally(accepted=approved, rejected=rejected, undecided=undecided) 

144 LOG().debug(f"Vote tally for dare {dare_id}: {tally}") 

145 return tally 

146 

147 except Exception as e: 

148 self.app_logger.system_error( 

149 job_type=AppJobType.CR_DARE_VOTE, 

150 error_code=AppErrorCode.VOTING, 

151 collection=FirestoreCollections.DARE, 

152 message=f"Error tallying votes for dare {dare_id}: {e!s}", 

153 data={"dare_id": dare_id}, 

154 ) 

155 return None