Coverage for src/tests/test_public.py: 100%

1import json 

2from datetime import UTC, datetime 

3from math import sqrt 

4from unittest.mock import patch 

5 

6import pytest 

7from google.protobuf import empty_pb2 

8 

9from couchers.db import session_scope 

10from couchers.jobs.enqueue import queue_job 

11from couchers.models import ( 

12 Invoice, 

13 InvoiceType, 

14 ProfilePublicVisibility, 

15) 

16from couchers.servicers.public import _get_donation_stats 

17from tests.test_fixtures import ( # noqa 

18 db, 

19 generate_user, 

20 process_jobs, 

21 public_session, 

22 testconfig, 

23) 

24 

25 

26@pytest.fixture(autouse=True) 

27def _(testconfig): 

28 pass 

29 

30 

31def test_GetPublicMapLayer(db): 

32 user1, _ = generate_user() 

33 user2, _ = generate_user(username="user2", public_visibility=ProfilePublicVisibility.nothing) 

34 user3, _ = generate_user() 

35 user4, _ = generate_user(username="user4", public_visibility=ProfilePublicVisibility.limited) 

36 user5, _ = generate_user() 

37 

38 # these are hardcoded in test_fixtures 

39 test_user_coordinates = [-73.9740, 40.7108] 

40 

41 with session_scope() as session: 

42 queue_job(session, "update_randomized_locations", empty_pb2.Empty()) 

43 

44 process_jobs() 

45 

46 with public_session() as public: 

47 http_body = public.GetPublicUsers(empty_pb2.Empty()) 

48 assert http_body.content_type == "application/json" 

49 data = json.loads(http_body.data) 

50 # Sort to ensure a deterministic order 

51 data["features"].sort(key=lambda f: f["geometry"]["coordinates"][0]) 

52 assert data == { 

53 "type": "FeatureCollection", 

54 "features": [ 

55 { 

56 "type": "Feature", 

57 "geometry": {"type": "Point", "coordinates": [-74.042643848, 40.706241098]}, 

58 "properties": {"username": None}, 

59 }, 

60 { 

61 "type": "Feature", 

62 "geometry": {"type": "Point", "coordinates": [-73.974, 40.7108]}, 

63 "properties": {"username": "user4"}, 

64 }, 

65 { 

66 "type": "Feature", 

67 "geometry": {"type": "Point", "coordinates": [-73.955417734, 40.691831306]}, 

68 "properties": {"username": None}, 

69 }, 

70 { 

71 "type": "Feature", 

72 "geometry": {"type": "Point", "coordinates": [-73.928380198, 40.729706144]}, 

73 "properties": {"username": None}, 

74 }, 

75 ], 

76 } 

77 

78 for user in data["features"]: 

79 coords = user["geometry"]["coordinates"] 

80 if user["properties"]["username"]: 

81 assert coords == test_user_coordinates 

82 else: 

83 xdiff = coords[0] - test_user_coordinates[0] 

84 ydiff = coords[1] - test_user_coordinates[1] 

85 dist = sqrt(xdiff**2 + ydiff**2) 

86 assert dist > 0.02 and dist < 0.1 

87 

88 

89def test_GetDonationStats_empty(db): 

90 """Test GetDonationStats with no donations returns zero and goal""" 

91 _get_donation_stats.cache.clear() 

92 

93 with ( 

94 patch("couchers.servicers.public.DONATION_GOAL_USD", 2500), 

95 patch("couchers.servicers.public.DONATION_OFFSET_USD", 700), 

96 ): 

97 with public_session() as public: 

98 res = public.GetDonationStats(empty_pb2.Empty()) 

99 assert res.total_donated_ytd == 0 

100 assert res.goal == 2500 

101 

102 

103def test_GetDonationStats_with_donations(db): 

104 """Test GetDonationStats sums on_platform donations correctly""" 

105 _get_donation_stats.cache.clear() 

106 user, _ = generate_user() 

107 

108 with session_scope() as session: 

109 # Add some on_platform donations (should be counted) 

110 session.add( 

111 Invoice( 

112 user_id=user.id, 

113 amount=100, 

114 stripe_payment_intent_id="pi_test_1", 

115 stripe_receipt_url="https://example.com/receipt/1", 

116 invoice_type=InvoiceType.on_platform, 

117 ) 

118 ) 

119 session.add( 

120 Invoice( 

121 user_id=user.id, 

122 amount=250, 

123 stripe_payment_intent_id="pi_test_2", 

124 stripe_receipt_url="https://example.com/receipt/2", 

125 invoice_type=InvoiceType.on_platform, 

126 ) 

127 ) 

128 session.add( 

129 Invoice( 

130 user_id=user.id, 

131 amount=500, 

132 stripe_payment_intent_id="pi_test_3", 

133 stripe_receipt_url="https://example.com/receipt/3", 

134 invoice_type=InvoiceType.on_platform, 

135 ) 

136 ) 

137 

138 with ( 

139 patch("couchers.servicers.public.DONATION_GOAL_USD", 5000), 

140 patch("couchers.servicers.public.DONATION_OFFSET_USD", 0), 

141 ): 

142 with public_session() as public: 

143 res = public.GetDonationStats(empty_pb2.Empty()) 

144 assert res.total_donated_ytd == 850 

145 assert res.goal == 5000 

146 

147 

148def test_GetDonationStats_excludes_merch(db): 

149 """Test GetDonationStats excludes external_shop (merch) invoices""" 

150 _get_donation_stats.cache.clear() 

151 user, _ = generate_user() 

152 

153 with session_scope() as session: 

154 # Add on_platform donation (should be counted) 

155 session.add( 

156 Invoice( 

157 user_id=user.id, 

158 amount=200, 

159 stripe_payment_intent_id="pi_test_donation", 

160 stripe_receipt_url="https://example.com/receipt/donation", 

161 invoice_type=InvoiceType.on_platform, 

162 ) 

163 ) 

164 # Add external_shop/merch purchase (should NOT be counted) 

165 session.add( 

166 Invoice( 

167 user_id=user.id, 

168 amount=50, 

169 stripe_payment_intent_id="pi_test_merch", 

170 stripe_receipt_url="https://example.com/receipt/merch", 

171 invoice_type=InvoiceType.external_shop, 

172 ) 

173 ) 

174 

175 with ( 

176 patch("couchers.servicers.public.DONATION_GOAL_USD", 5000), 

177 patch("couchers.servicers.public.DONATION_OFFSET_USD", 0), 

178 ): 

179 with public_session() as public: 

180 res = public.GetDonationStats(empty_pb2.Empty()) 

181 # Should only count the on_platform donation, not the merch 

182 assert res.total_donated_ytd == 200 

183 assert res.goal == 5000 

184 

185 

186def test_GetDonationStats_excludes_previous_years(db): 

187 """Test GetDonationStats only counts current year donations""" 

188 _get_donation_stats.cache.clear() 

189 user, _ = generate_user() 

190 

191 with session_scope() as session: 

192 # Add donation from this year (should be counted) 

193 session.add( 

194 Invoice( 

195 user_id=user.id, 

196 amount=300, 

197 stripe_payment_intent_id="pi_test_this_year", 

198 stripe_receipt_url="https://example.com/receipt/this_year", 

199 invoice_type=InvoiceType.on_platform, 

200 ) 

201 ) 

202 # Add donation from last year (should NOT be counted) 

203 last_year = datetime(datetime.now(UTC).year - 1, 6, 15, tzinfo=UTC) 

204 invoice = Invoice( 

205 user_id=user.id, 

206 amount=1000, 

207 stripe_payment_intent_id="pi_test_last_year", 

208 stripe_receipt_url="https://example.com/receipt/last_year", 

209 invoice_type=InvoiceType.on_platform, 

210 ) 

211 session.add(invoice) 

212 session.flush() 

213 # Manually set the created date to last year 

214 invoice.created = last_year 

215 

216 with ( 

217 patch("couchers.servicers.public.DONATION_GOAL_USD", 5000), 

218 patch("couchers.servicers.public.DONATION_OFFSET_USD", 0), 

219 ): 

220 with public_session() as public: 

221 res = public.GetDonationStats(empty_pb2.Empty()) 

222 # Should only count this year's donation 

223 assert res.total_donated_ytd == 300 

224 assert res.goal == 5000